dongwoo6kim commented on code in PR #100:
URL:
https://github.com/apache/flink-connector-kafka/pull/100#discussion_r1759155830
##########
flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableITCase.java:
##########
@@ -188,6 +190,66 @@ public void testKafkaSourceSink() throws Exception {
deleteTestTopic(topic);
}
+ @Test
+ public void testKafkaSourceEmptyResultOnDeletedOffsets() throws Exception {
+ // we always use a different topic name for each parameterized topic,
+ // in order to make sure the topic can be created.
+ final String topic = "bounded_" + format + "_" + UUID.randomUUID();
+ createTestTopic(topic, 1, 1);
+ // ---------- Produce an event time stream into Kafka
-------------------
+ String groupId = getStandardProps().getProperty("group.id");
+ String bootstraps = getBootstrapServers();
+
+ final String createTable =
+ String.format(
+ "CREATE TABLE kafka (\n"
+ + " `user_id` INT,\n"
+ + " `item_id` INT,\n"
+ + " `behavior` STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = '%s',\n"
+ + " 'topic' = '%s',\n"
+ + " 'properties.bootstrap.servers' = '%s',\n"
+ + " 'properties.group.id' = '%s',\n"
+ + " 'scan.startup.mode' =
'specific-offsets',\n"
+ + " 'scan.bounded.mode' =
'specific-offsets',\n"
+ + " 'scan.startup.specific-offsets' =
'partition:0,offset:1',\n"
+ + " 'scan.bounded.specific-offsets' =
'partition:0,offset:3',\n"
+ + " %s\n"
+ + ")\n",
+ KafkaDynamicTableFactory.IDENTIFIER,
+ topic,
+ bootstraps,
+ groupId,
+ formatOptions());
+ tEnv.executeSql(createTable);
+ List<Row> values =
+ Arrays.asList(
+ Row.of(1, 1102, "behavior 1"),
+ Row.of(2, 1103, "behavior 2"),
+ Row.of(3, 1104, "behavior 3"));
+ tEnv.fromValues(values).insertInto("kafka").execute().await();
+ // ---------- Delete events from Kafka -------------------
+ Map<Integer, Long> partitionOffsetsToDelete = new HashMap<>();
+ partitionOffsetsToDelete.put(0, 3L);
+ deleteRecords(topic, partitionOffsetsToDelete);
Review Comment:
Yes by deleting message in this step we can cause hanging.
This test code is for issue
[here](https://issues.apache.org/jira/browse/FLINK-35565)
--
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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
