[GitHub] [flink] wuchong commented on a change in pull request #13975: [FLINK-20041][connector kafka] Support watermark push down for kafka …

[GitBox <gi...@apache.org>]

wuchong commented on a change in pull request #13975:
URL: https://github.com/apache/flink/pull/13975#discussion_r519183645



##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicSource.java
##########
@@ -73,6 +75,9 @@
 	/** Metadata that is appended at the end of a physical source row. */
 	protected List<String> metadataKeys;
 
+	/** Watermark strategy that is used to generate per-partition watermark. */
+	protected WatermarkStrategy<RowData> watermarkStrategy;

Review comment:
       ```suggestion
   	protected @Nullable WatermarkStrategy<RowData> watermarkStrategy;
   ```
   
   Mark it nullable. 

##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -516,10 +518,103 @@ public void testKafkaSourceSinkWithKeyAndFullValue() throws Exception {
 		deleteTestTopic(topic);
 	}
 
+	@Test
+	public void testPerPartitionWatermarkKafka() throws Exception {
+		if (isLegacyConnector) {
+			return;
+		}
+		// we always use a different topic name for each parameterized topic,
+		// in order to make sure the topic can be created.
+		final String topic = "per_partition_watermark_topic_" + format;
+		createTestTopic(topic, 4, 1);
+
+		// ---------- Produce an event time stream into Kafka -------------------
+		String groupId = standardProps.getProperty("group.id");
+		String bootstraps = standardProps.getProperty("bootstrap.servers");
+
+		final String createTable = String.format(
+				"CREATE TABLE kafka (\n"
+						+ "  `partition_id` INT,\n"
+						+ "  `name` STRING,\n"
+						+ "  `timestamp` TIMESTAMP(3),\n"
+						+ "  WATERMARK FOR `timestamp` AS `timestamp`\n"
+						+ ") WITH (\n"
+						+ "  'connector' = 'kafka',\n"
+						+ "  'topic' = '%s',\n"
+						+ "  'properties.bootstrap.servers' = '%s',\n"
+						+ "  'properties.group.id' = '%s',\n"
+						+ "  'scan.startup.mode' = 'earliest-offset',\n"
+						+ "  'sink.partitioner' = 'org.apache.flink.streaming.connectors.kafka.table.KafkaTableITCase$TestPartitioner',\n"
+						+ "  'format' = '%s'\n"
+						+ ")",
+				topic,
+				bootstraps,
+				groupId,
+				format);

Review comment:
       Do not hard code the class path. 

##########
File path: flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
##########
@@ -291,6 +291,14 @@ private static RowKind parseRowKind(String rowKindShortString) {
 			"Optional map of 'metadata_key:data_type'. The order will be alphabetically. " +
 			"The metadata is part of the data when enabled.");
 
+	private static final ConfigOption<Integer> SINK_INDEX_OF_ROWTIME = ConfigOptions
+		.key("sink-index-of-rowtime")
+		.intType()
+		.defaultValue(-1)
+		.withDeprecatedKeys(
+			"Option index of the rowtime field. The default value -1 indicate that don't drop " +
+			"the late data.");

Review comment:
       If this option controls whether to drop late data, then the option name should be something like `sink.drop-late-event=true`.
   
   The rowtime field can be derived from WATERMARK statement (can also be declared on sink), instead of a redundant option. 
   
   Besides, I find you only support this option on append only sink, then you should throw exceptions if the sink is not append only during `createDynamicTableSink`. 

##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -516,10 +518,103 @@ public void testKafkaSourceSinkWithKeyAndFullValue() throws Exception {
 		deleteTestTopic(topic);
 	}
 
+	@Test
+	public void testPerPartitionWatermarkKafka() throws Exception {
+		if (isLegacyConnector) {
+			return;
+		}
+		// we always use a different topic name for each parameterized topic,
+		// in order to make sure the topic can be created.
+		final String topic = "per_partition_watermark_topic_" + format;
+		createTestTopic(topic, 4, 1);
+
+		// ---------- Produce an event time stream into Kafka -------------------
+		String groupId = standardProps.getProperty("group.id");
+		String bootstraps = standardProps.getProperty("bootstrap.servers");
+
+		final String createTable = String.format(
+				"CREATE TABLE kafka (\n"
+						+ "  `partition_id` INT,\n"
+						+ "  `name` STRING,\n"
+						+ "  `timestamp` TIMESTAMP(3),\n"
+						+ "  WATERMARK FOR `timestamp` AS `timestamp`\n"
+						+ ") WITH (\n"
+						+ "  'connector' = 'kafka',\n"
+						+ "  'topic' = '%s',\n"
+						+ "  'properties.bootstrap.servers' = '%s',\n"
+						+ "  'properties.group.id' = '%s',\n"
+						+ "  'scan.startup.mode' = 'earliest-offset',\n"
+						+ "  'sink.partitioner' = 'org.apache.flink.streaming.connectors.kafka.table.KafkaTableITCase$TestPartitioner',\n"
+						+ "  'format' = '%s'\n"
+						+ ")",
+				topic,
+				bootstraps,
+				groupId,
+				format);

Review comment:
       ```suggestion
   						+ "  'sink.partitioner' = '%s',\n"
   						+ "  'format' = '%s'\n"
   						+ ")",
   				topic,
   				bootstraps,
   				groupId,
   				format,
   				TestPartitioner.class.getCanonicalName());
   ```

##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java
##########
@@ -516,10 +518,103 @@ public void testKafkaSourceSinkWithKeyAndFullValue() throws Exception {
 		deleteTestTopic(topic);
 	}
 
+	@Test
+	public void testPerPartitionWatermarkKafka() throws Exception {
+		if (isLegacyConnector) {
+			return;
+		}
+		// we always use a different topic name for each parameterized topic,
+		// in order to make sure the topic can be created.
+		final String topic = "per_partition_watermark_topic_" + format;
+		createTestTopic(topic, 4, 1);
+
+		// ---------- Produce an event time stream into Kafka -------------------
+		String groupId = standardProps.getProperty("group.id");
+		String bootstraps = standardProps.getProperty("bootstrap.servers");
+
+		final String createTable = String.format(
+				"CREATE TABLE kafka (\n"
+						+ "  `partition_id` INT,\n"
+						+ "  `name` STRING,\n"
+						+ "  `timestamp` TIMESTAMP(3),\n"
+						+ "  WATERMARK FOR `timestamp` AS `timestamp`\n"
+						+ ") WITH (\n"
+						+ "  'connector' = 'kafka',\n"
+						+ "  'topic' = '%s',\n"
+						+ "  'properties.bootstrap.servers' = '%s',\n"
+						+ "  'properties.group.id' = '%s',\n"
+						+ "  'scan.startup.mode' = 'earliest-offset',\n"
+						+ "  'sink.partitioner' = 'org.apache.flink.streaming.connectors.kafka.table.KafkaTableITCase$TestPartitioner',\n"
+						+ "  'format' = '%s'\n"
+						+ ")",
+				topic,
+				bootstraps,
+				groupId,
+				format);
+
+		tEnv.executeSql(createTable);
+
+		String initialValues = "INSERT INTO kafka\n"
+				+ "VALUES\n"
+				+ " (0, 'partition-0-name-0', TIMESTAMP '2020-03-08 13:12:11.123'),\n"
+				+ " (0, 'partition-0-name-1', TIMESTAMP '2020-03-08 14:12:12.223'),\n"
+				+ " (0, 'partition-0-name-2', TIMESTAMP '2020-03-08 15:12:13.323'),\n"
+				+ " (1, 'partition-1-name-0', TIMESTAMP '2020-03-09 13:13:11.123'),\n"
+				+ " (1, 'partition-1-name-1', TIMESTAMP '2020-03-09 15:13:11.133'),\n"
+				+ " (1, 'partition-1-name-2', TIMESTAMP '2020-03-09 16:13:11.143'),\n"
+				+ " (2, 'partition-2-name-0', TIMESTAMP '2020-03-10 13:12:14.123'),\n"
+				+ " (3, 'partition-3-name-0', TIMESTAMP '2020-03-11 17:12:11.123')\n";
+		tEnv.executeSql(initialValues).await();
+
+		// ---------- Consume stream from Kafka -------------------
+		String createSink =
+				"CREATE TABLE MySink("
+						+ "  id INT,"
+						+ "  name STRING,"
+						+ "  ts TIMESTAMP(3)"
+						+ ") WITH ("
+						+ "  'connector' = 'values',"
+						+ "  'sink-index-of-rowtime' = '2'"
+						+ ")";
+		tEnv.executeSql(createSink);
+		tEnv.executeSql("INSERT INTO MySink SELECT * FROM kafka");
+		final List<String> expected = Arrays.asList(
+			"0,partition-0-name-0,2020-03-08T13:12:11.123",
+			"0,partition-0-name-1,2020-03-08T14:12:12.223",
+			"0,partition-0-name-2,2020-03-08T15:12:13.323",
+			"1,partition-1-name-0,2020-03-09T13:13:11.123",
+			"1,partition-1-name-1,2020-03-09T15:13:11.133",
+			"1,partition-1-name-2,2020-03-09T16:13:11.143",
+			"2,partition-2-name-0,2020-03-10T13:12:14.123",
+			"3,partition-3-name-0,2020-03-11T17:12:11.123"
+		);
+		KafkaTableTestUtils.waitingExpectedResults("MySink", expected, Duration.ofSeconds(5));
+
+		// ------------- cleanup -------------------
+
+		deleteTestTopic(topic);
+	}
+
+
+
 	// --------------------------------------------------------------------------------------------
 	// Utilities
 	// --------------------------------------------------------------------------------------------
 
+	/**
+	 * Extract the partition id from the row and set it on the record.
+	 */
+	public static class TestPartitioner extends FlinkKafkaPartitioner<RowData> {
+
+		private static final long serialVersionUID = 1L;
+		private static final int PARTITION_ID_FIELD_IN_SCHEMA = 0;
+
+		@Override
+		public int partition(RowData record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
+			return record.getInt(PARTITION_ID_FIELD_IN_SCHEMA) % partitions.length;

Review comment:
       ```suggestion
   			return partitions[record.getInt(PARTITION_ID_FIELD_IN_SCHEMA) % partitions.length];
   ```




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org