Add test using rest catalog

debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/IcebergChangeConsumerRestCatalogTest.java

@@ -0,0 +1,54 @@
+/*
+ *
+ *  * Copyright memiiso Authors.
+ *  *
+ *  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ */
+
+package io.debezium.server.iceberg;
+import com.google.common.collect.Lists;
+import io.debezium.server.iceberg.testresources.BaseTest;
+import io.debezium.server.iceberg.testresources.CatalogRest;
+import io.debezium.server.iceberg.testresources.S3Minio;
+import io.debezium.server.iceberg.testresources.SourcePostgresqlDB;
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+
+/**
+ * Integration test that verifies basic reading from PostgreSQL database and writing to iceberg destination.
+ *
+ * @author Ismail Simsek
+ */
+@QuarkusTest
+@QuarkusTestResource(value = CatalogRest.class, restrictToAnnotatedClass = true)
+@QuarkusTestResource(value = S3Minio.class, restrictToAnnotatedClass = true)
+@QuarkusTestResource(value = SourcePostgresqlDB.class, restrictToAnnotatedClass = true)
+public class IcebergChangeConsumerRestCatalogTest extends BaseTest {
+
+  @Test
+  public void testSimpleUpload() {
+    Awaitility.await().atMost(Duration.ofSeconds(120)).until(() -> {
+      try {
+        CloseableIterable<Record> result = getTableDataV2("testc.inventory.customers");
+        return Lists.newArrayList(result).size() >= 3;
+      } catch (Exception e) {
+        return false;
+      }
+    });
+
+    List<TableIdentifier> tables = consumer.icebergCatalog.listTables(Namespace.of(consumer.namespace));
+    Assertions.assertTrue(tables.contains(TableIdentifier.of(Namespace.of(consumer.namespace), "debezium_offset_storage_table")));
+    Assertions.assertTrue(tables.contains(TableIdentifier.of(Namespace.of(consumer.namespace), "debeziumcdc_testc_inventory_customers")));
+  }
+}

debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/TestConfigSource.java

@@ -22,7 +22,7 @@ public class TestConfigSource implements ConfigSource {
   public static final String S3_MINIO_SECRET_KEY = "12345678";
   public static final String ICEBERG_CATALOG_TABLE_NAMESPACE = "debeziumevents";
   public static final String ICEBERG_CATALOG_NAME = "iceberg";
-  public static final String ICEBERG_FILEIO = "org.apache.iceberg.aws.s3.S3FileIO";
+  public static final String ICEBERG_FILEIO = "org.apache.iceberg.io.ResolvingFileIO";
   public static final String ICEBERG_WAREHOUSE_S3A = "s3a://" + S3_BUCKET_NAME + "/iceberg_warehouse";
   protected Map<String, String> config = new HashMap<>();
 

debezium-server-iceberg-sink/src/test/java/io/debezium/server/iceberg/testresources/CatalogRest.java

@@ -0,0 +1,50 @@
+/*
+ *
+ *  * Copyright memiiso Authors.
+ *  *
+ *  * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ */
+
+package io.debezium.server.iceberg.testresources;
+
+import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class CatalogRest implements QuarkusTestResourceLifecycleManager {
+  public static final int REST_CATALOG_PORT = 8181;
+  public static final String REST_CATALOG_IMAGE = "apache/iceberg-rest-fixture";
+
+  public static final GenericContainer<?> container = new GenericContainer<>(DockerImageName.parse(REST_CATALOG_IMAGE))
+      .withExposedPorts(REST_CATALOG_PORT)
+      .waitingFor(Wait.forLogMessage(".*Started Server.*", 1));
+
+  public static String getHostUrl() {
+    return String.format("http://%s:%s", container.getHost(), container.getMappedPort(REST_CATALOG_PORT));
+  }
+
+  @Override
+  public Map<String, String> start() {
+    container.start();
+
+    Map<String, String> config = new ConcurrentHashMap<>();
+
+    config.put("debezium.sink.iceberg.type", "rest");
+    config.put("debezium.sink.iceberg.uri", CatalogRest.getHostUrl());
+
+    return config;
+  }
+
+  @Override
+  public void stop() {
+    if (container != null) {
+      container.stop();
+    }
+  }
+
+}