import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkTableIndirectDistributionExample {

    public static void main(String[] args) throws Exception {

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // Assume you've defined and registered the source table "etdr_mongo"

        // Define the sink table with partitioning keys implicitly influencing distribution
        // This is hypothetical and depends on the sink connector supporting such configuration
        tableEnv.executeSql(
            "CREATE TABLE etdr_kafka ("
            + "column1 STRING, "
            + "column2 INT, "
            + "pi STRING, "
            + "ts TIMESTAMP"
            + ") WITH ("
            + "'connector' = 'kafka', "
            + "'topic' = 'your-topic', "
            + "'properties.bootstrap.servers' = 'localhost:9092', "
            + "'key.format' = '...', " // Key format for partitioning if applicable
            + "'key.fields' = 'pi', " // Implicitly distributes data by 'pi'
            + "'format' = 'json'"
            + ")"
        );

        // Insert into the sink table
        tableEnv.executeSql(
            "INSERT INTO etdr_kafka SELECT * FROM etdr_mongo"
        );

        env.execute("Flink Table Indirect Distribution Example");
    }
}