Add Iceberg CDC source plugin (#6554)

Add Iceberg CDC source plugin (#6552)

* Merge UPDATE pairs into single INDEX for CDC events

When a CoW UPDATE produces a DELETE + INSERT pair with the same
document_id after carryover removal, emit only the INSERT as INDEX.
Since OpenSearch INDEX is an upsert, the DELETE is unnecessary.

This also eliminates a potential issue where multiple ProcessWorker
threads consuming from the buffer in parallel could reorder DELETE
and INDEX operations for the same document, causing data loss.

Signed-off-by: Sotaro Hikita <bering1814@gmail.com>

Author: lawofcycles

data-prepper-plugins/iceberg-source/build.gradle

@@ -0,0 +1,73 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ */
+
+sourceSets {
+    integrationTest {
+        java {
+            compileClasspath += main.output + test.output
+            runtimeClasspath += main.output + test.output
+            srcDir file('src/integrationTest/java')
+        }
+        resources.srcDir file('src/integrationTest/resources')
+    }
+}
+
+configurations {
+    integrationTestImplementation.extendsFrom testImplementation
+    integrationTestRuntime.extendsFrom testRuntime
+}
+
+dependencies {
+    implementation project(':data-prepper-api')
+    implementation project(':data-prepper-plugins:buffer-common')
+    implementation project(':data-prepper-plugins:aws-plugin-api')
+
+    implementation 'org.apache.iceberg:iceberg-core:1.10.1'
+    implementation 'org.apache.iceberg:iceberg-data:1.10.1'
+    implementation 'org.apache.iceberg:iceberg-parquet:1.10.1'
+    implementation 'org.apache.iceberg:iceberg-orc:1.10.1'
+    implementation 'org.apache.iceberg:iceberg-aws:1.10.1'
+
+    implementation libs.parquet.hadoop
+    implementation libs.avro.core
+    implementation libs.hadoop.common
+    implementation 'org.apache.orc:orc-core:1.9.5'
+
+    implementation 'software.amazon.awssdk:glue'
+    implementation 'software.amazon.awssdk:s3'
+    implementation 'software.amazon.awssdk:sts'
+
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+
+    testImplementation project(':data-prepper-test:test-common')
+
+    integrationTestImplementation project(':data-prepper-plugins:in-memory-source-coordination-store')
+    integrationTestImplementation project(':data-prepper-core')
+    integrationTestImplementation project(':data-prepper-event')
+    integrationTestImplementation 'org.awaitility:awaitility:4.2.2'
+}
+
+task integrationTest(type: Test) {
+    group = 'verification'
+    testClassesDirs = sourceSets.integrationTest.output.classesDirs
+
+    useJUnitPlatform()
+
+    classpath = sourceSets.integrationTest.runtimeClasspath
+    systemProperty 'tests.iceberg.rest.uri', System.getProperty('tests.iceberg.rest.uri', 'http://localhost:8181')
+    systemProperty 'tests.iceberg.s3.endpoint', System.getProperty('tests.iceberg.s3.endpoint', 'http://localhost:8333')
+    systemProperty 'tests.iceberg.s3.accessKey', System.getProperty('tests.iceberg.s3.accessKey', 'admin')
+    systemProperty 'tests.iceberg.s3.secretKey', System.getProperty('tests.iceberg.s3.secretKey', 'password')
+    systemProperty 'tests.iceberg.s3.region', System.getProperty('tests.iceberg.s3.region', 'us-east-1')
+
+    filter {
+        includeTestsMatching '*IT'
+    }
+}

data-prepper-plugins/iceberg-source/docker/docker-compose.yml

@@ -0,0 +1,59 @@
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+
+services:
+  seaweedfs:
+    image: chrislusf/seaweedfs:4.06
+    ports:
+      - "8333:8333"
+    command: server -s3 -s3.port=8333 -s3.config=/etc/seaweedfs/s3.json -master.port=9333 -volume.port=8080 -filer -filer.port=8888
+    configs:
+      - source: s3-config
+        target: /etc/seaweedfs/s3.json
+    healthcheck:
+      test: ["CMD-SHELL", "curl -sf http://localhost:8333 || exit 1"]
+      interval: 2s
+      timeout: 5s
+      retries: 15
+      start_period: 10s
+
+  create-bucket:
+    image: chrislusf/seaweedfs:4.06
+    depends_on:
+      seaweedfs:
+        condition: service_healthy
+    entrypoint: /bin/sh
+    command: >
+      -c "
+      until wget -q --spider http://seaweedfs:8333 2>/dev/null; do sleep 1; done;
+      /usr/bin/weed shell -master=seaweedfs:9333 -filer=seaweedfs:8888 <<EOF
+      s3.bucket.create -name warehouse
+      EOF
+      "
+
+  iceberg-rest:
+    image: apache/iceberg-rest-fixture
+    depends_on:
+      create-bucket:
+        condition: service_completed_successfully
+    ports:
+      - "8181:8181"
+    environment:
+      AWS_REGION: us-east-1
+      AWS_ACCESS_KEY_ID: admin
+      AWS_SECRET_ACCESS_KEY: password
+      CATALOG_WAREHOUSE: s3://warehouse/
+      CATALOG_IO__IMPL: org.apache.iceberg.aws.s3.S3FileIO
+      CATALOG_S3_ENDPOINT: http://seaweedfs:8333
+      CATALOG_S3_ACCESS__KEY__ID: admin
+      CATALOG_S3_SECRET__ACCESS__KEY: password
+      CATALOG_S3_PATH__STYLE__ACCESS: "true"
+
+configs:
+  s3-config:
+    content: |
+      { "identities": [ { "name": "anonymous", "credentials": [ { "accessKey": "admin", "secretKey": "password" } ], "actions": [ "Admin", "Read", "Write" ] } ] }