fix test

memiiso · Aug 3, 2021 · 980fc9b · 980fc9b
1 parent b842dfb
commit 980fc9b
Show file tree

Hide file tree

Showing 9 changed files with 39 additions and 12 deletions.
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
@@ -8,6 +8,7 @@
 
 package io.debezium.server.iceberg;
 
+import io.debezium.DebeziumException;
 import io.debezium.engine.ChangeEvent;
 import io.debezium.engine.DebeziumEngine;
 import io.debezium.engine.format.Json;
@@ -24,6 +25,9 @@
 import java.util.stream.Collectors;
 import javax.annotation.PostConstruct;
 import javax.enterprise.context.Dependent;
+import javax.enterprise.inject.Any;
+import javax.enterprise.inject.Instance;
+import javax.enterprise.inject.literal.NamedLiteral;
 import javax.inject.Inject;
 import javax.inject.Named;
 
@@ -87,7 +91,12 @@ public class IcebergChangeConsumer extends BaseChangeConsumer implements Debeziu
   @ConfigProperty(name = "debezium.sink.iceberg.upsert-source-ts-ms-column", defaultValue = "__source_ts_ms")
   String sourceTsMsColumn;
 
+  @ConfigProperty(name = "debezium.sink.batch.batch-size-wait", defaultValue = "NoBatchSizeWait")
+  String batchSizeWaitName;
+
   @Inject
+  @Any
+  Instance<InterfaceBatchSizeWait> batchSizeWaitInstances;
   InterfaceBatchSizeWait batchSizeWait;
   static ImmutableMap<String, Integer> cdcOperations = ImmutableMap.of("c", 1, "r", 2, "u", 3, "d", 4);
 
@@ -115,7 +124,16 @@ void connect() throws InterruptedException {
 
     valSerde.configure(Collections.emptyMap(), false);
     valDeserializer = valSerde.deserializer();
+
+    Instance<InterfaceBatchSizeWait> instance = batchSizeWaitInstances.select(NamedLiteral.of(batchSizeWaitName));
+    if (instance.isAmbiguous()) {
+      throw new DebeziumException("Multiple batch size wait class named '" + batchSizeWaitName + "' were found");
+    } else if (instance.isUnsatisfied()) {
+      throw new DebeziumException("No batch size wait class named '" + batchSizeWaitName + "' is available");
+    }
+    batchSizeWait = instance.get();
     batchSizeWait.initizalize();
+    LOGGER.info("Using {}", batchSizeWait.getClass().getName());
   }
 
   public String map(String destination) {

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
@@ -8,6 +8,7 @@
 
 package io.debezium.server.iceberg;
 
+import io.debezium.DebeziumException;
 import io.debezium.engine.ChangeEvent;
 import io.debezium.engine.DebeziumEngine;
 import io.debezium.engine.format.Json;
@@ -28,6 +29,9 @@
 import java.util.stream.Collectors;
 import javax.annotation.PostConstruct;
 import javax.enterprise.context.Dependent;
+import javax.enterprise.inject.Any;
+import javax.enterprise.inject.Instance;
+import javax.enterprise.inject.literal.NamedLiteral;
 import javax.inject.Inject;
 import javax.inject.Named;
 
@@ -91,7 +95,12 @@ public class IcebergEventsChangeConsumer extends BaseChangeConsumer implements D
   String namespace;
   @ConfigProperty(name = "debezium.sink.iceberg.catalog-name", defaultValue = "default")
   String catalogName;
+  @ConfigProperty(name = "debezium.sink.batch.batch-size-wait", defaultValue = "NoBatchSizeWait")
+  String batchSizeWaitName;
+
   @Inject
+  @Any
+  Instance<InterfaceBatchSizeWait> batchSizeWaitInstances;
   InterfaceBatchSizeWait batchSizeWait;
 
   private TableIdentifier tableIdentifier;
@@ -131,7 +140,16 @@ void connect() throws InterruptedException {
     }
     // load table
     eventTable = icebergCatalog.loadTable(tableIdentifier);
+
+    Instance<InterfaceBatchSizeWait> instance = batchSizeWaitInstances.select(NamedLiteral.of(batchSizeWaitName));
+    if (instance.isAmbiguous()) {
+      throw new DebeziumException("Multiple batch size wait class named '" + batchSizeWaitName + "' were found");
+    } else if (instance.isUnsatisfied()) {
+      throw new DebeziumException("No batch size wait class named '" + batchSizeWaitName + "' is available");
+    }
+    batchSizeWait = instance.get();
     batchSizeWait.initizalize();
+    LOGGER.info("Using {}", batchSizeWait.getClass().getName());
   }
 
   public GenericRecord getIcebergRecord(String destination, ChangeEvent<Object, Object> record, OffsetDateTime batchTime) {

diff --git a/...erg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/DynamicBatchSizeWait.java b/...erg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/DynamicBatchSizeWait.java
@@ -11,7 +11,6 @@
 import java.util.IntSummaryStatistics;
 import java.util.LinkedList;
 import javax.enterprise.context.Dependent;
-import javax.enterprise.inject.Alternative;
 
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.slf4j.Logger;
@@ -24,7 +23,6 @@
  * @author Ismail Simsek
  */
 @Dependent
-@Alternative
 public class DynamicBatchSizeWait implements InterfaceBatchSizeWait {
   protected static final Logger LOGGER = LoggerFactory.getLogger(DynamicBatchSizeWait.class);
 

diff --git a/...g-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/InterfaceBatchSizeWait.java b/...g-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/InterfaceBatchSizeWait.java
@@ -8,14 +8,11 @@
 
 package io.debezium.server.iceberg.batchsizewait;
 
-import javax.enterprise.context.Dependent;
-
 /**
  * Implementation of the consumer that delivers the messages into Amazon S3 destination.
  *
  * @author Ismail Simsek
  */
-@Dependent
 public interface InterfaceBatchSizeWait {
 
   default void initizalize() {

diff --git a/...iceberg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/MaxBatchSizeWait.java b/...iceberg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/MaxBatchSizeWait.java
@@ -13,7 +13,6 @@
 import io.debezium.server.iceberg.DebeziumMetrics;
 
 import javax.enterprise.context.Dependent;
-import javax.enterprise.inject.Alternative;
 import javax.inject.Inject;
 
 import org.eclipse.microprofile.config.inject.ConfigProperty;
@@ -26,7 +25,6 @@
  * @author Ismail Simsek
  */
 @Dependent
-@Alternative
 public class MaxBatchSizeWait implements InterfaceBatchSizeWait {
   protected static final Logger LOGGER = LoggerFactory.getLogger(MaxBatchSizeWait.class);
 

diff --git a/...-iceberg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/NoBatchSizeWait.java b/...-iceberg-sink/src/main/java/io/debezium/server/iceberg/batchsizewait/NoBatchSizeWait.java
@@ -9,15 +9,13 @@
 package io.debezium.server.iceberg.batchsizewait;
 
 import javax.enterprise.context.Dependent;
-import javax.enterprise.inject.Default;
 
 /**
  * Optimizes batch size around 85%-90% of max,batch.size using dynamically calculated sleep(ms)
  *
  * @author Ismail Simsek
  */
 @Dependent
-@Default
 public class NoBatchSizeWait implements InterfaceBatchSizeWait {
 
   public void waitMs(Integer numRecordsProcessed, Integer processingTimeMs) {

diff --git a/...sink/src/test/java/io/debezium/server/iceberg/batchsizewait/DynamicBatchSizeWaitTest.java b/...sink/src/test/java/io/debezium/server/iceberg/batchsizewait/DynamicBatchSizeWaitTest.java
@@ -22,7 +22,7 @@
 class DynamicBatchSizeWaitTest {
 
   @Inject
-  InterfaceBatchSizeWait waitBatchSize;
+  DynamicBatchSizeWait waitBatchSize;
 
   @ConfigProperty(name = "debezium.source.poll.interval.ms", defaultValue = "1000")
   Integer pollIntervalMs;

diff --git a/...erg-sink/src/test/java/io/debezium/server/iceberg/batchsizewait/MaxBatchSizeWaitTest.java b/...erg-sink/src/test/java/io/debezium/server/iceberg/batchsizewait/MaxBatchSizeWaitTest.java
@@ -25,7 +25,7 @@
 @TestProfile(MaxBatchSizeWaitTestProfile.class)
 class MaxBatchSizeWaitTest extends BaseSparkTest {
   @Inject
-  InterfaceBatchSizeWait waitBatchSize;
+  MaxBatchSizeWait waitBatchSize;
   @ConfigProperty(name = "debezium.source.poll.interval.ms", defaultValue = "1000")
   Integer pollIntervalMs;
   @ConfigProperty(name = "debezium.source.max.batch.size", defaultValue = "1000")

diff --git a/...k/src/test/java/io/debezium/server/iceberg/batchsizewait/MaxBatchSizeWaitTestProfile.java b/...k/src/test/java/io/debezium/server/iceberg/batchsizewait/MaxBatchSizeWaitTestProfile.java
@@ -19,7 +19,7 @@ public class MaxBatchSizeWaitTestProfile implements QuarkusTestProfile {
   public Map<String, String> getConfigOverrides() {
     Map<String, String> config = new HashMap<>();
     // wait
-    config.put("quarkus.arc.selected-alternatives", "MaxBatchSizeWait");
+    config.put("debezium.sink.batch.batch-size-wait", "MaxBatchSizeWait");
     config.put("debezium.sink.batch.metrics.snapshot-mbean", "debezium.postgres:type=connector-metrics,context=snapshot,server=testc");
     config.put("debezium.sink.batch.metrics.streaming-mbean", "debezium.postgres:type=connector-metrics,context=streaming,server=testc");
     config.put("debezium.source.connector.class", "io.debezium.connector.postgresql.PostgresConnector");