nsivabalan commented on code in PR #17758:
URL: https://github.com/apache/hudi/pull/17758#discussion_r2655763414
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java:
##########
@@ -460,8 +460,10 @@ private static String resetTarget(Config configuration,
String database, String
public void sync() {
for (TableExecutionContext context : tableExecutionContexts) {
try {
- new HoodieStreamer(context.getConfig(), jssc,
Option.ofNullable(context.getProperties())).sync();
+ HoodieStreamer streamer = new HoodieStreamer(context.getConfig(),
jssc, Option.ofNullable(context.getProperties()));
+ streamer.sync();
successTables.add(Helpers.getTableWithDatabase(context));
+ streamer.shutdownGracefully();
Review Comment:
minor. should we move this to finally block?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/BaseTestKafkaSource.java:
##########
@@ -71,16 +71,16 @@ public abstract class BaseTestKafkaSource extends
SparkClientFunctionalTestHarne
protected final Option<SourceProfileSupplier> sourceProfile =
Option.of(mock(SourceProfileSupplier.class));
protected SchemaProvider schemaProvider;
- protected KafkaTestUtils testUtils;
+ protected static KafkaTestUtils testUtils;
- @BeforeEach
- public void initClass() {
+ @BeforeAll
+ public static void setupKafka() {
testUtils = new KafkaTestUtils();
testUtils.setup();
}
- @AfterEach
- public void cleanupClass() {
+ @AfterAll
Review Comment:
should we keep `AfterEach` and add `testUtils.deleteTopics();` in that?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java:
##########
@@ -87,21 +86,17 @@ public class TestAvroKafkaSource extends
SparkClientFunctionalTestHarness {
protected SchemaProvider schemaProvider;
- protected KafkaTestUtils testUtils;
+ protected static KafkaTestUtils testUtils;
@BeforeAll
public static void initClass() {
dataGen = new HoodieTestDataGenerator(0xDEED);
- }
-
- @BeforeEach
- public void setup() {
testUtils = new KafkaTestUtils();
testUtils.setup();
}
- @AfterEach
- public void tearDown() {
+ @AfterAll
Review Comment:
should we keep `AfterEach` and add `testUtils.deleteTopics();` in that?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/debezium/TestAbstractDebeziumSource.java:
##########
@@ -62,28 +60,19 @@ public abstract class TestAbstractDebeziumSource extends
UtilitiesTestBase {
private final String testTopicName = "hoodie_test_" + UUID.randomUUID();
private final HoodieIngestionMetrics metrics =
mock(HoodieIngestionMetrics.class);
- private KafkaTestUtils testUtils;
+ private static KafkaTestUtils testUtils;
@BeforeAll
public static void initClass() throws Exception {
UtilitiesTestBase.initTestServices();
- }
-
- @BeforeEach
- public void setUpKafkaTestUtils() {
testUtils = new KafkaTestUtils();
testUtils.setup();
}
- @AfterEach
Review Comment:
should we keep `AfterEach` and add `testUtils.deleteTopics();` in that?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaOffsetGen.java:
##########
@@ -89,16 +89,16 @@ public class TestKafkaOffsetGen {
private final String testTopicName = "hoodie_test_" + UUID.randomUUID();
private HoodieIngestionMetrics metrics = mock(HoodieIngestionMetrics.class);
- private KafkaTestUtils testUtils;
+ private static KafkaTestUtils testUtils;
- @BeforeEach
- public void setup() throws Exception {
+ @BeforeAll
+ public static void setup() throws Exception {
testUtils = new KafkaTestUtils();
testUtils.setup();
}
- @AfterEach
- public void teardown() throws Exception {
+ @AfterAll
Review Comment:
same comment as above.
--
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]
