Merge pull request #723 from Aiven-Open/iceberg-demo

Update Iceberg demo

Author: AnatolyPopov

README.md

@@ -9,7 +9,7 @@ The project follows the API specifications according to the latest version of [K
 
 ## NEW! Iceberg mode 🧊🎉
 
-The Iceberg mode was added recently. It's currenly in the "alpha" stage, but could already be explored. Please have a look at the [whitepaper](iceberg_whitepaper.md).
+The Iceberg mode was added recently. It's currenly in the "alpha" stage, but could already be explored. Please have a look at the [whitepaper](iceberg_whitepaper.md) and [demo](demo/iceberg/README.md).
 
 ## References
 

demo/iceberg/.gitignore

@@ -0,0 +1 @@
+plugin

demo/iceberg/Makefile

@@ -1,10 +1,9 @@
 .PHONY: plugin
 plugin:
-	cd ../../ && ./gradlew :core:distTar
-	cd ../../ && ./gradlew :storage:core:distTar
-	cd ../../ && ./gradlew :storage:filesystem:distTar
 	rm -r plugin || true
 	mkdir -p plugin
-	tar -xf ../../core/build/distributions/core-0.0.1-SNAPSHOT.tgz -C plugin --strip-components=1
-	tar -xf ../../storage/core/build/distributions/core-0.0.1-SNAPSHOT.tgz -C plugin --strip-components=1
-	tar -xf ../../storage/filesystem/build/distributions/filesystem-0.0.1-SNAPSHOT.tgz -C plugin --strip-components=1
+	../../gradlew -p ../../ :core:distTar :storage:core:distTar :storage:filesystem:distTar :storage:s3:distTar
+	tar -xf ../../core/build/distributions/core-1.1.0-SNAPSHOT.tgz -C plugin --strip-components=1
+	tar -xf ../../storage/core/build/distributions/core-1.1.0-SNAPSHOT.tgz -C plugin --strip-components=1
+	tar -xf ../../storage/filesystem/build/distributions/filesystem-1.1.0-SNAPSHOT.tgz -C plugin --strip-components=1
+	tar -xf ../../storage/s3/build/distributions/s3-1.1.0-SNAPSHOT.tgz -C plugin --strip-components=1

demo/iceberg/clients/app/src/main/java/io/aiven/Clients.java

@@ -83,9 +83,9 @@ private static void createTopic() throws ExecutionException, InterruptedExceptio
     }
 
     private static void produce() {
-        try (final KafkaProducer<GenericData.Record, GenericData.Record> producer = new KafkaProducer<>(Map.of(
+        try (final KafkaProducer<Object, GenericData.Record> producer = new KafkaProducer<>(Map.of(
             "bootstrap.servers", BOOTSTRAP_SERVER,
-            "key.serializer", ByteArraySerializer.class.getCanonicalName(),
+            "key.serializer", KafkaAvroSerializer.class.getCanonicalName(),
             "value.serializer", KafkaAvroSerializer.class.getCanonicalName(),
             "schema.registry.url", SCHEMA_REGISTRY_URL
         ))) {
@@ -103,7 +103,7 @@ private static void produce() {
                     .set("birthdate", person.getBirthdate().toString())
                     .set("comments", person.getComments())
                     .build();
-                producer.send(new ProducerRecord<>(TOPIC_NAME, 0, Time.SYSTEM.milliseconds(), null, value));
+                producer.send(new ProducerRecord<>(TOPIC_NAME, 0, Time.SYSTEM.milliseconds(), new byte[1], value));
             }
             producer.flush();
         }

demo/iceberg/docker-compose.yml

@@ -39,9 +39,19 @@ services:
       KAFKA_RSM_CONFIG_STORAGE_ROOT: "/tmp"
       KAFKA_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS: "5000"
       KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: "5000"
-#      KAFKA_RSM_CONFIG_SCHEMA_REGISTRY_URL: "http://karapace:8081"
-#      KAFKA_RSM_CONFIG_CATALOG_URL: "http://rest:8181"
-#      KAFKA_RSM_CONFIG_S3_ENDPOINT: "http://minio:9000"
+      # Iceberg configuration.
+      KAFKA_RSM_CONFIG_STRUCTURE_PROVIDER_CLASS: "io.aiven.kafka.tieredstorage.iceberg.AvroSchemaRegistryStructureProvider"
+      KAFKA_RSM_CONFIG_STRUCTURE_PROVIDER_SERDE_SCHEMA_REGISTRY_URL: "http://karapace:8081"
+      KAFKA_RSM_CONFIG_SEGMENT_FORMAT: "iceberg"
+      KAFKA_RSM_CONFIG_ICEBERG_NAMESPACE: "default"
+      KAFKA_RSM_CONFIG_ICEBERG_CATALOG_CLASS: "org.apache.iceberg.rest.RESTCatalog"
+      KAFKA_RSM_CONFIG_ICEBERG_CATALOG_URI: "http://rest:8181"
+      "KAFKA_RSM_CONFIG_ICEBERG_CATALOG_IO-IMPL": "org.apache.iceberg.aws.s3.S3FileIO"
+      KAFKA_RSM_CONFIG_ICEBERG_CATALOG_WAREHOUSE: "s3://warehouse/"
+      KAFKA_RSM_CONFIG_ICEBERG_CATALOG_S3_ENDPOINT: "http://minio:9000"
+      "KAFKA_RSM_CONFIG_ICEBERG_CATALOG_S3_ACCESS-KEY-ID": "admin"
+      "KAFKA_RSM_CONFIG_ICEBERG_CATALOG_S3_SECRET-ACCESS-KEY": "password"
+      KAFKA_RSM_CONFIG_ICEBERG_CATALOG_CLIENT_REGION: "us-east-1"
 
   # Karapace schema registry
   karapace: