Move batchSizeWaitName config to correct class (#521)

memiiso · Feb 24, 2025 · 32b61af · 32b61af
1 parent 46d536d
commit 32b61af
Show file tree

Hide file tree

Showing 5 changed files with 10 additions and 6 deletions.
diff --git a/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/GlobalConfig.java b/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/GlobalConfig.java
@@ -1,5 +1,6 @@
 package io.debezium.server.iceberg;
 
+import io.debezium.server.iceberg.batchsizewait.BatchSizeWaitConfig;
 import io.quarkus.runtime.annotations.ConfigRoot;
 import io.smallrye.config.ConfigMapping;
 import io.smallrye.config.WithDefault;
@@ -17,6 +18,9 @@ public interface GlobalConfig {
   @WithParentName
   DebeziumConfig debezium();
 
+  @WithParentName
+  BatchSizeWaitConfig batch();
+
   @WithName("quarkus.log.level")
   @WithDefault("INFO")
   Logger.Level quarkusLogLevel();

diff --git a/...m-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergChangeConsumer.java b/...m-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergChangeConsumer.java
@@ -90,7 +90,7 @@ void connect() {
     config.iceberg().icebergConfigs().forEach(this.hadoopConf::set);
 
     icebergCatalog = CatalogUtil.buildIcebergCatalog(config.iceberg().catalogName(), config.iceberg().icebergConfigs(), hadoopConf);
-    batchSizeWait = IcebergUtil.selectInstance(batchSizeWaitInstances, config.iceberg().batchSizeWaitName());
+    batchSizeWait = IcebergUtil.selectInstance(batchSizeWaitInstances, config.batch().batchSizeWaitName());
     batchSizeWait.initizalize();
 
     // configure and set 

diff --git a/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergConfig.java b/debezium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergConfig.java
@@ -72,10 +72,6 @@ public interface IcebergConfig {
   @WithDefault("true")
   boolean createIdentifierFields();
 
-  @WithName("debezium.sink.batch.batch-size-wait")
-  @WithDefault("NoBatchSizeWait")
-  String batchSizeWaitName();
-
   @WithName("debezium.sink.iceberg." + DEFAULT_FILE_FORMAT)
   @WithDefault(DEFAULT_FILE_FORMAT_DEFAULT)
   String writeFormat();

diff --git a/...er-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergEventsChangeConsumer.java b/...er-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergEventsChangeConsumer.java
@@ -136,7 +136,7 @@ void connect() {
     // load table
     eventTable = icebergCatalog.loadTable(tableIdentifier);
 
-    batchSizeWait = IcebergUtil.selectInstance(batchSizeWaitInstances, config.iceberg().batchSizeWaitName());
+    batchSizeWait = IcebergUtil.selectInstance(batchSizeWaitInstances, config.batch().batchSizeWaitName());
     batchSizeWait.initizalize();
 
     // configure and set 

diff --git a/...berg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/BatchSizeWaitConfig.java b/...berg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/BatchSizeWaitConfig.java
@@ -25,4 +25,8 @@ public interface BatchSizeWaitConfig {
   @WithDefault("10000")
   int batchSizeWaitWaitIntervalMs();
 
+  @WithName("debezium.sink.batch.batch-size-wait")
+  @WithDefault("NoBatchSizeWait")
+  String batchSizeWaitName();
+
 }