StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

Orders orders = ...; // Kafka source for Orders
Product products = ...; // Kafka source for Product

orders.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Orders>() {
    @Override
    public long extractAscendingTimestamp(Orders element) {
        return element.orderTimestamp;
    }
});

products.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Product>() {
    @Override
    public long extractAscendingTimestamp(Product element) {
        return element.productTimestamp;
    }
});

TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
tEnv.getConfig().setIdleStateRetentionTime(Time.hours(1), Time.hours(1));

DataStream<Product> productStream = ...; // assume this is your product stream

// register the DataStream as a Table
tEnv.createTemporaryView("Products", productStream, $("id"), $("productName"), $("productTimestamp"));

// create a Table Function representing the latest product information
TableFunction<Product> latestProduct = new LatestProductTableFunction();

// register the Table Function
tEnv.registerFunction("latestProduct", latestProduct);

// convert Orders stream to Table
DataStream<Orders> ordersStream = ...; // Your Orders stream
tEnv.createTemporaryView("Orders", ordersStream, $("orderId"), $("productId"), $("orderTimestamp"));

// Perform the temporal join
Table result = tEnv.sqlQuery(
    "SELECT Orders.orderId, Orders.productId, Orders.orderTimestamp, latestProduct.productName, latestProduct.productTimestamp " +
    "FROM Orders AS Orders, LATEST_BY_OFFSET(Products) AS latestProduct " + 
    "WHERE Orders.productId = latestProduct.id AND ABS(TIMESTAMPDIFF(SECOND, Orders.orderTimestamp, latestProduct.productTimestamp)) <= 3600"
);

// convert the result back to a DataStream if necessary
DataStream<ResultType> resultStream = tEnv.toAppendStream(result, ResultType.class);

(ABS(TIMESTAMPDIFF(SECOND, Orders.orderTimestamp, latestProduct.productTimestamp)) <= 3600).

CREATE TABLE Orders (
    orderId BIGINT,
    productId BIGINT,
    orderTimestamp TIMESTAMP(3),
    WATERMARK FOR orderTimestamp AS orderTimestamp - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka', 
    -- other Kafka connector properties
);

CREATE TABLE Products (
    id BIGINT,
    productName STRING,
    productTimestamp TIMESTAMP(3),
    WATERMARK FOR productTimestamp AS productTimestamp - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka', 
    -- other Kafka connector properties
);

WITH 
    JoinedData AS (
        SELECT 
            Orders.orderId, 
            Orders.productId, 
            Orders.orderTimestamp, 
            Products.productName, 
            Products.productTimestamp
        FROM 
            Orders
        LEFT JOIN 
            Products FOR SYSTEM_TIME AS OF Orders.orderTimestamp
        ON 
            Orders.productId = Products.id
    )
SELECT *
FROM JoinedData
WHERE ABS(TIMESTAMPDIFF(SECOND, orderTimestamp, productTimestamp)) <= 3600;