HenryCaiHaiying commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r4279358911
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/TestCoordinator.java:
##########
@@ -344,47 +347,52 @@ private UUID coordinatorTest(
SinkTaskContext context = mock(SinkTaskContext.class);
Coordinator coordinator =
new Coordinator(catalog, config, ImmutableList.of(), clientFactory,
context);
- coordinator.start();
-
- // init consumer after subscribe()
- initConsumer();
-
- coordinator.process();
-
- assertThat(producer.transactionCommitted()).isTrue();
- assertThat(producer.history()).hasSize(1);
-
- byte[] bytes = producer.history().get(0).value();
- Event commitRequest = AvroUtil.decode(bytes);
- assertThat(commitRequest.type()).isEqualTo(PayloadType.START_COMMIT);
-
- UUID commitId = ((StartCommit) commitRequest.payload()).commitId();
-
- Event commitResponse =
- new Event(
- config.connectGroupId(),
- new DataWritten(
- StructType.of(),
- commitId,
- TableReference.of("catalog", TableIdentifier.of("db", "tbl"),
null),
- dataFiles,
- deleteFiles));
- bytes = AvroUtil.encode(commitResponse);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
bytes));
-
- Event commitReady =
- new Event(
- config.connectGroupId(),
- new DataComplete(
- commitId, ImmutableList.of(new TopicPartitionOffset("topic",
1, 1L, ts))));
- bytes = AvroUtil.encode(commitReady);
- consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
bytes));
-
- when(config.commitIntervalMs()).thenReturn(0);
-
- coordinator.process();
-
- return commitId;
+ try {
+ coordinator.start();
+
+ // init consumer after subscribe()
+ initConsumer();
+
+ coordinator.process();
+
+ assertThat(producer.transactionCommitted()).isTrue();
+ assertThat(producer.history()).hasSize(1);
+
+ byte[] bytes = producer.history().get(0).value();
+ Event commitRequest = AvroUtil.decode(bytes);
+ assertThat(commitRequest.type()).isEqualTo(PayloadType.START_COMMIT);
+
+ UUID commitId = ((StartCommit) commitRequest.payload()).commitId();
+
+ Event commitResponse =
+ new Event(
+ config.connectGroupId(),
+ new DataWritten(
+ StructType.of(),
+ commitId,
+ TableReference.of("catalog", TableIdentifier.of("db",
"tbl"), null),
+ dataFiles,
+ deleteFiles));
+ bytes = AvroUtil.encode(commitResponse);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 1, "key",
bytes));
+
+ Event commitReady =
+ new Event(
+ config.connectGroupId(),
+ new DataComplete(
+ commitId, ImmutableList.of(new TopicPartitionOffset("topic",
1, 1L, ts))));
+ bytes = AvroUtil.encode(commitReady);
+ consumer.addRecord(new ConsumerRecord<>(CTL_TOPIC_NAME, 0, 2, "key",
bytes));
+
+ when(config.commitIntervalMs()).thenReturn(0);
+
+ coordinator.process();
+
+ return commitId;
+ } finally {
+ coordinator.terminate();
Review Comment:
Good catch. I will add try/catch for all other tests (looks like there are
6 tests creating Coordinator objects), and do another try/catch for terminate()
as well.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
