Fix __source_ts_ms values with 0 conversion and enable mongodb test (#…

…292) * Update IcebergChangeConsumerMangodbTest.java * Update SourceMangoDB.java * Update IcebergChangeEvent.java * Update SourceMangoDB.java
memiiso · Mar 13, 2024 · 1581e59 · 1581e59
1 parent dee8d05
commit 1581e59
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 2 deletions.
diff --git a/...zium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergChangeEvent.java b/...zium-server-iceberg-sink/src/main/java/io/debezium/server/iceberg/IcebergChangeEvent.java
@@ -149,7 +149,7 @@ private static Object jsonValToIcebergVal(Types.NestedField field, JsonNode node
         val = node.isValueNode() ? UUID.fromString(node.asText(null)) : UUID.fromString(node.toString());
         break;
       case TIMESTAMP:
-        if (node.isLong() && TS_MS_FIELDS.contains(field.name())) {
+        if ((node.isLong() || node.isNumber()) && TS_MS_FIELDS.contains(field.name())) {
           val = OffsetDateTime.ofInstant(Instant.ofEpochMilli(node.longValue()), ZoneOffset.UTC);
         } else if (node.isTextual()) {
           val = OffsetDateTime.parse(node.asText());

diff --git a/...eberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerMangodbTest.java b/...eberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerMangodbTest.java
@@ -31,7 +31,6 @@
  * @author Ismail Simsek
  */
 @QuarkusTest
-@Disabled // @TODO fix
 @QuarkusTestResource(value = S3Minio.class, restrictToAnnotatedClass = true)
 @QuarkusTestResource(value = SourceMangoDB.class, restrictToAnnotatedClass = true)
 @TestProfile(IcebergChangeConsumerMangodbTest.TestProfile.class)

diff --git a/...er-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/SourceMangoDB.java b/...er-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/SourceMangoDB.java
@@ -11,6 +11,7 @@
 import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -31,6 +32,7 @@ public class SourceMangoDB implements QuarkusTestResourceLifecycleManager {
 
   @Override
   public Map<String, String> start() {
+    container.setPortBindings(List.of(MONGODB_PORT+":"+MONGODB_PORT));
     container.withExposedPorts(MONGODB_PORT).start();
 
     Map<String, String> params = new ConcurrentHashMap<>();