[FLINK-29324] Add unit test case

apache · Sep 19, 2022 · dc4d256 · dc4d256
1 parent dba969a
commit dc4d256
Showing 1 changed file with 8 additions and 0 deletions.
diff --git a/...src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java b/...src/test/java/org/apache/flink/streaming/connectors/kinesis/FlinkKinesisConsumerTest.java
@@ -1184,6 +1184,14 @@ public void emitWatermark(Watermark mark) {
         testHarness.close();
     }
 
+    @Test
+    public void testCloseConnectorBeforeSubtaskStart() throws Exception {
+        Properties config = TestUtils.getStandardProperties();
+        FlinkKinesisConsumer<String> consumer =
+                new FlinkKinesisConsumer<>("fakeStream", new SimpleStringSchema(), config);
+        consumer.close();
+    }
+
     private void awaitRecordCount(ConcurrentLinkedQueue<? extends Object> queue, int count)
             throws Exception {
         Deadline deadline = Deadline.fromNow(Duration.ofSeconds(10));