You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2025/11/05 18:36:34 UTC
(iggy) branch master updated: feat(connectors/flink): add flink processor (#2300)
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 8ee56a971 feat(connectors/flink): add flink processor (#2300)
8ee56a971 is described below
commit 8ee56a971c913a9d9fdedfbb40f975cdf4b964cf
Author: Chiradip Mandal <ch...@chiradip.com>
AuthorDate: Wed Nov 5 10:36:30 2025 -0800
feat(connectors/flink): add flink processor (#2300)
---
.../iggy-connector-flink/docker-compose.yml | 273 ++++++++++++++
.../docker/Dockerfile.flink2_10-java21 | 11 +
.../iggy-connector-library/build.gradle.kts | 7 +-
.../iggy/connector/flink/sink/IggyCommittable.java | 129 +++++++
.../apache/iggy/connector/flink/sink/IggySink.java | 223 +++++++++++
.../iggy/connector/flink/sink/IggySinkBuilder.java | 220 +++++++++++
.../iggy/connector/flink/sink/IggySinkWriter.java | 303 +++++++++++++++
.../flink/source/IggyPartitionSplitReader.java | 257 +++++++++++++
.../iggy/connector/flink/source/IggySource.java | 271 ++++++++++++++
.../connector/flink/source/IggySourceBuilder.java | 191 ++++++++++
.../flink/source/IggySourceEnumeratorState.java | 91 +++++
.../IggySourceEnumeratorStateSerializer.java | 100 +++++
.../connector/flink/source/IggySourceReader.java | 192 ++++++++++
.../connector/flink/source/IggySourceSplit.java | 163 ++++++++
.../flink/source/IggySourceSplitEnumerator.java | 279 ++++++++++++++
.../flink/source/IggySourceSplitSerializer.java | 101 +++++
.../serialization/JsonDeserializationSchema.java | 133 +++++++
.../serialization/JsonSerializationSchema.java | 169 +++++++++
.../serialization/StringDeserializationSchema.java | 87 +++++
.../iggy-flink-examples/build.gradle.kts | 108 ++++++
.../data-population-explanation.md | 279 ++++++++++++++
.../deploy-and-run-word-count-job.md | 258 +++++++++++++
.../iggy/flink/example/MultiStreamJoinJob.java | 242 ++++++++++++
.../iggy/flink/example/SendTestMessages.java | 73 ++++
.../iggy/flink/example/StreamTransformJob.java | 222 +++++++++++
.../apache/iggy/flink/example/WordCountJob.java | 166 +++++++++
.../org/apache/iggy/flink/example/model/Alert.java | 107 ++++++
.../iggy/flink/example/model/EnrichedActivity.java | 137 +++++++
.../iggy/flink/example/model/SensorReading.java | 98 +++++
.../iggy/flink/example/model/UserActivity.java | 98 +++++
.../iggy/flink/example/model/UserProfile.java | 97 +++++
.../apache/iggy/flink/example/model/WordCount.java | 75 ++++
.../flink/example/AsyncTcpMessagePollTest.java | 414 +++++++++++++++++++++
.../flink/example/AsyncTcpMessageSendTest.java | 308 +++++++++++++++
.../iggy/flink/example/SendTextDataTest.java | 139 +++++++
foreign/java/java-sdk/build.gradle.kts | 4 +
36 files changed, 6024 insertions(+), 1 deletion(-)
diff --git a/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml b/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml
new file mode 100644
index 000000000..da7011c30
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/docker-compose.yml
@@ -0,0 +1,273 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+services:
+ # Iggy Message Broker
+ iggy:
+ image: apache/iggy:edge
+ container_name: iggy-server
+ ports:
+ - "8080:8080" # QUIC transport
+ - "8090:8090" # TCP transport
+ - "3000:3000" # HTTP API / Web UI
+ environment:
+ - IGGY_TCP_ADDRESS=0.0.0.0:8090
+ - IGGY_HTTP_ADDRESS=0.0.0.0:3000
+ - IGGY_QUIC_ADDRESS=0.0.0.0:8080
+ - IGGY_SYSTEM_LOGGING_LEVEL=info
+ - IGGY_ROOT_USERNAME=iggy
+ - IGGY_ROOT_PASSWORD=iggy
+ volumes:
+ - iggy-data:/local_data
+ networks:
+ - flink-iggy-network
+
+ # Flink JobManager
+ jobmanager:
+ image: flink:2.1.0-scala_2.12-java21
+ container_name: flink-jobmanager
+ ports:
+ - "8081:8081" # Web UI
+ - "6123:6123" # RPC
+ user: root # Start as root to fix permissions
+ entrypoint: /bin/bash
+ command:
+ - -c
+ - |
+ # Create checkpoint/savepoint directories with correct permissions
+ mkdir -p /tmp/flink-checkpoints /tmp/flink-savepoints
+ chown -R flink:flink /tmp/flink-checkpoints /tmp/flink-savepoints
+ chmod -R 755 /tmp/flink-checkpoints /tmp/flink-savepoints
+ # Switch to flink user and run jobmanager
+ exec gosu flink /docker-entrypoint.sh jobmanager
+ environment:
+ - IGGY_SERVER=iggy:8090
+ - |
+ FLINK_PROPERTIES=
+ jobmanager.rpc.address: jobmanager
+ jobmanager.memory.process.size: 1600m
+ state.backend.type: hashmap
+ state.checkpoints.dir: file:///tmp/flink-checkpoints
+ state.savepoints.dir: file:///tmp/flink-savepoints
+ execution.checkpointing.interval: 60000
+ execution.checkpointing.mode: EXACTLY_ONCE
+ execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
+ parallelism.default: 2
+ taskmanager.numberOfTaskSlots: 4
+ volumes:
+ - flink-checkpoints:/tmp/flink-checkpoints
+ - flink-savepoints:/tmp/flink-savepoints
+ # Mount your fat JARs here
+ - ./iggy-flink-examples/build/libs:/opt/flink/usrlib
+ # OR if you need shared connector libraries:
+ # - ./lib/iggy-connector.jar:/opt/flink/lib/iggy-connector.jar
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8081"]
+ interval: 10s
+ timeout: 5s
+ retries: 5
+ networks:
+ - flink-iggy-network
+ depends_on:
+ - iggy
+
+ # Flink TaskManager
+ taskmanager:
+ image: flink:2.1.0-scala_2.12-java21
+ container_name: flink-taskmanager
+ user: root # Start as root to fix permissions
+ entrypoint: /bin/bash
+ command:
+ - -c
+ - |
+ # Create checkpoint/savepoint directories with correct permissions
+ mkdir -p /tmp/flink-checkpoints /tmp/flink-savepoints
+ chown -R flink:flink /tmp/flink-checkpoints /tmp/flink-savepoints
+ chmod -R 755 /tmp/flink-checkpoints /tmp/flink-savepoints
+ # Switch to flink user and run taskmanager
+ exec gosu flink /docker-entrypoint.sh taskmanager
+ environment:
+ - IGGY_SERVER=iggy:8090
+ - |
+ FLINK_PROPERTIES=
+ jobmanager.rpc.address: jobmanager
+ taskmanager.memory.process.size: 1728m
+ taskmanager.numberOfTaskSlots: 4
+ volumes:
+ - flink-checkpoints:/tmp/flink-checkpoints
+ - flink-savepoints:/tmp/flink-savepoints
+ # Same library mounts as jobmanager if needed
+ # - ./lib/iggy-connector.jar:/opt/flink/lib/iggy-connector.jar
+ networks:
+ - flink-iggy-network
+ depends_on:
+ - jobmanager
+
+ # Iggy Setup - Creates streams and topics
+ iggy-setup:
+ image: alpine:latest
+ container_name: iggy-setup
+ depends_on:
+ - iggy
+ networks:
+ - flink-iggy-network
+ entrypoint: /bin/sh
+ command:
+ - -c
+ - |
+ # Install dependencies
+ echo "Installing dependencies..."
+ apk add --no-cache curl jq
+
+ echo "Waiting for Iggy to be ready..."
+ # Wait with health check instead of blind sleep
+ MAX_RETRIES=30
+ RETRY=0
+ while [ $$RETRY -lt $$MAX_RETRIES ]; do
+ if curl -f -s http://iggy:3000/stats > /dev/null 2>&1; then
+ echo "Iggy is ready!"
+ break
+ fi
+ RETRY=$$((RETRY + 1))
+ echo "Waiting for Iggy... ($$RETRY/$$MAX_RETRIES)"
+ sleep 2
+ done
+
+ if [ $$RETRY -eq $$MAX_RETRIES ]; then
+ echo "Iggy failed to start within timeout"
+ exit 1
+ fi
+
+ echo "Authenticating with Iggy..."
+ RESPONSE=$$(curl -s -X POST http://iggy:3000/users/login \
+ -H "Content-Type: application/json" \
+ -d '{"username":"iggy","password":"iggy"}')
+
+ TOKEN=$$(echo "$$RESPONSE" | jq -r '.access_token.token')
+
+ if [ -z "$$TOKEN" ] || [ "$$TOKEN" = "null" ]; then
+ echo "Failed to authenticate"
+ echo "Response was: $$RESPONSE"
+ exit 1
+ fi
+
+ echo "Authentication successful!"
+ echo "Creating streams and topics for examples..."
+
+ # StreamTransformJob streams
+ echo "Creating 'sensors' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "sensors"}' || true
+
+ echo "Creating 'readings' topic in 'sensors' stream..."
+ curl -s -X POST http://iggy:3000/streams/sensors/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "readings", "partitions_count": 4, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ echo "Creating 'alerts' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "alerts"}' || true
+
+ echo "Creating 'critical' topic in 'alerts' stream..."
+ curl -s -X POST http://iggy:3000/streams/alerts/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "critical", "partitions_count": 4, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ # WordCountJob streams
+ echo "Creating 'text-input' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "text-input"}' || true
+
+ echo "Creating 'lines' topic in 'text-input' stream..."
+ curl -s -X POST http://iggy:3000/streams/text-input/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "lines", "partitions_count": 2, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ echo "Creating 'word-counts' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "word-counts"}' || true
+
+ echo "Creating 'results' topic in 'word-counts' stream..."
+ curl -s -X POST http://iggy:3000/streams/word-counts/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "results", "partitions_count": 2, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ echo "Creating consumer group 'flink-word-counter' for text-input/lines..."
+ curl -s -X POST http://iggy:3000/streams/text-input/topics/lines/consumer-groups \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "flink-word-counter"}' || true
+
+ # MultiStreamJoinJob streams
+ echo "Creating 'user-activities' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "user-activities"}' || true
+
+ echo "Creating 'events' topic in 'user-activities' stream..."
+ curl -s -X POST http://iggy:3000/streams/user-activities/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "events", "partitions_count": 4, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ echo "Creating 'user-profiles' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "user-profiles"}' || true
+
+ echo "Creating 'updates' topic in 'user-profiles' stream..."
+ curl -s -X POST http://iggy:3000/streams/user-profiles/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "updates", "partitions_count": 2, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ echo "Creating 'enriched-activities' stream..."
+ curl -s -X POST http://iggy:3000/streams \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "enriched-activities"}' || true
+
+ echo "Creating 'results' topic in 'enriched-activities' stream..."
+ curl -s -X POST http://iggy:3000/streams/enriched-activities/topics \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $$TOKEN" \
+ -d '{"name": "results", "partitions_count": 4, "compression_algorithm": "none", "message_expiry": 0, "max_topic_size": 10737418240}' || true
+
+ echo "Stream and topic creation completed!"
+
+networks:
+ flink-iggy-network:
+ driver: bridge
+
+volumes:
+ flink-checkpoints:
+ flink-savepoints:
+ iggy-data:
diff --git a/foreign/java/external-processors/iggy-connector-flink/docker/Dockerfile.flink2_10-java21 b/foreign/java/external-processors/iggy-connector-flink/docker/Dockerfile.flink2_10-java21
new file mode 100644
index 000000000..85c140e6d
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/docker/Dockerfile.flink2_10-java21
@@ -0,0 +1,11 @@
+FROM flink:2.1.0-scala_2.12-java21
+
+# Set up directories for checkpoints and savepoints
+RUN mkdir -p /tmp/flink-checkpoints /tmp/flink-savepoints && \
+ chown -R flink:flink /tmp/flink-checkpoints /tmp/flink-savepoints && \
+ chmod -R 755 /tmp/flink-checkpoints /tmp/flink-savepoints
+
+USER flink
+
+# Default command (can be overridden)
+CMD ["jobmanager"]
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build.gradle.kts b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build.gradle.kts
index ce4768fcf..a7adb8812 100644
--- a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build.gradle.kts
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build.gradle.kts
@@ -31,6 +31,10 @@ repositories {
}
java {
+ // Target Java 17 for CI compatibility (Java 21 Flink Docker can run Java 17 bytecode)
+ sourceCompatibility = JavaVersion.VERSION_17
+ targetCompatibility = JavaVersion.VERSION_17
+
withJavadocJar()
withSourcesJar()
}
@@ -40,7 +44,7 @@ checkstyle {
configFile = file("../../../dev-support/checkstyle/checkstyle.xml")
}
-val flinkVersion = "1.18.0"
+val flinkVersion = "2.1.0"
val iggyVersion = "0.5.0-SNAPSHOT"
dependencies {
@@ -53,6 +57,7 @@ dependencies {
// Serialization support
implementation("com.fasterxml.jackson.core:jackson-databind:2.18.0")
+ implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.18.0")
// Logging
compileOnly("org.slf4j:slf4j-api:2.0.16")
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggyCommittable.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggyCommittable.java
new file mode 100644
index 000000000..dba00a269
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggyCommittable.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.sink;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Represents committable information for exactly-once delivery guarantees.
+ * Used in two-phase commit protocol with Flink checkpointing.
+ */
+public class IggyCommittable implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String streamId;
+ private final String topicId;
+ private final int partitionId;
+ private final long lastWrittenOffset;
+ private final int messageCount;
+
+ /**
+ * Creates a new committable instance.
+ *
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param partitionId the partition ID
+ * @param lastWrittenOffset the last offset written to this partition
+ * @param messageCount the number of messages in this commit
+ */
+ public IggyCommittable(
+ String streamId,
+ String topicId,
+ int partitionId,
+ long lastWrittenOffset,
+ int messageCount) {
+
+ if (streamId == null || streamId.isEmpty()) {
+ throw new IllegalArgumentException("streamId cannot be null or empty");
+ }
+ if (topicId == null || topicId.isEmpty()) {
+ throw new IllegalArgumentException("topicId cannot be null or empty");
+ }
+ if (partitionId < 0) {
+ throw new IllegalArgumentException("partitionId must be >= 0");
+ }
+ if (lastWrittenOffset < 0) {
+ throw new IllegalArgumentException("lastWrittenOffset must be >= 0");
+ }
+ if (messageCount < 0) {
+ throw new IllegalArgumentException("messageCount must be >= 0");
+ }
+
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.partitionId = partitionId;
+ this.lastWrittenOffset = lastWrittenOffset;
+ this.messageCount = messageCount;
+ }
+
+ public String getStreamId() {
+ return streamId;
+ }
+
+ public String getTopicId() {
+ return topicId;
+ }
+
+ public int getPartitionId() {
+ return partitionId;
+ }
+
+ public long getLastWrittenOffset() {
+ return lastWrittenOffset;
+ }
+
+ public int getMessageCount() {
+ return messageCount;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ IggyCommittable that = (IggyCommittable) o;
+ return partitionId == that.partitionId
+ && lastWrittenOffset == that.lastWrittenOffset
+ && messageCount == that.messageCount
+ && Objects.equals(streamId, that.streamId)
+ && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(streamId, topicId, partitionId, lastWrittenOffset, messageCount);
+ }
+
+ @Override
+ public String toString() {
+ return "IggyCommittable{"
+ + "streamId='" + streamId + '\''
+ + ", topicId='" + topicId + '\''
+ + ", partitionId=" + partitionId
+ + ", lastWrittenOffset=" + lastWrittenOffset
+ + ", messageCount=" + messageCount
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java
new file mode 100644
index 000000000..c2bb24f0c
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySink.java
@@ -0,0 +1,223 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.sink;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.serialization.SerializationSchema;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Duration;
+
+/**
+ * Flink Sink implementation for writing to Iggy streams.
+ * Implements the Flink Sink V2 API for integration with DataStream API.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * events.sinkTo(
+ * IggySink.<Event>builder()
+ * .setConnectionConfig(connectionConfig)
+ * .setStreamId("my-stream")
+ * .setTopicId("my-topic")
+ * .setSerializer(new JsonSerializationSchema<>())
+ * .setBatchSize(100)
+ * .setFlushInterval(Duration.ofSeconds(5))
+ * .withBalancedPartitioning()
+ * .build()
+ * ).name("Iggy Sink");
+ * }</pre>
+ *
+ * @param <T> the type of records to write
+ */
+public class IggySink<T> implements Sink<T>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IggyConnectionConfig connectionConfig;
+ private final String streamId;
+ private final String topicId;
+ private final SerializationSchema<T> serializer;
+ private final int batchSize;
+ private final Duration flushInterval;
+ private final IggySinkWriter.PartitioningStrategy partitioningStrategy;
+
+ /**
+ * Creates a new Iggy sink.
+ * Use {@link #builder()} to construct instances.
+ *
+ * @param connectionConfig the connection configuration
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param serializer the serialization schema
+ * @param batchSize the batch size for buffering
+ * @param flushInterval the maximum flush interval
+ * @param partitioningStrategy the partitioning strategy
+ */
+ public IggySink(
+ IggyConnectionConfig connectionConfig,
+ String streamId,
+ String topicId,
+ SerializationSchema<T> serializer,
+ int batchSize,
+ Duration flushInterval,
+ IggySinkWriter.PartitioningStrategy partitioningStrategy) {
+
+ this.connectionConfig = connectionConfig;
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.serializer = serializer;
+ this.batchSize = batchSize;
+ this.flushInterval = flushInterval;
+ this.partitioningStrategy = partitioningStrategy;
+ }
+
+ /**
+ * Creates a new builder for configuring the sink.
+ *
+ * @param <T> the type of records to write
+ * @return a new builder instance
+ */
+ public static <T> IggySinkBuilder<T> builder() {
+ return new IggySinkBuilder<>();
+ }
+
+ @Override
+ public SinkWriter<T> createWriter(WriterInitContext context) throws IOException {
+ IggyHttpClient httpClient = createHttpClient();
+ return new IggySinkWriter<>(
+ httpClient,
+ streamId,
+ topicId,
+ serializer,
+ batchSize,
+ flushInterval,
+ partitioningStrategy);
+ }
+
+ /**
+ * Creates an HTTP Iggy client based on connection configuration.
+ *
+ * @return configured HTTP Iggy client
+ */
+ private IggyHttpClient createHttpClient() {
+ try {
+ // Build HTTP URL from server address using URI for proper parsing
+ String serverAddress = connectionConfig.getServerAddress();
+
+ // Parse server address to extract host
+ java.net.URI uri = serverAddress.contains("://")
+ ? new java.net.URI(serverAddress)
+ : new java.net.URI("tcp://" + serverAddress);
+
+ String host = uri.getHost();
+ if (host == null) {
+ throw new IllegalArgumentException("Cannot extract host from: " + serverAddress);
+ }
+
+ // Build HTTP URL with port 3000 (Iggy HTTP API default port)
+ String httpUrl = "http://" + host + ":3000";
+
+ // Create HTTP client
+ IggyHttpClient httpClient = new IggyHttpClient(httpUrl);
+
+ // Login
+ httpClient.users().login(
+ connectionConfig.getUsername(),
+ connectionConfig.getPassword()
+ );
+
+ return httpClient;
+
+ } catch (java.net.URISyntaxException e) {
+ throw new RuntimeException("Invalid server address format: "
+ + connectionConfig.getServerAddress(), e);
+ } catch (RuntimeException e) {
+ throw new RuntimeException("Failed to create HTTP Iggy client", e);
+ }
+ }
+
+ /**
+ * Gets the connection configuration.
+ *
+ * @return connection configuration
+ */
+ public IggyConnectionConfig getConnectionConfig() {
+ return connectionConfig;
+ }
+
+ /**
+ * Gets the stream identifier.
+ *
+ * @return stream ID
+ */
+ public String getStreamId() {
+ return streamId;
+ }
+
+ /**
+ * Gets the topic identifier.
+ *
+ * @return topic ID
+ */
+ public String getTopicId() {
+ return topicId;
+ }
+
+ /**
+ * Gets the serialization schema.
+ *
+ * @return serializer
+ */
+ public SerializationSchema<T> getSerializer() {
+ return serializer;
+ }
+
+ /**
+ * Gets the batch size.
+ *
+ * @return batch size
+ */
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ /**
+ * Gets the flush interval.
+ *
+ * @return flush interval
+ */
+ public Duration getFlushInterval() {
+ return flushInterval;
+ }
+
+ /**
+ * Gets the partitioning strategy.
+ *
+ * @return partitioning strategy
+ */
+ public IggySinkWriter.PartitioningStrategy getPartitioningStrategy() {
+ return partitioningStrategy;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkBuilder.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkBuilder.java
new file mode 100644
index 000000000..f8fc7f6e8
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkBuilder.java
@@ -0,0 +1,220 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.sink;
+
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.serialization.SerializationSchema;
+
+import java.io.Serializable;
+import java.time.Duration;
+
+/**
+ * Builder for creating IggySink instances with fluent API.
+ *
+ * @param <T> the type of records to write
+ */
+public class IggySinkBuilder<T> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private IggyConnectionConfig connectionConfig;
+ private String streamId;
+ private String topicId;
+ private SerializationSchema<T> serializer;
+ private int batchSize = 100;
+ private Duration flushInterval = Duration.ofSeconds(5);
+ private IggySinkWriter.PartitioningStrategy partitioningStrategy =
+ IggySinkWriter.PartitioningStrategy.BALANCED;
+
+ /**
+ * Sets the connection configuration.
+ *
+ * @param connectionConfig the connection configuration
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setConnectionConfig(IggyConnectionConfig connectionConfig) {
+ this.connectionConfig = connectionConfig;
+ return this;
+ }
+
+ /**
+ * Sets the stream identifier.
+ *
+ * @param streamId the stream ID (numeric or string)
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setStreamId(String streamId) {
+ this.streamId = streamId;
+ return this;
+ }
+
+ /**
+ * Sets the stream identifier (numeric).
+ *
+ * @param streamId the stream ID
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setStreamId(long streamId) {
+ this.streamId = String.valueOf(streamId);
+ return this;
+ }
+
+ /**
+ * Sets the topic identifier.
+ *
+ * @param topicId the topic ID (numeric or string)
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setTopicId(String topicId) {
+ this.topicId = topicId;
+ return this;
+ }
+
+ /**
+ * Sets the topic identifier (numeric).
+ *
+ * @param topicId the topic ID
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setTopicId(long topicId) {
+ this.topicId = String.valueOf(topicId);
+ return this;
+ }
+
+ /**
+ * Sets the serialization schema.
+ *
+ * @param serializer the serialization schema
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setSerializer(SerializationSchema<T> serializer) {
+ this.serializer = serializer;
+ return this;
+ }
+
+ /**
+ * Sets the batch size (number of records to buffer before flushing).
+ *
+ * @param batchSize the batch size
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setBatchSize(int batchSize) {
+ if (batchSize <= 0) {
+ throw new IllegalArgumentException("batchSize must be > 0");
+ }
+ this.batchSize = batchSize;
+ return this;
+ }
+
+ /**
+ * Sets the flush interval (maximum time to wait before flushing).
+ *
+ * @param flushInterval the flush interval
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setFlushInterval(Duration flushInterval) {
+ if (flushInterval == null || flushInterval.isNegative()) {
+ throw new IllegalArgumentException("flushInterval must be positive");
+ }
+ this.flushInterval = flushInterval;
+ return this;
+ }
+
+ /**
+ * Sets the partitioning strategy.
+ *
+ * @param partitioningStrategy the partitioning strategy
+ * @return this builder
+ */
+ public IggySinkBuilder<T> setPartitioningStrategy(
+ IggySinkWriter.PartitioningStrategy partitioningStrategy) {
+ this.partitioningStrategy = partitioningStrategy;
+ return this;
+ }
+
+ /**
+ * Enables balanced partitioning (round-robin across partitions).
+ *
+ * @return this builder
+ */
+ public IggySinkBuilder<T> withBalancedPartitioning() {
+ this.partitioningStrategy = IggySinkWriter.PartitioningStrategy.BALANCED;
+ return this;
+ }
+
+ /**
+ * Enables partition ID based partitioning (uses partition key from records).
+ *
+ * @return this builder
+ */
+ public IggySinkBuilder<T> withPartitionIdPartitioning() {
+ this.partitioningStrategy = IggySinkWriter.PartitioningStrategy.PARTITION_ID;
+ return this;
+ }
+
+ /**
+ * Enables message key based partitioning (uses partition key as message key).
+ *
+ * @return this builder
+ */
+ public IggySinkBuilder<T> withMessageKeyPartitioning() {
+ this.partitioningStrategy = IggySinkWriter.PartitioningStrategy.MESSAGE_KEY;
+ return this;
+ }
+
+ /**
+ * Builds the IggySink instance.
+ *
+ * @return the configured IggySink
+ * @throws IllegalStateException if required configuration is missing
+ */
+ public IggySink<T> build() {
+ validate();
+
+ return new IggySink<>(
+ connectionConfig,
+ streamId,
+ topicId,
+ serializer,
+ batchSize,
+ flushInterval,
+ partitioningStrategy);
+ }
+
+ /**
+ * Validates the builder configuration.
+ *
+ * @throws IllegalStateException if required configuration is missing
+ */
+ private void validate() {
+ if (connectionConfig == null) {
+ throw new IllegalStateException("connectionConfig is required");
+ }
+ if (streamId == null || streamId.isEmpty()) {
+ throw new IllegalStateException("streamId is required");
+ }
+ if (topicId == null || topicId.isEmpty()) {
+ throw new IllegalStateException("topicId is required");
+ }
+ if (serializer == null) {
+ throw new IllegalStateException("serializer is required");
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
new file mode 100644
index 000000000..e12ecb88a
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/sink/IggySinkWriter.java
@@ -0,0 +1,303 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.sink;
+
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.MessageHeader;
+import org.apache.iggy.message.MessageId;
+import org.apache.iggy.message.Partitioning;
+import org.apache.iggy.connector.error.ConnectorException;
+import org.apache.iggy.connector.serialization.SerializationSchema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.zip.CRC32;
+import java.util.zip.Checksum;
+
+/**
+ * Sink writer implementation for writing records to Iggy.
+ * Buffers records and flushes them in batches for efficiency.
+ */
+public class IggySinkWriter<T> implements SinkWriter<T> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(IggySinkWriter.class);
+
+ private final IggyHttpClient httpClient;
+ private final String streamId;
+ private final String topicId;
+ private final SerializationSchema<T> serializer;
+ private final int batchSize;
+ private final Duration flushInterval;
+ private final PartitioningStrategy partitioningStrategy;
+
+ private final List<T> buffer;
+ private long lastFlushTime;
+ private long totalWritten;
+
+ /**
+ * Strategy for determining partition assignment.
+ */
+ public enum PartitioningStrategy {
+ BALANCED,
+ PARTITION_ID,
+ MESSAGE_KEY
+ }
+
+ /**
+ * Creates a new sink writer.
+ *
+ * @param httpClient the HTTP Iggy client
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param serializer the serialization schema
+ * @param batchSize maximum number of records to buffer before flushing
+ * @param flushInterval maximum time to wait before flushing
+ * @param partitioningStrategy the partitioning strategy
+ */
+ public IggySinkWriter(
+ IggyHttpClient httpClient,
+ String streamId,
+ String topicId,
+ SerializationSchema<T> serializer,
+ int batchSize,
+ Duration flushInterval,
+ PartitioningStrategy partitioningStrategy) {
+
+ if (httpClient == null) {
+ throw new IllegalArgumentException("httpClient cannot be null");
+ }
+ if (streamId == null || streamId.isEmpty()) {
+ throw new IllegalArgumentException("streamId cannot be null or empty");
+ }
+ if (topicId == null || topicId.isEmpty()) {
+ throw new IllegalArgumentException("topicId cannot be null or empty");
+ }
+ if (serializer == null) {
+ throw new IllegalArgumentException("serializer cannot be null");
+ }
+ if (batchSize <= 0) {
+ throw new IllegalArgumentException("batchSize must be > 0");
+ }
+ if (flushInterval == null || flushInterval.isNegative()) {
+ throw new IllegalArgumentException("flushInterval must be positive");
+ }
+
+ this.httpClient = httpClient;
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.serializer = serializer;
+ this.batchSize = batchSize;
+ this.flushInterval = flushInterval;
+ this.partitioningStrategy = partitioningStrategy != null
+ ? partitioningStrategy
+ : PartitioningStrategy.BALANCED;
+
+ this.buffer = new ArrayList<>(batchSize);
+ this.lastFlushTime = System.currentTimeMillis();
+ this.totalWritten = 0;
+ }
+
+ @Override
+ public void write(T element, Context context) throws IOException {
+ LOGGER.debug("IggySinkWriter.write() called - element: {}, buffer size: {}",
+ element, buffer.size());
+ buffer.add(element);
+
+ // Flush if batch size reached or flush interval exceeded
+ if (buffer.size() >= batchSize || shouldFlushByTime()) {
+ LOGGER.debug("IggySinkWriter: Flushing buffer of size {}", buffer.size());
+ flush(false);
+ }
+ }
+
+ @Override
+ public void flush(boolean endOfInput) throws IOException {
+ if (buffer.isEmpty()) {
+ LOGGER.debug("IggySinkWriter.flush() - buffer is empty, skipping");
+ return;
+ }
+
+ LOGGER.debug("IggySinkWriter.flush() - flushing {} messages to stream={}, topic={}",
+ buffer.size(), streamId, topicId);
+
+ try {
+ // Serialize all buffered records
+ List<Message> messages = new ArrayList<>(buffer.size());
+ for (T element : buffer) {
+ byte[] payload = serializer.serialize(element);
+ // Use createMessage to avoid String conversion
+ messages.add(createMessage(payload));
+ }
+
+ // Determine partitioning
+ Partitioning partitioning = determinePartitioning();
+
+ // Send messages to Iggy via HTTP
+ StreamId stream = parseStreamId(streamId);
+ TopicId topic = parseTopicId(topicId);
+
+ LOGGER.debug("IggySinkWriter: Sending {} messages with partitioning={}",
+ messages.size(), partitioning);
+ httpClient.messages().sendMessages(stream, topic, partitioning, messages);
+
+ totalWritten += buffer.size();
+ LOGGER.debug("IggySinkWriter: Successfully sent {} messages. Total written: {}",
+ buffer.size(), totalWritten);
+
+ // Clear buffer and update flush time
+ buffer.clear();
+ lastFlushTime = System.currentTimeMillis();
+
+ } catch (Exception e) {
+ LOGGER.error("IggySinkWriter.flush() - ERROR: {}", e.getMessage(), e);
+ throw new ConnectorException(
+ "Failed to send messages to Iggy",
+ e,
+ ConnectorException.ErrorCode.RESOURCE_EXHAUSTED,
+ true);
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ // Flush any remaining buffered records
+ flush(true);
+ // Note: HTTP client doesn't have close() method - connections managed by Java HttpClient pool
+ }
+
+ /**
+ * Checks if flush should occur based on time interval.
+ *
+ * @return true if flush interval has been exceeded
+ */
+ private boolean shouldFlushByTime() {
+ long timeSinceLastFlush = System.currentTimeMillis() - lastFlushTime;
+ return timeSinceLastFlush >= flushInterval.toMillis();
+ }
+
+ /**
+ * Determines the partitioning strategy for the current batch.
+ *
+ * @return the partitioning configuration
+ */
+ private Partitioning determinePartitioning() {
+ switch (partitioningStrategy) {
+ case BALANCED:
+ return Partitioning.balanced();
+
+ case PARTITION_ID:
+ // Use first element's partition key if available
+ if (!buffer.isEmpty()) {
+ Optional<Integer> partitionKey = serializer.extractPartitionKey(buffer.get(0));
+ if (partitionKey.isPresent()) {
+ return Partitioning.partitionId(partitionKey.get().longValue());
+ }
+ }
+ return Partitioning.balanced();
+
+ case MESSAGE_KEY:
+ // Use first element's partition key as message key
+ if (!buffer.isEmpty()) {
+ Optional<Integer> partitionKey = serializer.extractPartitionKey(buffer.get(0));
+ if (partitionKey.isPresent()) {
+ return Partitioning.messagesKey(String.valueOf(partitionKey.get()));
+ }
+ }
+ return Partitioning.balanced();
+
+ default:
+ return Partitioning.balanced();
+ }
+ }
+
+ /**
+ * Parses stream ID from string (supports both numeric and named streams).
+ */
+ private StreamId parseStreamId(String streamIdStr) {
+ try {
+ return StreamId.of(Long.parseLong(streamIdStr));
+ } catch (NumberFormatException e) {
+ return StreamId.of(streamIdStr);
+ }
+ }
+
+ /**
+ * Parses topic ID from string (supports both numeric and named topics).
+ */
+ private TopicId parseTopicId(String topicIdStr) {
+ try {
+ return TopicId.of(Long.parseLong(topicIdStr));
+ } catch (NumberFormatException e) {
+ return TopicId.of(topicIdStr);
+ }
+ }
+
+ /**
+ * Gets the total number of records written.
+ *
+ * @return total records written
+ */
+ public long getTotalWritten() {
+ return totalWritten;
+ }
+
+ /**
+ * Gets the current buffer size.
+ *
+ * @return number of buffered records
+ */
+ public int getBufferSize() {
+ return buffer.size();
+ }
+
+ /**
+ * Creates a Message from raw bytes without String conversion.
+ * This avoids charset encoding issues that can corrupt binary data.
+ *
+ * @param payload the message payload as bytes
+ * @return a Message instance
+ */
+ private Message createMessage(byte[] payload) {
+ Checksum crc32 = new CRC32();
+ crc32.update(payload, 0, payload.length);
+ BigInteger checksum = BigInteger.valueOf(crc32.getValue());
+
+ MessageHeader header = new MessageHeader(
+ checksum,
+ MessageId.serverGenerated(),
+ BigInteger.ZERO, // offset
+ BigInteger.ZERO, // timestamp
+ BigInteger.ZERO, // originTimestamp
+ 0L, // userHeadersLength
+ (long) payload.length); // payloadLength
+
+ return new Message(header, payload, Optional.empty());
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggyPartitionSplitReader.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggyPartitionSplitReader.java
new file mode 100644
index 000000000..5ae2fff99
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggyPartitionSplitReader.java
@@ -0,0 +1,257 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.PolledMessages;
+import org.apache.iggy.message.PollingStrategy;
+import org.apache.iggy.connector.error.ConnectorException;
+import org.apache.iggy.connector.serialization.DeserializationSchema;
+import org.apache.iggy.connector.serialization.RecordMetadata;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reads messages from a single Iggy partition.
+ * Manages offset tracking and message deserialization.
+ */
+public class IggyPartitionSplitReader<T> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(IggyPartitionSplitReader.class);
+
+ private final AsyncIggyTcpClient asyncClient;
+ private final IggySourceSplit split;
+ private final DeserializationSchema<T> deserializer;
+ private final Consumer consumer;
+ private final long pollBatchSize;
+
+ private long currentOffset;
+ private boolean hasMoreData;
+ private boolean consumerGroupJoined;
+
+ /**
+ * Creates a new partition split reader.
+ *
+ * @param asyncClient the async Iggy client
+ * @param split the source split to read from
+ * @param deserializer the deserialization schema
+ * @param consumer the consumer identifier
+ * @param pollBatchSize number of messages to fetch per poll
+ */
+ public IggyPartitionSplitReader(
+ AsyncIggyTcpClient asyncClient,
+ IggySourceSplit split,
+ DeserializationSchema<T> deserializer,
+ Consumer consumer,
+ long pollBatchSize) {
+
+ if (asyncClient == null) {
+ throw new IllegalArgumentException("asyncClient cannot be null");
+ }
+ if (split == null) {
+ throw new IllegalArgumentException("split cannot be null");
+ }
+ if (deserializer == null) {
+ throw new IllegalArgumentException("deserializer cannot be null");
+ }
+ if (consumer == null) {
+ throw new IllegalArgumentException("consumer cannot be null");
+ }
+ if (pollBatchSize <= 0) {
+ throw new IllegalArgumentException("pollBatchSize must be > 0");
+ }
+
+ this.asyncClient = asyncClient;
+ this.split = split;
+ this.deserializer = deserializer;
+ this.consumer = consumer;
+ this.pollBatchSize = pollBatchSize;
+ this.currentOffset = split.getCurrentOffset();
+ this.hasMoreData = true;
+ this.consumerGroupJoined = false;
+ }
+
+ /**
+ * Polls messages from the partition.
+ *
+ * @return list of deserialized records, empty if no messages available
+ * @throws ConnectorException if polling or deserialization fails
+ */
+ public List<T> poll() {
+ // For unbounded streams, always poll (don't check hasMoreData)
+
+ try {
+ StreamId streamId = parseStreamId(split.getStreamId());
+ TopicId topicId = parseTopicId(split.getTopicId());
+
+ // Join consumer group on first poll (idempotent operation in Iggy)
+ if (!consumerGroupJoined) {
+ LOGGER.info("IggyPartitionSplitReader: Joining consumer group for stream={}, "
+ + "topic={}, consumer={}, partition={}",
+ split.getStreamId(), split.getTopicId(), consumer.id(),
+ split.getPartitionId());
+ asyncClient.consumerGroups()
+ .joinConsumerGroup(streamId, topicId, consumer.id())
+ .join();
+ consumerGroupJoined = true;
+ LOGGER.info("IggyPartitionSplitReader: Successfully joined consumer group");
+ }
+ Optional<Long> partitionId = Optional.of((long) split.getPartitionId());
+
+ // CRITICAL FIX: Use consumer group managed offset (next available message)
+ // When using consumer groups, Iggy manages offsets automatically.
+ // Using PollingStrategy.offset() conflicts with consumer group offset management.
+ PollingStrategy strategy = PollingStrategy.next();
+
+ LOGGER.info("IggyPartitionSplitReader: Polling partition={}, "
+ + "strategy=NEXT (consumer-group-managed), batchSize={}",
+ split.getPartitionId(), pollBatchSize);
+
+ // CRITICAL FIX: Enable autoCommit to advance consumer group offset after each poll
+ // Without this, offset never advances and we read the same messages repeatedly
+ // Poll messages from Iggy (async with blocking)
+ PolledMessages polledMessages = asyncClient.messages()
+ .pollMessagesAsync(streamId, topicId, partitionId, consumer, strategy,
+ pollBatchSize, true)
+ .join();
+
+ LOGGER.info("IggyPartitionSplitReader: Polled partition={}, messagesCount={}, "
+ + "currentOffset={}",
+ split.getPartitionId(), polledMessages.messages().size(),
+ polledMessages.currentOffset() != null
+ ? polledMessages.currentOffset().longValue() : "null");
+
+ if (polledMessages.messages().isEmpty()) {
+ // For unbounded streams, keep polling even if no messages available
+ // Don't set hasMoreData = false
+ return List.of();
+ }
+
+ // Deserialize messages
+ List<T> records = new ArrayList<>(polledMessages.messages().size());
+ for (Message message : polledMessages.messages()) {
+ T record = deserializeMessage(message);
+ if (record != null) {
+ LOGGER.info("IggyPartitionSplitReader: Deserialized message at "
+ + "offset={}, record={}",
+ message.header().offset(), record);
+ records.add(record);
+ }
+ }
+
+ // Update current offset
+ if (polledMessages.currentOffset() != null) {
+ currentOffset = polledMessages.currentOffset().longValue() + 1;
+ LOGGER.info("IggyPartitionSplitReader: Updated currentOffset to {}",
+ currentOffset);
+ }
+
+ return records;
+
+ } catch (Exception e) {
+ throw new ConnectorException(
+ "Failed to poll messages from partition " + split.getPartitionId(),
+ e,
+ ConnectorException.ErrorCode.RESOURCE_EXHAUSTED,
+ true);
+ }
+ }
+
+ /**
+ * Deserializes a single message.
+ *
+ * @param message the Iggy message
+ * @return deserialized record, or null if deserialization fails (logs warning)
+ */
+ private T deserializeMessage(Message message) {
+ try {
+ RecordMetadata metadata = RecordMetadata.of(
+ split.getStreamId(),
+ split.getTopicId(),
+ split.getPartitionId(),
+ message.header().offset().longValue());
+
+ return deserializer.deserialize(message.payload(), metadata);
+
+ } catch (IOException e) {
+ // Log warning and skip this message (can be made configurable)
+ LOGGER.warn("Failed to deserialize message at offset {}: {}",
+ message.header().offset(), e.getMessage(), e);
+ return null;
+ }
+ }
+
+ /**
+ * Gets the current offset position.
+ *
+ * @return current offset
+ */
+ public long getCurrentOffset() {
+ return currentOffset;
+ }
+
+ /**
+ * Gets the source split being read.
+ *
+ * @return the source split
+ */
+ public IggySourceSplit getSplit() {
+ return split.withCurrentOffset(currentOffset);
+ }
+
+ /**
+ * Checks if there is potentially more data to read.
+ *
+ * @return true if more data might be available
+ */
+ public boolean hasMoreData() {
+ return hasMoreData;
+ }
+
+ /**
+ * Parses stream ID from string (supports both numeric and named streams).
+ */
+ private StreamId parseStreamId(String streamIdStr) {
+ try {
+ return StreamId.of(Long.parseLong(streamIdStr));
+ } catch (NumberFormatException e) {
+ return StreamId.of(streamIdStr);
+ }
+ }
+
+ /**
+ * Parses topic ID from string (supports both numeric and named topics).
+ */
+ private TopicId parseTopicId(String topicIdStr) {
+ try {
+ return TopicId.of(Long.parseLong(topicIdStr));
+ } catch (NumberFormatException e) {
+ return TopicId.of(topicIdStr);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java
new file mode 100644
index 000000000..9c25982f9
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySource.java
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.flink.api.connector.source.*;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.config.OffsetConfig;
+
+import java.io.Serializable;
+
+/**
+ * Flink Source implementation for reading from Iggy streams.
+ * Implements the Flink Source API for integration with DataStream API.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * DataStream<Event> events = env.fromSource(
+ * IggySource.<Event>builder()
+ * .setConnectionConfig(connectionConfig)
+ * .setStreamId("my-stream")
+ * .setTopicId("my-topic")
+ * .setConsumerGroup("flink-job-1")
+ * .setDeserializer(new JsonDeserializationSchema<>(Event.class))
+ * .build(),
+ * WatermarkStrategy.noWatermarks(),
+ * "Iggy Source"
+ * );
+ * }</pre>
+ *
+ * @param <T> the type of records produced by this source
+ */
+public class IggySource<T> implements Source<T, IggySourceSplit, IggySourceEnumeratorState>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final IggyConnectionConfig connectionConfig;
+ private final String streamId;
+ private final String topicId;
+ private final String consumerGroupName; // Serializable representation
+ private final org.apache.iggy.connector.serialization.DeserializationSchema<T> deserializer;
+ private final OffsetConfig offsetConfig;
+ private final long pollBatchSize;
+
+ /**
+ * Creates a new Iggy source.
+ * Use {@link #builder()} to construct instances.
+ *
+ * @param connectionConfig the connection configuration
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param consumerGroupName the consumer group name
+ * @param deserializer the deserialization schema
+ * @param offsetConfig the offset configuration
+ * @param pollBatchSize the number of messages to fetch per poll
+ */
+ public IggySource(
+ IggyConnectionConfig connectionConfig,
+ String streamId,
+ String topicId,
+ String consumerGroupName,
+ org.apache.iggy.connector.serialization.DeserializationSchema<T> deserializer,
+ OffsetConfig offsetConfig,
+ long pollBatchSize) {
+
+ this.connectionConfig = connectionConfig;
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.consumerGroupName = consumerGroupName;
+ this.deserializer = deserializer;
+ this.offsetConfig = offsetConfig;
+ this.pollBatchSize = pollBatchSize;
+ }
+
+ /**
+ * Creates a new builder for configuring the source.
+ *
+ * @param <T> the type of records produced by the source
+ * @return a new builder instance
+ */
+ public static <T> IggySourceBuilder<T> builder() {
+ return new IggySourceBuilder<>();
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<T, IggySourceSplit> createReader(SourceReaderContext readerContext) throws Exception {
+ AsyncIggyTcpClient asyncClient = createAsyncIggyClient();
+ Consumer consumer = createConsumer();
+ return new IggySourceReader<>(
+ readerContext,
+ asyncClient,
+ deserializer,
+ consumer,
+ pollBatchSize);
+ }
+
+ @Override
+ public SplitEnumerator<IggySourceSplit, IggySourceEnumeratorState> createEnumerator(
+ SplitEnumeratorContext<IggySourceSplit> enumContext) throws Exception {
+
+ AsyncIggyTcpClient asyncClient = createAsyncIggyClient();
+ return new IggySourceSplitEnumerator(
+ enumContext,
+ asyncClient,
+ streamId,
+ topicId,
+ consumerGroupName,
+ offsetConfig,
+ null);
+ }
+
+ @Override
+ public SplitEnumerator<IggySourceSplit, IggySourceEnumeratorState> restoreEnumerator(
+ SplitEnumeratorContext<IggySourceSplit> enumContext,
+ IggySourceEnumeratorState checkpoint) throws Exception {
+
+ AsyncIggyTcpClient asyncClient = createAsyncIggyClient();
+ return new IggySourceSplitEnumerator(
+ enumContext,
+ asyncClient,
+ streamId,
+ topicId,
+ consumerGroupName,
+ offsetConfig,
+ checkpoint);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<IggySourceSplit> getSplitSerializer() {
+ return new IggySourceSplitSerializer();
+ }
+
+ @Override
+ public SimpleVersionedSerializer<IggySourceEnumeratorState> getEnumeratorCheckpointSerializer() {
+ return new IggySourceEnumeratorStateSerializer();
+ }
+
+ /**
+ * Creates an async Iggy TCP client based on connection configuration.
+ *
+ * @return configured async Iggy client
+ */
+ private AsyncIggyTcpClient createAsyncIggyClient() {
+ try {
+ // Parse host and port from server address
+ String serverAddress = connectionConfig.getServerAddress();
+ String host;
+ int port = 8090; // Default TCP port
+
+ if (serverAddress.contains(":")) {
+ String[] parts = serverAddress.split(":");
+ host = parts[0];
+ port = Integer.parseInt(parts[1]);
+ } else {
+ host = serverAddress;
+ }
+
+ // Create async TCP client using builder pattern
+ AsyncIggyTcpClient asyncClient = AsyncIggyTcpClient.builder()
+ .host(host)
+ .port(port)
+ .credentials(connectionConfig.getUsername(), connectionConfig.getPassword())
+ .connectionPoolSize(4)
+ .build();
+
+ // Connect and block (auto-login happens during connect)
+ asyncClient.connect().join();
+
+ return asyncClient;
+
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create async Iggy client", e);
+ }
+ }
+
+ /**
+ * Creates a Consumer object from the stored consumer group name.
+ *
+ * @return configured Consumer
+ */
+ private Consumer createConsumer() {
+ return Consumer.group(org.apache.iggy.identifier.ConsumerId.of(consumerGroupName));
+ }
+
+ /**
+ * Gets the connection configuration.
+ *
+ * @return connection configuration
+ */
+ public IggyConnectionConfig getConnectionConfig() {
+ return connectionConfig;
+ }
+
+ /**
+ * Gets the stream identifier.
+ *
+ * @return stream ID
+ */
+ public String getStreamId() {
+ return streamId;
+ }
+
+ /**
+ * Gets the topic identifier.
+ *
+ * @return topic ID
+ */
+ public String getTopicId() {
+ return topicId;
+ }
+
+ /**
+ * Gets the consumer group name.
+ *
+ * @return consumer group name
+ */
+ public String getConsumerGroupName() {
+ return consumerGroupName;
+ }
+
+ /**
+ * Gets the deserialization schema.
+ *
+ * @return deserializer
+ */
+ public org.apache.iggy.connector.serialization.DeserializationSchema<T> getDeserializer() {
+ return deserializer;
+ }
+
+ /**
+ * Gets the offset configuration.
+ *
+ * @return offset config
+ */
+ public OffsetConfig getOffsetConfig() {
+ return offsetConfig;
+ }
+
+ /**
+ * Gets the poll batch size.
+ *
+ * @return batch size
+ */
+ public long getPollBatchSize() {
+ return pollBatchSize;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceBuilder.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceBuilder.java
new file mode 100644
index 000000000..827b58301
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceBuilder.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.config.OffsetConfig;
+import org.apache.iggy.connector.serialization.DeserializationSchema;
+
+import java.io.Serializable;
+
+/**
+ * Builder for creating IggySource instances with fluent API.
+ *
+ * @param <T> the type of records produced by the source
+ */
+public class IggySourceBuilder<T> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private IggyConnectionConfig connectionConfig;
+ private String streamId;
+ private String topicId;
+ private String consumerGroupName; // Store as String for serialization
+ private DeserializationSchema<T> deserializer;
+ private OffsetConfig offsetConfig = OffsetConfig.builder().build();
+ private long pollBatchSize = 100L;
+
+ /**
+ * Sets the connection configuration.
+ *
+ * @param connectionConfig the connection configuration
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setConnectionConfig(IggyConnectionConfig connectionConfig) {
+ this.connectionConfig = connectionConfig;
+ return this;
+ }
+
+ /**
+ * Sets the stream identifier.
+ *
+ * @param streamId the stream ID (numeric or string)
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setStreamId(String streamId) {
+ this.streamId = streamId;
+ return this;
+ }
+
+ /**
+ * Sets the stream identifier (numeric).
+ *
+ * @param streamId the stream ID
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setStreamId(long streamId) {
+ this.streamId = String.valueOf(streamId);
+ return this;
+ }
+
+ /**
+ * Sets the topic identifier.
+ *
+ * @param topicId the topic ID (numeric or string)
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setTopicId(String topicId) {
+ this.topicId = topicId;
+ return this;
+ }
+
+ /**
+ * Sets the topic identifier (numeric).
+ *
+ * @param topicId the topic ID
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setTopicId(long topicId) {
+ this.topicId = String.valueOf(topicId);
+ return this;
+ }
+
+ /**
+ * Sets the consumer group name.
+ *
+ * @param consumerGroup the consumer group name
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setConsumerGroup(String consumerGroup) {
+ this.consumerGroupName = consumerGroup;
+ return this;
+ }
+
+ /**
+ * Sets the deserialization schema.
+ *
+ * @param deserializer the deserialization schema
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setDeserializer(DeserializationSchema<T> deserializer) {
+ this.deserializer = deserializer;
+ return this;
+ }
+
+ /**
+ * Sets the offset configuration.
+ *
+ * @param offsetConfig the offset configuration
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setOffsetConfig(OffsetConfig offsetConfig) {
+ this.offsetConfig = offsetConfig;
+ return this;
+ }
+
+ /**
+ * Sets the number of messages to fetch per poll.
+ *
+ * @param pollBatchSize the batch size
+ * @return this builder
+ */
+ public IggySourceBuilder<T> setPollBatchSize(long pollBatchSize) {
+ if (pollBatchSize <= 0) {
+ throw new IllegalArgumentException("pollBatchSize must be > 0");
+ }
+ this.pollBatchSize = pollBatchSize;
+ return this;
+ }
+
+ /**
+ * Builds the IggySource instance.
+ *
+ * @return the configured IggySource
+ * @throws IllegalStateException if required configuration is missing
+ */
+ public IggySource<T> build() {
+ validate();
+
+ return new IggySource<>(
+ connectionConfig,
+ streamId,
+ topicId,
+ consumerGroupName,
+ deserializer,
+ offsetConfig,
+ pollBatchSize);
+ }
+
+ /**
+ * Validates the builder configuration.
+ *
+ * @throws IllegalStateException if required configuration is missing
+ */
+ private void validate() {
+ if (connectionConfig == null) {
+ throw new IllegalStateException("connectionConfig is required");
+ }
+ if (streamId == null || streamId.isEmpty()) {
+ throw new IllegalStateException("streamId is required");
+ }
+ if (topicId == null || topicId.isEmpty()) {
+ throw new IllegalStateException("topicId is required");
+ }
+ if (consumerGroupName == null || consumerGroupName.isEmpty()) {
+ throw new IllegalStateException("consumerGroup is required (use setConsumerGroup)");
+ }
+ if (deserializer == null) {
+ throw new IllegalStateException("deserializer is required");
+ }
+ if (offsetConfig == null) {
+ throw new IllegalStateException("offsetConfig cannot be null");
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceEnumeratorState.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceEnumeratorState.java
new file mode 100644
index 000000000..2ca39a493
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceEnumeratorState.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import java.io.Serializable;
+import java.util.*;
+
+/**
+ * State for the Iggy source split enumerator.
+ * Used for checkpointing and recovery.
+ */
+public class IggySourceEnumeratorState implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Set<IggySourceSplit> assignedSplits;
+ private final Set<Integer> discoveredPartitions;
+
+ /**
+ * Creates a new enumerator state.
+ *
+ * @param assignedSplits the splits that have been assigned to readers
+ * @param discoveredPartitions the partition IDs that have been discovered
+ */
+ public IggySourceEnumeratorState(
+ Set<IggySourceSplit> assignedSplits,
+ Set<Integer> discoveredPartitions) {
+ this.assignedSplits = new HashSet<>(assignedSplits);
+ this.discoveredPartitions = new HashSet<>(discoveredPartitions);
+ }
+
+ /**
+ * Creates an empty initial state.
+ *
+ * @return empty enumerator state
+ */
+ public static IggySourceEnumeratorState empty() {
+ return new IggySourceEnumeratorState(Collections.emptySet(), Collections.emptySet());
+ }
+
+ public Set<IggySourceSplit> getAssignedSplits() {
+ return Collections.unmodifiableSet(assignedSplits);
+ }
+
+ public Set<Integer> getDiscoveredPartitions() {
+ return Collections.unmodifiableSet(discoveredPartitions);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ IggySourceEnumeratorState that = (IggySourceEnumeratorState) o;
+ return Objects.equals(assignedSplits, that.assignedSplits)
+ && Objects.equals(discoveredPartitions, that.discoveredPartitions);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(assignedSplits, discoveredPartitions);
+ }
+
+ @Override
+ public String toString() {
+ return "IggySourceEnumeratorState{"
+ + "assignedSplits=" + assignedSplits.size()
+ + ", discoveredPartitions=" + discoveredPartitions.size()
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceEnumeratorStateSerializer.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceEnumeratorStateSerializer.java
new file mode 100644
index 000000000..333777df6
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceEnumeratorStateSerializer.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.*;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Serializer for IggySourceEnumeratorState for checkpointing.
+ */
+public class IggySourceEnumeratorStateSerializer
+ implements SimpleVersionedSerializer<IggySourceEnumeratorState> {
+
+ private static final int VERSION = 1;
+
+ private final IggySourceSplitSerializer splitSerializer = new IggySourceSplitSerializer();
+
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public byte[] serialize(IggySourceEnumeratorState state) throws IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+
+ // Write assigned splits
+ Set<IggySourceSplit> assignedSplits = state.getAssignedSplits();
+ out.writeInt(assignedSplits.size());
+ for (IggySourceSplit split : assignedSplits) {
+ byte[] splitBytes = splitSerializer.serialize(split);
+ out.writeInt(splitBytes.length);
+ out.write(splitBytes);
+ }
+
+ // Write discovered partitions
+ Set<Integer> discoveredPartitions = state.getDiscoveredPartitions();
+ out.writeInt(discoveredPartitions.size());
+ for (Integer partitionId : discoveredPartitions) {
+ out.writeInt(partitionId);
+ }
+
+ out.flush();
+ return baos.toByteArray();
+ }
+ }
+
+ @Override
+ public IggySourceEnumeratorState deserialize(int version, byte[] serialized) throws IOException {
+ if (version != VERSION) {
+ throw new IOException("Unsupported version: " + version);
+ }
+
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ DataInputStream in = new DataInputStream(bais)) {
+
+ // Read assigned splits
+ int splitsCount = in.readInt();
+ Set<IggySourceSplit> assignedSplits = new HashSet<>();
+ for (int i = 0; i < splitsCount; i++) {
+ int splitBytesLength = in.readInt();
+ byte[] splitBytes = new byte[splitBytesLength];
+ in.readFully(splitBytes);
+ IggySourceSplit split = splitSerializer.deserialize(
+ splitSerializer.getVersion(), splitBytes);
+ assignedSplits.add(split);
+ }
+
+ // Read discovered partitions
+ int partitionsCount = in.readInt();
+ Set<Integer> discoveredPartitions = new HashSet<>();
+ for (int i = 0; i < partitionsCount; i++) {
+ discoveredPartitions.add(in.readInt());
+ }
+
+ return new IggySourceEnumeratorState(assignedSplits, discoveredPartitions);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceReader.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceReader.java
new file mode 100644
index 000000000..46e4e5c1b
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceReader.java
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.SourceEvent;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.connector.serialization.DeserializationSchema;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Flink SourceReader implementation for Iggy.
+ * Manages multiple partition split readers and coordinates reading from them.
+ */
+public class IggySourceReader<T> implements SourceReader<T, IggySourceSplit> {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(IggySourceReader.class);
+
+ private final SourceReaderContext context;
+ private final AsyncIggyTcpClient asyncClient;
+ private final DeserializationSchema<T> deserializer;
+ private final Consumer consumer;
+ private final long pollBatchSize;
+
+ private final Map<String, IggyPartitionSplitReader<T>> splitReaders;
+ private final Queue<T> recordQueue;
+
+ private boolean noMoreSplits;
+
+ /**
+ * Creates a new Iggy source reader.
+ *
+ * @param context the source reader context
+ * @param asyncClient the async Iggy client
+ * @param deserializer the deserialization schema
+ * @param consumer the consumer identifier
+ * @param pollBatchSize number of messages to fetch per poll
+ */
+ public IggySourceReader(
+ SourceReaderContext context,
+ AsyncIggyTcpClient asyncClient,
+ DeserializationSchema<T> deserializer,
+ Consumer consumer,
+ long pollBatchSize) {
+
+ this.context = context;
+ this.asyncClient = asyncClient;
+ this.deserializer = deserializer;
+ this.consumer = consumer;
+ this.pollBatchSize = pollBatchSize;
+ this.splitReaders = new HashMap<>();
+ this.recordQueue = new ArrayDeque<>();
+ this.noMoreSplits = false;
+ }
+
+ @Override
+ public void start() {
+ // Initialization if needed
+ }
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<T> output) throws Exception {
+ // If we have records in the queue, emit them
+ if (!recordQueue.isEmpty()) {
+ T record = recordQueue.poll();
+ output.collect(record);
+ return InputStatus.MORE_AVAILABLE;
+ }
+
+ // Poll from all split readers
+ boolean hasData = false;
+ for (IggyPartitionSplitReader<T> splitReader : splitReaders.values()) {
+ List<T> records = splitReader.poll();
+ if (!records.isEmpty()) {
+ recordQueue.addAll(records);
+ hasData = true;
+ }
+ }
+
+ // If we got data, emit the first record
+ if (hasData) {
+ T record = recordQueue.poll();
+ output.collect(record);
+ return InputStatus.MORE_AVAILABLE;
+ }
+
+ // No data available
+ if (noMoreSplits && splitReaders.isEmpty()) {
+ return InputStatus.END_OF_INPUT;
+ }
+
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+
+ @Override
+ public List<IggySourceSplit> snapshotState(long checkpointId) {
+ // Return current state of all splits for checkpointing
+ List<IggySourceSplit> splits = new ArrayList<>();
+ for (IggyPartitionSplitReader<T> splitReader : splitReaders.values()) {
+ splits.add(splitReader.getSplit());
+ }
+ return splits;
+ }
+
+ @Override
+ public CompletableFuture<Void> isAvailable() {
+ // If we have data in queue, immediately available
+ if (!recordQueue.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ // Check if any split reader has more data
+ for (IggyPartitionSplitReader<T> splitReader : splitReaders.values()) {
+ if (splitReader.hasMoreData()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ }
+
+ // No data available right now, return a future that will complete later
+ // For simplicity, we return completed future and rely on polling
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void addSplits(List<IggySourceSplit> splits) {
+ for (IggySourceSplit split : splits) {
+ String splitId = split.splitId();
+ if (!splitReaders.containsKey(splitId)) {
+ IggyPartitionSplitReader<T> splitReader = new IggyPartitionSplitReader<>(
+ asyncClient,
+ split,
+ deserializer,
+ consumer,
+ pollBatchSize);
+ splitReaders.put(splitId, splitReader);
+ }
+ }
+ }
+
+ @Override
+ public void notifyNoMoreSplits() {
+ noMoreSplits = true;
+ }
+
+ @Override
+ public void handleSourceEvents(SourceEvent sourceEvent) {
+ // Handle custom source events if needed
+ }
+
+ @Override
+ public void close() throws Exception {
+ // Clean up resources
+ splitReaders.clear();
+ recordQueue.clear();
+
+ // CRITICAL FIX: Close async client to shutdown EventLoopGroup and prevent memory leak
+ if (asyncClient != null) {
+ try {
+ asyncClient.close().join();
+ } catch (RuntimeException e) {
+ // Log but don't throw - we're in cleanup phase
+ LOGGER.error("Error closing async Iggy client: " + e.getMessage());
+ }
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplit.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplit.java
new file mode 100644
index 000000000..e9f5f077a
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplit.java
@@ -0,0 +1,163 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.flink.api.connector.source.SourceSplit;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Represents a single Iggy partition as a Flink source split.
+ * Each split corresponds to one partition in an Iggy topic.
+ */
+public class IggySourceSplit implements SourceSplit, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String streamId;
+ private final String topicId;
+ private final int partitionId;
+ private final long startOffset;
+ private final long currentOffset;
+
+ /**
+ * Creates a new Iggy source split.
+ *
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param partitionId the partition ID
+ * @param startOffset the initial offset to start reading from
+ * @param currentOffset the current offset (for checkpointing)
+ */
+ public IggySourceSplit(
+ String streamId,
+ String topicId,
+ int partitionId,
+ long startOffset,
+ long currentOffset) {
+
+ if (streamId == null || streamId.isEmpty()) {
+ throw new IllegalArgumentException("streamId cannot be null or empty");
+ }
+ if (topicId == null || topicId.isEmpty()) {
+ throw new IllegalArgumentException("topicId cannot be null or empty");
+ }
+ if (partitionId < 0) {
+ throw new IllegalArgumentException("partitionId must be >= 0");
+ }
+ if (startOffset < 0) {
+ throw new IllegalArgumentException("startOffset must be >= 0");
+ }
+ if (currentOffset < startOffset) {
+ throw new IllegalArgumentException("currentOffset must be >= startOffset");
+ }
+
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.partitionId = partitionId;
+ this.startOffset = startOffset;
+ this.currentOffset = currentOffset;
+ }
+
+ /**
+ * Creates a new split starting from the specified offset.
+ *
+ * @param streamId the stream identifier
+ * @param topicId the topic identifier
+ * @param partitionId the partition ID
+ * @param startOffset the offset to start reading from
+ * @return a new source split
+ */
+ public static IggySourceSplit create(
+ String streamId,
+ String topicId,
+ int partitionId,
+ long startOffset) {
+ return new IggySourceSplit(streamId, topicId, partitionId, startOffset, startOffset);
+ }
+
+ @Override
+ public String splitId() {
+ return String.format("%s-%s-%d", streamId, topicId, partitionId);
+ }
+
+ /**
+ * Creates a new split with updated current offset (for checkpointing).
+ *
+ * @param newOffset the new current offset
+ * @return a new split with the updated offset
+ */
+ public IggySourceSplit withCurrentOffset(long newOffset) {
+ return new IggySourceSplit(streamId, topicId, partitionId, startOffset, newOffset);
+ }
+
+ public String getStreamId() {
+ return streamId;
+ }
+
+ public String getTopicId() {
+ return topicId;
+ }
+
+ public int getPartitionId() {
+ return partitionId;
+ }
+
+ public long getStartOffset() {
+ return startOffset;
+ }
+
+ public long getCurrentOffset() {
+ return currentOffset;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ IggySourceSplit that = (IggySourceSplit) o;
+ return partitionId == that.partitionId
+ && startOffset == that.startOffset
+ && currentOffset == that.currentOffset
+ && Objects.equals(streamId, that.streamId)
+ && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(streamId, topicId, partitionId, startOffset, currentOffset);
+ }
+
+ @Override
+ public String toString() {
+ return "IggySourceSplit{"
+ + "streamId='" + streamId + '\''
+ + ", topicId='" + topicId + '\''
+ + ", partitionId=" + partitionId
+ + ", startOffset=" + startOffset
+ + ", currentOffset=" + currentOffset
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplitEnumerator.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplitEnumerator.java
new file mode 100644
index 000000000..069a4d26f
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplitEnumerator.java
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.connector.source.SplitsAssignment;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.connector.config.OffsetConfig;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.partition.Partition;
+import org.apache.iggy.topic.TopicDetails;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * Split enumerator for Iggy source.
+ * Discovers partitions and assigns them to source readers.
+ */
+public class IggySourceSplitEnumerator implements
+ SplitEnumerator<IggySourceSplit, IggySourceEnumeratorState> {
+
+ private final SplitEnumeratorContext<IggySourceSplit> context;
+ private final AsyncIggyTcpClient asyncClient;
+ private final String streamId;
+ private final String topicId;
+ private final String consumerGroupName;
+ private final OffsetConfig offsetConfig;
+
+ private final Set<IggySourceSplit> assignedSplits;
+ private final Set<Integer> discoveredPartitions;
+ private final Set<String> sentToReaders; // Track splits already sent to readers
+
+ /**
+ * Creates a new split enumerator.
+ *
+ * @param context the split enumerator context
+ * @param asyncClient the async Iggy client
+ * @param streamId the stream ID
+ * @param topicId the topic ID
+ * @param consumerGroupName the consumer group name
+ * @param offsetConfig the offset configuration
+ * @param initialState the initial state (for recovery)
+ */
+ public IggySourceSplitEnumerator(
+ SplitEnumeratorContext<IggySourceSplit> context,
+ AsyncIggyTcpClient asyncClient,
+ String streamId,
+ String topicId,
+ String consumerGroupName,
+ OffsetConfig offsetConfig,
+ @Nullable IggySourceEnumeratorState initialState) {
+
+ this.context = context;
+ this.asyncClient = asyncClient;
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.consumerGroupName = consumerGroupName;
+ this.offsetConfig = offsetConfig;
+
+ if (initialState != null) {
+ this.assignedSplits = new HashSet<>(initialState.getAssignedSplits());
+ this.discoveredPartitions = new HashSet<>(initialState.getDiscoveredPartitions());
+ } else {
+ this.assignedSplits = new HashSet<>();
+ this.discoveredPartitions = new HashSet<>();
+ }
+ this.sentToReaders = new HashSet<>();
+ }
+
+ @Override
+ public void start() {
+ // Create consumer group if it doesn't exist
+ ensureConsumerGroupExists();
+
+ // Discover initial partitions
+ discoverPartitions();
+ }
+
+ /**
+ * Ensures the consumer group exists, creating it if necessary.
+ *
+ * Note: In Iggy SDK async API, consumer group creation is not available.
+ * Consumer groups must be created beforehand using the blocking HTTP client
+ * or through the setup scripts. The async API only supports joining/leaving
+ * consumer groups.
+ */
+ private void ensureConsumerGroupExists() {
+ // Consumer group creation removed - must be created externally
+ // The docker-compose setup already creates consumer groups via iggy-setup
+ // Users should ensure consumer groups exist before starting Flink jobs
+ }
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
+ // Not used in batch assignment mode
+ }
+
+ @Override
+ public void addSplitsBack(List<IggySourceSplit> splits, int subtaskId) {
+ // Re-add splits that were returned (e.g., on reader failure)
+ for (IggySourceSplit split : splits) {
+ assignedSplits.remove(split);
+ }
+ // Reassign them
+ assignSplits();
+ }
+
+ @Override
+ public void addReader(int subtaskId) {
+ // New reader registered, assign splits to it
+ assignSplits();
+ }
+
+ @Override
+ public IggySourceEnumeratorState snapshotState(long checkpointId) throws Exception {
+ return new IggySourceEnumeratorState(assignedSplits, discoveredPartitions);
+ }
+
+ @Override
+ public void close() throws IOException {
+ // Clean up resources if needed
+ }
+
+ /**
+ * Discovers partitions from Iggy topic.
+ */
+ private void discoverPartitions() {
+ try {
+ StreamId stream = parseStreamId(streamId);
+ TopicId topic = parseTopicId(topicId);
+
+ // Get topic details including partitions
+ Optional<TopicDetails> topicDetailsOpt = asyncClient.topics()
+ .getTopicAsync(stream, topic)
+ .join();
+
+ if (topicDetailsOpt.isPresent()) {
+ TopicDetails topicDetails = topicDetailsOpt.get();
+ if (topicDetails.partitions() != null) {
+ for (Partition partition : topicDetails.partitions()) {
+ int partitionId = partition.id().intValue();
+
+ // Skip if already discovered
+ if (discoveredPartitions.contains(partitionId)) {
+ continue;
+ }
+
+ // Determine starting offset
+ long startOffset = determineStartOffset(partition);
+
+ // Create split
+ IggySourceSplit split = IggySourceSplit.create(
+ streamId,
+ topicId,
+ partitionId,
+ startOffset);
+
+ discoveredPartitions.add(partitionId);
+
+ // Check if this split is already assigned
+ boolean alreadyAssigned = assignedSplits.stream()
+ .anyMatch(s -> s.getPartitionId() == partitionId);
+
+ if (!alreadyAssigned) {
+ assignedSplits.add(split);
+ }
+ }
+ }
+ }
+
+ // Assign discovered splits to readers
+ assignSplits();
+
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to discover partitions", e);
+ }
+ }
+
+ /**
+ * Assigns pending splits to available readers.
+ */
+ private void assignSplits() {
+ if (context.registeredReaders().isEmpty()) {
+ return;
+ }
+
+ // Get only splits that haven't been sent to readers yet
+ List<IggySourceSplit> unassignedSplits = assignedSplits.stream()
+ .filter(split -> !sentToReaders.contains(split.splitId()))
+ .collect(java.util.stream.Collectors.toList());
+
+ if (unassignedSplits.isEmpty()) {
+ return;
+ }
+
+ // Simple round-robin assignment
+ Map<Integer, List<IggySourceSplit>> assignments = new HashMap<>();
+ List<Integer> readers = new ArrayList<>(context.registeredReaders().keySet());
+
+ for (int i = 0; i < unassignedSplits.size(); i++) {
+ int readerIndex = i % readers.size();
+ int readerId = readers.get(readerIndex);
+
+ assignments.computeIfAbsent(readerId, k -> new ArrayList<>())
+ .add(unassignedSplits.get(i));
+ }
+
+ // Assign splits to readers and mark them as sent
+ for (Map.Entry<Integer, List<IggySourceSplit>> entry : assignments.entrySet()) {
+ context.assignSplits(new SplitsAssignment<>(
+ Map.of(entry.getKey(), entry.getValue())));
+
+ // Mark these splits as sent
+ for (IggySourceSplit split : entry.getValue()) {
+ sentToReaders.add(split.splitId());
+ }
+ }
+
+ // DON'T signal no more splits - this is a continuous stream!
+ // Removed: context.signalNoMoreSplits(readerId);
+ }
+
+ /**
+ * Determines the starting offset for a partition based on offset configuration.
+ *
+ * @param partition the partition
+ * @return the starting offset
+ */
+ private long determineStartOffset(Partition partition) {
+ return switch (offsetConfig.getResetStrategy()) {
+ case EARLIEST -> 0L;
+ case LATEST -> partition.currentOffset().longValue();
+ case NONE -> 0L; // Default to earliest
+ };
+ }
+
+ /**
+ * Parses stream ID from string (supports both numeric and named streams).
+ */
+ private StreamId parseStreamId(String streamIdStr) {
+ try {
+ return StreamId.of(Long.parseLong(streamIdStr));
+ } catch (NumberFormatException e) {
+ return StreamId.of(streamIdStr);
+ }
+ }
+
+ /**
+ * Parses topic ID from string (supports both numeric and named topics).
+ */
+ private TopicId parseTopicId(String topicIdStr) {
+ try {
+ return TopicId.of(Long.parseLong(topicIdStr));
+ } catch (NumberFormatException e) {
+ return TopicId.of(topicIdStr);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplitSerializer.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplitSerializer.java
new file mode 100644
index 000000000..c34d7f9b1
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/flink/source/IggySourceSplitSerializer.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.flink.source;
+
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+
+import java.io.*;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Serializer for IggySourceSplit for checkpointing.
+ */
+public class IggySourceSplitSerializer implements SimpleVersionedSerializer<IggySourceSplit> {
+
+ private static final int VERSION = 1;
+
+ @Override
+ public int getVersion() {
+ return VERSION;
+ }
+
+ @Override
+ public byte[] serialize(IggySourceSplit split) throws IOException {
+ try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ DataOutputStream out = new DataOutputStream(baos)) {
+
+ // Write streamId
+ byte[] streamIdBytes = split.getStreamId().getBytes(StandardCharsets.UTF_8);
+ out.writeInt(streamIdBytes.length);
+ out.write(streamIdBytes);
+
+ // Write topicId
+ byte[] topicIdBytes = split.getTopicId().getBytes(StandardCharsets.UTF_8);
+ out.writeInt(topicIdBytes.length);
+ out.write(topicIdBytes);
+
+ // Write partitionId
+ out.writeInt(split.getPartitionId());
+
+ // Write startOffset
+ out.writeLong(split.getStartOffset());
+
+ // Write currentOffset
+ out.writeLong(split.getCurrentOffset());
+
+ out.flush();
+ return baos.toByteArray();
+ }
+ }
+
+ @Override
+ public IggySourceSplit deserialize(int version, byte[] serialized) throws IOException {
+ if (version != VERSION) {
+ throw new IOException("Unsupported version: " + version);
+ }
+
+ try (ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+ DataInputStream in = new DataInputStream(bais)) {
+
+ // Read streamId
+ int streamIdLength = in.readInt();
+ byte[] streamIdBytes = new byte[streamIdLength];
+ in.readFully(streamIdBytes);
+ String streamId = new String(streamIdBytes, StandardCharsets.UTF_8);
+
+ // Read topicId
+ int topicIdLength = in.readInt();
+ byte[] topicIdBytes = new byte[topicIdLength];
+ in.readFully(topicIdBytes);
+ String topicId = new String(topicIdBytes, StandardCharsets.UTF_8);
+
+ // Read partitionId
+ int partitionId = in.readInt();
+
+ // Read startOffset
+ long startOffset = in.readLong();
+
+ // Read currentOffset
+ long currentOffset = in.readLong();
+
+ return new IggySourceSplit(streamId, topicId, partitionId, startOffset, currentOffset);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/JsonDeserializationSchema.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/JsonDeserializationSchema.java
new file mode 100644
index 000000000..f8107a610
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/JsonDeserializationSchema.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.serialization;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+
+import java.io.IOException;
+
+/**
+ * Deserialization schema for JSON using Jackson.
+ * Supports Java 8 time types and handles unknown properties gracefully.
+ *
+ * @param <T> the type to deserialize to
+ */
+public class JsonDeserializationSchema<T> implements DeserializationSchema<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Class<T> typeClass;
+ private final transient ObjectMapper objectMapper;
+
+ /**
+ * Creates a new JSON deserializer for the specified type.
+ *
+ * @param typeClass the class to deserialize to
+ */
+ public JsonDeserializationSchema(Class<T> typeClass) {
+ this(typeClass, createDefaultObjectMapper());
+ }
+
+ /**
+ * Creates a new JSON deserializer with a custom ObjectMapper.
+ *
+ * @param typeClass the class to deserialize to
+ * @param objectMapper the Jackson ObjectMapper to use
+ */
+ public JsonDeserializationSchema(Class<T> typeClass, ObjectMapper objectMapper) {
+ if (typeClass == null) {
+ throw new IllegalArgumentException("typeClass cannot be null");
+ }
+ if (objectMapper == null) {
+ throw new IllegalArgumentException("objectMapper cannot be null");
+ }
+ this.typeClass = typeClass;
+ this.objectMapper = objectMapper;
+ }
+
+ @Override
+ public T deserialize(byte[] data, RecordMetadata metadata) throws IOException {
+ if (data == null || data.length == 0) {
+ return null;
+ }
+
+ try {
+ return getObjectMapper().readValue(data, typeClass);
+ } catch (IOException e) {
+ throw new IOException(
+ "Failed to deserialize JSON for type " + typeClass.getName(), e);
+ }
+ }
+
+ @Override
+ public TypeDescriptor<T> getProducedType() {
+ return TypeDescriptor.of(typeClass);
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Gets the ObjectMapper, creating a default one if needed (for deserialization).
+ *
+ * @return the ObjectMapper instance
+ */
+ private ObjectMapper getObjectMapper() {
+ if (objectMapper != null) {
+ return objectMapper;
+ }
+ // Fallback for deserialized instances
+ return createDefaultObjectMapper();
+ }
+
+ /**
+ * Creates a default ObjectMapper configured for common use cases.
+ *
+ * @return configured ObjectMapper
+ */
+ private static ObjectMapper createDefaultObjectMapper() {
+ ObjectMapper mapper = new ObjectMapper();
+
+ // Register Java 8 time module for LocalDateTime, Instant, etc.
+ mapper.registerModule(new JavaTimeModule());
+
+ // Don't fail on unknown properties
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+ // Write dates as ISO-8601 strings, not timestamps
+ mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+
+ return mapper;
+ }
+
+ /**
+ * Gets the target type class.
+ *
+ * @return the type class
+ */
+ public Class<T> getTypeClass() {
+ return typeClass;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/JsonSerializationSchema.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/JsonSerializationSchema.java
new file mode 100644
index 000000000..2bf418da6
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/JsonSerializationSchema.java
@@ -0,0 +1,169 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.serialization;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+
+import java.io.IOException;
+import java.util.Optional;
+import java.util.function.Function;
+
+/**
+ * Serialization schema for JSON using Jackson.
+ * Supports Java 8 time types and optional partition key extraction.
+ *
+ * @param <T> the type to serialize from
+ */
+public class JsonSerializationSchema<T> implements SerializationSchema<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JsonSerializationSchema.class);
+
+ private static final long serialVersionUID = 1L;
+
+ private final transient ObjectMapper objectMapper;
+ private final Function<T, Integer> partitionKeyExtractor;
+
+ /**
+ * Creates a new JSON serializer.
+ */
+ public JsonSerializationSchema() {
+ this(createDefaultObjectMapper(), null);
+ }
+
+ /**
+ * Creates a new JSON serializer with partition key extractor.
+ *
+ * @param partitionKeyExtractor function to extract partition key from elements
+ */
+ public JsonSerializationSchema(Function<T, Integer> partitionKeyExtractor) {
+ this(createDefaultObjectMapper(), partitionKeyExtractor);
+ }
+
+ /**
+ * Creates a new JSON serializer with partition key extractor.
+ * Static factory method to avoid constructor ambiguity.
+ *
+ * @param partitionKeyExtractor function to extract partition key from elements
+ * @param <T> the type to serialize
+ * @return new JsonSerializationSchema instance
+ */
+ public static <T> JsonSerializationSchema<T> withPartitionKeyExtractor(
+ Function<T, Integer> partitionKeyExtractor) {
+ return new JsonSerializationSchema<>(partitionKeyExtractor);
+ }
+
+ /**
+ * Creates a new JSON serializer with a custom ObjectMapper.
+ *
+ * @param objectMapper the Jackson ObjectMapper to use
+ */
+ public JsonSerializationSchema(ObjectMapper objectMapper) {
+ this(objectMapper, null);
+ }
+
+ /**
+ * Creates a new JSON serializer with custom ObjectMapper and partition key extractor.
+ *
+ * @param objectMapper the Jackson ObjectMapper to use
+ * @param partitionKeyExtractor function to extract partition key from elements
+ */
+ public JsonSerializationSchema(
+ ObjectMapper objectMapper,
+ Function<T, Integer> partitionKeyExtractor) {
+ if (objectMapper == null) {
+ throw new IllegalArgumentException("objectMapper cannot be null");
+ }
+ this.objectMapper = objectMapper;
+ this.partitionKeyExtractor = partitionKeyExtractor;
+ }
+
+ @Override
+ public byte[] serialize(T element) throws IOException {
+ if (element == null) {
+ return null;
+ }
+
+ try {
+ return getObjectMapper().writeValueAsBytes(element);
+ } catch (IOException e) {
+ throw new IOException(
+ "Failed to serialize object of type " + element.getClass().getName(), e);
+ }
+ }
+
+ @Override
+ public Optional<Integer> extractPartitionKey(T element) {
+ if (partitionKeyExtractor != null && element != null) {
+ try {
+ Integer key = partitionKeyExtractor.apply(element);
+ return Optional.ofNullable(key);
+ } catch (RuntimeException e) {
+ // Log and return empty if extraction fails
+ LOG.warn("Failed to extract partitionKey: {}", e.getMessage());
+ return Optional.empty();
+ }
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+
+ /**
+ * Gets the ObjectMapper, creating a default one if needed (for deserialization).
+ *
+ * @return the ObjectMapper instance
+ */
+ private ObjectMapper getObjectMapper() {
+ if (objectMapper != null) {
+ return objectMapper;
+ }
+ // Fallback for deserialized instances
+ return createDefaultObjectMapper();
+ }
+
+ /**
+ * Creates a default ObjectMapper configured for common use cases.
+ *
+ * @return configured ObjectMapper
+ */
+ private static ObjectMapper createDefaultObjectMapper() {
+ ObjectMapper mapper = new ObjectMapper();
+
+ // Register Java 8 time module for LocalDateTime, Instant, etc.
+ mapper.registerModule(new JavaTimeModule());
+
+ // Don't fail on unknown properties
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+
+ // Write dates as ISO-8601 strings, not timestamps
+ mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+
+ return mapper;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/StringDeserializationSchema.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/StringDeserializationSchema.java
new file mode 100644
index 000000000..ce1e9232f
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/StringDeserializationSchema.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.connector.serialization;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Deserialization schema for UTF-8 strings.
+ */
+public class StringDeserializationSchema implements DeserializationSchema<String> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String charsetName;
+ private transient Charset charset;
+
+ /**
+ * Creates a new string deserializer using UTF-8 encoding.
+ */
+ public StringDeserializationSchema() {
+ this(StandardCharsets.UTF_8);
+ }
+
+ /**
+ * Creates a new string deserializer with specified charset.
+ *
+ * @param charset the character encoding to use
+ */
+ public StringDeserializationSchema(Charset charset) {
+ this.charsetName = charset.name();
+ this.charset = charset;
+ }
+
+ /**
+ * Custom serialization to handle non-serializable Charset.
+ */
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+ }
+
+ /**
+ * Custom deserialization to recreate the Charset from its name.
+ */
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+ this.charset = Charset.forName(charsetName);
+ }
+
+ @Override
+ public String deserialize(byte[] data, RecordMetadata metadata) throws IOException {
+ if (data == null) {
+ return null;
+ }
+ return new String(data, charset);
+ }
+
+ @Override
+ public TypeDescriptor<String> getProducedType() {
+ return TypeDescriptor.of(String.class);
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/build.gradle.kts b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/build.gradle.kts
new file mode 100644
index 000000000..a199f1307
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/build.gradle.kts
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.gradle.external.javadoc.StandardJavadocDocletOptions
+
+plugins {
+ application
+ id("com.github.johnrengelman.shadow") version "8.1.1"
+}
+
+group = "org.apache.iggy"
+version = "0.5.0-SNAPSHOT"
+
+repositories {
+ mavenCentral()
+}
+
+val flinkVersion = "2.1.0"
+
+application {
+ mainClass.set("org.apache.iggy.flink.example.StreamTransformJob")
+}
+
+dependencies {
+ // Depends on the connector library
+ implementation(project(":iggy-connector-library"))
+
+ // Flink runtime (provided by cluster in production)
+ compileOnly("org.apache.flink:flink-streaming-java:${flinkVersion}")
+ compileOnly("org.apache.flink:flink-clients:${flinkVersion}")
+
+ // For local development/testing, include Flink at runtime
+ runtimeOnly("org.apache.flink:flink-streaming-java:${flinkVersion}")
+ runtimeOnly("org.apache.flink:flink-clients:${flinkVersion}")
+
+ // Jackson for JSON serialization in model classes
+ implementation("com.fasterxml.jackson.core:jackson-databind:2.18.0")
+ implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.18.0")
+
+ // Include in fat jar for standalone execution
+ implementation("org.slf4j:slf4j-simple:2.0.16")
+ implementation("com.typesafe:config:1.4.3")
+
+ // Testing
+ testImplementation("org.junit.jupiter:junit-jupiter:5.11.3")
+ testImplementation("org.assertj:assertj-core:3.26.3")
+ testRuntimeOnly("org.slf4j:slf4j-simple:2.0.16")
+}
+
+tasks.withType<Test> {
+ useJUnitPlatform()
+ // Exclude integration tests that require a running Iggy server
+ // These should be run separately with: ./gradlew integrationTest
+ exclude("**/AsyncTcp*Test.class", "**/SendTextDataTest.class")
+}
+
+tasks.shadowJar {
+ archiveBaseName.set("flink-iggy-examples")
+ archiveVersion.set("")
+ archiveClassifier.set("")
+
+ // Merge service files
+ mergeServiceFiles()
+
+ // Relocate conflicting dependencies if needed
+ // relocate("com.google", "org.apache.iggy.shaded.com.google")
+}
+
+java {
+ // Target Java 17 for CI compatibility (Java 21 Flink Docker can run Java 17 bytecode)
+ sourceCompatibility = JavaVersion.VERSION_17
+ targetCompatibility = JavaVersion.VERSION_17
+}
+
+tasks.withType<JavaCompile> {
+ options.encoding = "UTF-8"
+ options.compilerArgs.add("-parameters")
+}
+
+tasks.withType<Javadoc> {
+ options.encoding = "UTF-8"
+ (options as StandardJavadocDocletOptions).apply {
+ addStringOption("Xdoclint:none", "-quiet")
+ addBooleanOption("html5", true)
+ }
+}
+
+tasks.withType<Jar> {
+ manifest {
+ attributes["Main-Class"] = application.mainClass.get()
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/data-population-explanation.md b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/data-population-explanation.md
new file mode 100644
index 000000000..ed52f0ea3
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/data-population-explanation.md
@@ -0,0 +1,279 @@
+# Break Down Iggy HTTP API Message Payload
+
+## Example curl Command
+
+```bash
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r"}]}'
+```
+
+## URL Breakdown
+
+```text
+http://localhost:3000/streams/3/topics/1/messages
+ │ │
+ │ └─ Topic ID: 1 (named "lines")
+ └─────────── Stream ID: 3 (named "text-input")
+```
+
+You can use either numeric IDs or names:
+
+- `/streams/3/topics/1/messages` ← numeric IDs
+- `/streams/text-input/topics/lines/messages` ← names
+
+## JSON Payload Structure
+
+```json
+{
+ "partitioning": {
+ "kind": "balanced",
+ "value": ""
+ },
+ "messages": [
+ {
+ "payload": "c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r"
+ }
+ ]
+}
+```
+
+## 1. partitioning Object
+
+Determines which partition the message goes to:
+
+```json
+{
+ "kind": "balanced",
+ "value": ""
+}
+```
+
+Available partitioning strategies:
+
+### "balanced" (Round-robin distribution)
+
+```json
+{"kind": "balanced", "value": ""}
+```
+
+- Distributes messages evenly across all partitions
+- `value` must be empty string
+
+### "partition_id" (Direct partition assignment)
+
+```json
+{"kind": "partition_id", "value": "AQ=="}
+```
+
+Partition 1 (base64 encoded)
+
+- Send to a specific partition
+- `value` must be the partition ID encoded in base64
+- Example: Partition 1 → `\x01` → base64 → `"AQ=="`
+- Example: Partition 2 → `\x02` → base64 → `"Ag=="`
+
+### "messages_key" (Hash-based partitioning)
+
+```json
+{"kind": "messages_key", "value": "dXNlcjEyMw=="}
+```
+
+"user123" in base64
+
+- Messages with the same key go to the same partition
+- `value` is the key encoded in base64
+- Ensures ordering for messages with the same key
+
+## 2. messages Array
+
+Array of messages to send (can batch multiple messages):
+
+```json
+{
+ "messages": [
+ {
+ "payload": "c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r",
+ "id": 123,
+ "headers": {
+ "user": "alice",
+ "trace_id": "abc123"
+ }
+ },
+ {
+ "payload": "YW5vdGhlciBtZXNzYWdl"
+ }
+ ]
+}
+```
+
+Field descriptions:
+
+- `payload`: Base64-encoded message content (required)
+- `id`: Optional message ID (auto-generated if omitted)
+- `headers`: Optional custom headers
+
+## 3. payload Field (BASE64 ENCODED)
+
+This is the actual message content, but it MUST be base64 encoded.
+
+```json
+"payload": "c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r"
+```
+
+Let me decode this:
+
+```bash
+echo "c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r" | base64 -d
+```
+
+Output: `streaming data processing with flink`
+
+### Why Base64?
+
+1. **Binary safety**: Iggy messages are arbitrary bytes, not necessarily UTF-8 text
+2. **JSON compatibility**: Raw bytes can't be represented in JSON
+3. **Consistency**: Same API works for text, binary data, images, etc.
+
+## How to Create Payloads
+
+### For text messages
+
+Encode text to base64:
+
+```bash
+echo -n "hello world" | base64
+```
+
+Output: `aGVsbG8gd29ybGQ=`
+
+Use in curl:
+
+```bash
+curl -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"aGVsbG8gd29ybGQ="}]}'
+```
+
+### For binary data
+
+Encode a file to base64:
+
+```bash
+cat image.png | base64 > image.b64
+```
+
+Use the base64 string in your JSON payload.
+
+## Complete Examples
+
+### Example 1: Simple text message
+
+Original text: "hello world"
+Base64: `aGVsbG8gd29ybGQ=`
+
+```bash
+curl -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{
+ "partitioning": {"kind": "balanced", "value": ""},
+ "messages": [{"payload": "aGVsbG8gd29ybGQ="}]
+ }'
+```
+
+### Example 2: Multiple messages
+
+```bash
+curl -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{
+ "partitioning": {"kind": "balanced", "value": ""},
+ "messages": [
+ {"payload": "aGVsbG8gd29ybGQ="},
+ {"payload": "aGVsbG8gZmxpbms="}
+ ]
+ }'
+```
+
+Note: Payloads decode to "hello world" and "hello flink"
+
+### Example 3: With message ID and headers
+
+```bash
+curl -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{
+ "partitioning": {"kind": "balanced", "value": ""},
+ "messages": [{
+ "id": 12345,
+ "payload": "aGVsbG8gd29ybGQ=",
+ "headers": {
+ "user_id": "alice",
+ "request_id": "req-001"
+ }
+ }]
+ }'
+```
+
+### Example 4: Partition-specific
+
+Send to partition 1:
+
+- Partition ID 1 as bytes: `[0x01]`
+- Base64 encoded: `"AQ=="`
+
+```bash
+curl -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{
+ "partitioning": {"kind": "partition_id", "value": "AQ=="},
+ "messages": [{"payload": "aGVsbG8gd29ybGQ="}]
+ }'
+```
+
+## How Flink Receives It
+
+When your Flink job reads this message:
+
+1. IggySource polls messages from Iggy
+1. Receives raw bytes (Iggy automatically decodes base64)
+1. StringDeserializationSchema converts bytes to String:
+
+ ```java
+ public String deserialize(byte[] data, RecordMetadata metadata) {
+ return new String(data, charset); // Converts bytes to "hello world"
+ }
+ ```
+
+1. Your Flink job processes the text: "hello world"
+
+## Helper Script for Encoding
+
+Save this as `encode-message.sh`:
+
+```bash
+#!/bin/bash
+TEXT="$1"
+BASE64=$(echo -n "$TEXT" | base64)
+echo "Original: $TEXT"
+echo "Base64: $BASE64"
+echo ""
+echo "curl command:"
+echo "curl -X POST \"http://localhost:3000/streams/3/topics/1/messages\" \\"
+echo " -H \"Content-Type: application/json\" \\"
+echo " -H \"Authorization: Bearer \$TOKEN\" \\"
+echo " -d '{\"partitioning\":{\"kind\":\"balanced\",\"value\":\"\"},\"messages\":[{\"payload\":\"$BASE64\"}]}'"
+```
+
+Usage:
+
+```bash
+./encode-message.sh "hello world"
+```
+
+Shows you the base64 and complete curl command.
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/deploy-and-run-word-count-job.md b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/deploy-and-run-word-count-job.md
new file mode 100644
index 000000000..bc7a704b2
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/deploy-and-run-word-count-job.md
@@ -0,0 +1,258 @@
+# Here's a complete step-by-step guide to redeploy, test, and monitor the WordCountJob
+
+1. Build the Project
+
+ ```bash
+ cd $IGGY_HOME/foreign/java
+ ./gradlew :iggy-flink-examples:shadowJar --console=plain
+ ```
+
+1. Start Docker Compose (if not already running)
+
+ ```bash
+ cd $IGGY_HOME/foreign/java/external-processors/iggy-connector-flink
+ docker-compose up -d
+ ```
+
+ Wait for services to be healthy (~30 seconds):
+
+ ```bash
+ sleep 30
+ ```
+
+1. Submit the Job to Flink
+
+ ```bash
+ docker exec flink-jobmanager flink run -d \
+ -c org.apache.iggy.flink.example.WordCountJob \
+ /opt/flink/usrlib/flink-iggy-examples.jar
+ ```
+
+ Note the JobID that's returned (e.g., Job has been submitted with JobID abc123...)
+
+1. Verify Job is Running
+
+ ```bash
+ docker exec flink-jobmanager flink list
+ ```
+
+ You should see your job with status RUNNING.
+
+1. Send Test Messages to Iggy
+
+## Get authentication token
+
+<!-- markdownlint-disable MD034 -->
+
+```bash
+TOKEN=$(curl -s -X POST http://localhost:3000/users/login \
+ -H "Content-Type: application/json" \
+ -d '{"username":"iggy","password":"iggy"}' | \
+ grep -o '"token":"[^"]*"' | cut -d'"' -f4)
+```
+
+## Send test messages (base64 encoded payloads)
+
+```bash
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"aGVsbG8gd29ybGQgaGVsbG8gZmxpbms="}]}'
+echo "✓ Sent: hello world hello flink"
+
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"YXBhY2hlIGZsaW5rIGNvbm5lY3RvciBmb3IgaWdneQ=="}]}'
+echo "✓ Sent: apache flink connector for iggy"
+
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r"}]}'
+echo "✓ Sent: streaming data processing with flink"
+```
+
+1. Monitor Job Processing
+
+ Check TaskManager logs for message processing:
+
+ ```bash
+ docker logs flink-taskmanager --tail 50 | grep -E "Deserialized message"
+ ```
+
+ You should see messages being deserialized.
+
+ Check offset advancement (ensure no infinite loop):
+
+ ```bash
+ docker logs flink-taskmanager --tail 30 | grep -E "currentOffset"
+ ```
+
+ The offset should advance and stay at the new position (not reset).
+
+1. Check Output Topic for Word Counts
+
+## Get fresh token
+
+```bash
+TOKEN=$(curl -s -X POST http://localhost:3000/users/login \
+ -H "Content-Type: application/json" \
+ -d '{"username":"iggy","password":"iggy"}' | \
+ grep -o '"token":"[^"]*"' | cut -d'"' -f4)
+```
+
+## Check word-counts stream status
+
+```bash
+curl -s "http://localhost:3000/streams/word-counts" \
+ -H "Authorization: Bearer $TOKEN" | python3 -m json.tool
+```
+
+Look for messages_count - it should show the number of word count results.
+
+1. Monitor via Web UIs
+
+ - Flink Web UI: <http://localhost:8081>
+ - View job details, metrics, task managers
+ - Check "Records Sent" and "Records Received" counters
+ - Iggy Web UI: <http://localhost:3000>
+ - Monitor stream/topic stats
+ - View consumer groups
+
+1. Useful Monitoring Commands
+
+ Check job is still running:
+
+ ```bash
+ docker exec flink-jobmanager flink list
+ ```
+
+ View JobManager logs:
+
+ ```bash
+ docker logs flink-jobmanager --tail 100
+ ```
+
+ View TaskManager logs:
+
+ ```bash
+ docker logs flink-taskmanager --tail 100
+ ```
+
+ Check Iggy server logs:
+
+ ```bash
+ docker logs iggy-server --tail 100
+ ```
+
+ Verify no infinite message loop:
+
+## Wait 10 seconds and check record counts haven't exploded
+
+```bash
+sleep 10
+docker logs flink-taskmanager --tail 20 | grep "messagesCount"
+```
+
+The messagesCount should be reasonable (0 when no new messages, or small numbers matching what you sent).
+
+1. Stop/Cancel Job (if needed)
+
+## List jobs to get JobID
+
+```bash
+docker exec flink-jobmanager flink list
+```
+
+## Cancel job
+
+```bash
+docker exec flink-jobmanager flink cancel <JOB_ID>
+```
+
+1. Clean Up (when done)
+
+ ```bash
+ cd $IGGY_HOME/foreign/java/external-processors/iggy-connector-flink
+ docker-compose down -v # -v removes volumes (full cleanup)
+ ```
+
+---
+
+## Quick Test Script
+
+Save this as test-word-count.sh:
+
+```bash
+#!/bin/bash
+set -e
+
+echo "=== WordCount Job Test ==="
+echo ""
+```
+
+## Get token
+
+```bash
+echo "1. Authenticating..."
+TOKEN=$(curl -s -X POST http://localhost:3000/users/login \
+ -H "Content-Type: application/json" \
+ -d '{"username":"iggy","password":"iggy"}' | \
+ grep -o '"token":"[^"]*"' | cut -d'"' -f4)
+echo "✓ Authenticated"
+echo ""
+```
+
+## Send messages
+
+```bash
+echo "2. Sending test messages..."
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"aGVsbG8gd29ybGQgaGVsbG8gZmxpbms="}]}' > /dev/null
+echo "✓ Message 1 sent"
+
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"YXBhY2hlIGZsaW5rIGNvbm5lY3RvciBmb3IgaWdneQ=="}]}' > /dev/null
+echo "✓ Message 2 sent"
+
+curl -s -X POST "http://localhost:3000/streams/3/topics/1/messages" \
+ -H "Content-Type: application/json" \
+ -H "Authorization: Bearer $TOKEN" \
+ -d '{"partitioning":{"kind":"balanced","value":""},"messages":[{"payload":"c3RyZWFtaW5nIGRhdGEgcHJvY2Vzc2luZyB3aXRoIGZsaW5r"}]}' > /dev/null
+echo "✓ Message 3 sent"
+echo ""
+```
+
+## Wait and check
+
+```bash
+echo "3. Waiting for processing (10s)..."
+sleep 10
+echo ""
+
+echo "4. Checking word counts output..."
+WORD_COUNTS=$(curl -s "http://localhost:3000/streams/word-counts" \
+ -H "Authorization: Bearer $TOKEN" | python3 -c "import json,sys; data=json.load(sys.stdin); print(f\"Messages: {data['messages_count']}, Size: {data['size']}\")")
+echo "Output topic: $WORD_COUNTS"
+echo ""
+
+echo "5. Recent processing logs:"
+docker logs flink-taskmanager --since 30s 2>&1 | grep "Deserialized message" | tail -5
+echo ""
+
+echo "✓ Test complete!"
+```
+
+Make it executable and run:
+
+```bash
+chmod +x test-word-count.sh
+./test-word-count.sh
+```
+
+<!-- markdownlint-enable MD034 -->
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/MultiStreamJoinJob.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/MultiStreamJoinJob.java
new file mode 100644
index 000000000..5eabe96d9
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/MultiStreamJoinJob.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
+import org.apache.flink.util.Collector;
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.flink.sink.IggySink;
+import org.apache.iggy.connector.flink.source.IggySource;
+import org.apache.iggy.connector.serialization.JsonDeserializationSchema;
+import org.apache.iggy.connector.serialization.JsonSerializationSchema;
+import org.apache.iggy.flink.example.model.EnrichedActivity;
+import org.apache.iggy.flink.example.model.UserActivity;
+import org.apache.iggy.flink.example.model.UserProfile;
+
+import java.time.Duration;
+
+/**
+ * Example demonstrating multi-stream join using Iggy connector.
+ *
+ * <p>This job:
+ * <ul>
+ * <li>Reads user activities from Iggy stream "user-activities" topic "events"</li>
+ * <li>Reads user profiles from Iggy stream "user-profiles" topic "updates"</li>
+ * <li>Enriches activities with profile information using stateful join</li>
+ * <li>Writes enriched activities to Iggy stream "enriched-activities" topic "results"</li>
+ * </ul>
+ *
+ * <p>The join maintains user profile state and enriches incoming activities with the
+ * latest profile information. If a profile is not yet available for a user, the activity
+ * is buffered until the profile arrives.
+ *
+ * <p>Usage:
+ * <pre>
+ * # Run locally
+ * ./gradlew :iggy-flink-examples:run -PmainClass=org.apache.iggy.flink.example.MultiStreamJoinJob
+ *
+ * # Submit to Flink cluster
+ * flink run -c org.apache.iggy.flink.example.MultiStreamJoinJob build/libs/flink-iggy-examples.jar
+ *
+ * # Send sample data (requires iggy CLI)
+ * # Send user profile
+ * echo '{"userId":"user123","userName":"Alice","userTier":"premium","country":"US"}' \
+ * | iggy stream send user-profiles updates
+ *
+ * # Send user activity
+ * echo '{"userId":"user123","activityType":"VIEW","resourceId":"article-456","timestamp":"2025-10-22T10:30:00Z"}' \
+ * | iggy stream send user-activities events
+ * </pre>
+ */
+public class MultiStreamJoinJob {
+
+ private static final String IGGY_SERVER = getEnv("IGGY_SERVER", "localhost:8090");
+ private static final String IGGY_USERNAME = getEnv("IGGY_USERNAME", "iggy");
+ private static final String IGGY_PASSWORD = getEnv("IGGY_PASSWORD", "iggy");
+
+ // Use numeric IDs to match docker-compose.yml setup
+ private static final String ACTIVITY_STREAM = "5"; // user-activities
+ private static final String ACTIVITY_TOPIC = "1"; // events
+ private static final String PROFILE_STREAM = "6"; // user-profiles
+ private static final String PROFILE_TOPIC = "1"; // updates
+ private static final String OUTPUT_STREAM = "7"; // enriched-activities
+ private static final String OUTPUT_TOPIC = "1"; // results
+
+ public static void main(String[] args) throws Exception {
+ // Set up Flink execution environment
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ // Enable checkpointing for fault tolerance
+ env.enableCheckpointing(60000); // Checkpoint every 60 seconds
+
+ // Create Iggy connection configuration
+ IggyConnectionConfig connectionConfig = IggyConnectionConfig.builder()
+ .serverAddress(IGGY_SERVER)
+ .username(IGGY_USERNAME)
+ .password(IGGY_PASSWORD)
+ .connectionTimeout(Duration.ofSeconds(30))
+ .build();
+
+ // Read user activities from Iggy
+ DataStream<UserActivity> activities = env.fromSource(
+ IggySource.<UserActivity>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(ACTIVITY_STREAM)
+ .setTopicId(ACTIVITY_TOPIC)
+ .setConsumerGroup("flink-activity-enricher")
+ .setDeserializer(new JsonDeserializationSchema<>(UserActivity.class))
+ .setPollBatchSize(100)
+ .build(),
+ WatermarkStrategy
+ .<UserActivity>forBoundedOutOfOrderness(Duration.ofSeconds(10))
+ .withTimestampAssigner((activity, timestamp) ->
+ activity.getTimestamp().toEpochMilli()),
+ "Iggy Activity Source")
+ .returns(UserActivity.class);
+
+ // Read user profiles from Iggy
+ DataStream<UserProfile> profiles = env.fromSource(
+ IggySource.<UserProfile>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(PROFILE_STREAM)
+ .setTopicId(PROFILE_TOPIC)
+ .setConsumerGroup("flink-profile-reader")
+ .setDeserializer(new JsonDeserializationSchema<>(UserProfile.class))
+ .setPollBatchSize(50)
+ .build(),
+ WatermarkStrategy.noWatermarks(),
+ "Iggy Profile Source")
+ .returns(UserProfile.class);
+
+ // Enrich activities with profile information using stateful join
+ DataStream<EnrichedActivity> enrichedActivities = activities
+ .keyBy(UserActivity::getUserId)
+ .connect(profiles.keyBy(UserProfile::getUserId))
+ .process(new ProfileEnrichmentFunction())
+ .name("Enrich Activities");
+
+ // Write enriched activities back to Iggy
+ enrichedActivities.sinkTo(
+ IggySink.<EnrichedActivity>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(OUTPUT_STREAM)
+ .setTopicId(OUTPUT_TOPIC)
+ .setSerializer(new JsonSerializationSchema<>())
+ .setBatchSize(50)
+ .setFlushInterval(Duration.ofSeconds(5))
+ .withBalancedPartitioning() // Use balanced partitioning instead
+ .build())
+ .name("Iggy Enriched Activity Sink");
+
+ // Execute the job
+ env.execute("Multi-Stream Join with Iggy");
+ }
+
+ /**
+ * Co-process function that enriches activities with profile information.
+ * Maintains profile state and buffers activities if profile is not yet available.
+ */
+ public static class ProfileEnrichmentFunction
+ extends KeyedCoProcessFunction<String, UserActivity, UserProfile, EnrichedActivity> {
+
+ private transient ValueState<UserProfile> profileState;
+ private transient ValueState<UserActivity> pendingActivityState;
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {
+ super.open(openContext);
+
+ // State to store the latest user profile
+ ValueStateDescriptor<UserProfile> profileDescriptor =
+ new ValueStateDescriptor<>("user-profile", UserProfile.class);
+ profileState = getRuntimeContext().getState(profileDescriptor);
+
+ // State to buffer activities waiting for profile
+ ValueStateDescriptor<UserActivity> activityDescriptor =
+ new ValueStateDescriptor<>("pending-activity", UserActivity.class);
+ pendingActivityState = getRuntimeContext().getState(activityDescriptor);
+ }
+
+ @Override
+ public void processElement1(
+ UserActivity activity,
+ Context context,
+ Collector<EnrichedActivity> out) throws Exception {
+ // Process activity event
+ System.out.println(">>> processElement1 - Activity: userId=" + activity.getUserId() + ", type=" + activity.getActivityType());
+ UserProfile profile = profileState.value();
+
+ if (profile != null) {
+ // Profile is available, enrich and emit
+ System.out.println(">>> MATCH! Profile found for userId=" + activity.getUserId() + ", emitting enriched activity");
+ out.collect(EnrichedActivity.from(activity, profile));
+ } else {
+ // Profile not yet available, buffer the activity
+ System.out.println(">>> NO MATCH - Buffering activity for userId=" + activity.getUserId());
+ pendingActivityState.update(activity);
+ }
+ }
+
+ @Override
+ public void processElement2(
+ UserProfile profile,
+ Context context,
+ Collector<EnrichedActivity> out) throws Exception {
+ // Process profile update
+ System.out.println(">>> processElement2 - Profile: userId=" + profile.getUserId() + ", name=" + profile.getUserName());
+ profileState.update(profile);
+
+ // Check if there's a pending activity for this user
+ UserActivity pendingActivity = pendingActivityState.value();
+ if (pendingActivity != null) {
+ // Enrich and emit the pending activity
+ System.out.println(">>> FOUND PENDING! Emitting buffered activity for userId=" + profile.getUserId());
+ out.collect(EnrichedActivity.from(pendingActivity, profile));
+ // Clear the pending activity
+ pendingActivityState.clear();
+ } else {
+ System.out.println(">>> No pending activity for userId=" + profile.getUserId());
+ }
+ }
+ }
+
+ private static String getEnv(String key, String defaultValue) {
+ String value = System.getenv(key);
+ return value != null ? value : defaultValue;
+ }
+
+ /**
+ * Serializable partition key extractor for enriched activities.
+ */
+ private static class EnrichedActivityPartitionKeyExtractor
+ implements java.util.function.Function<EnrichedActivity, Integer>, java.io.Serializable {
+ @Override
+ public Integer apply(EnrichedActivity enriched) {
+ return enriched.getUserId().hashCode();
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/SendTestMessages.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/SendTestMessages.java
new file mode 100644
index 000000000..cc0eb7e9f
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/SendTestMessages.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.Partitioning;
+
+import java.util.List;
+
+/**
+ * Simple utility to send test messages to Iggy for WordCountJob testing.
+ */
+public class SendTestMessages {
+ public static void main(String[] args) throws Exception {
+ String serverAddress = System.getenv().getOrDefault("IGGY_SERVER", "localhost:3000");
+ String username = System.getenv().getOrDefault("IGGY_USERNAME", "iggy");
+ String password = System.getenv().getOrDefault("IGGY_PASSWORD", "iggy");
+
+ // Create HTTP client
+ IggyHttpClient client = new IggyHttpClient("http://" + serverAddress);
+
+ System.out.println("Connecting to Iggy at " + serverAddress);
+
+ // Login
+ client.users().login(username, password);
+ System.out.println("Logged in successfully!");
+
+ // Test messages
+ String[] messages = {
+ "hello world hello flink",
+ "apache flink connector for iggy",
+ "streaming data processing with flink",
+ "hello iggy hello streaming",
+ "real time analytics with apache flink",
+ "message broker and stream processing"
+ };
+
+ // Send messages using stream/topic names
+ System.out.println("\nSending messages to text-input/lines...");
+ for (String text : messages) {
+ client.messages().sendMessages(
+ StreamId.of("text-input"),
+ TopicId.of("lines"),
+ Partitioning.balanced(),
+ List.of(Message.of(text))
+ );
+ System.out.println("✓ Sent: " + text);
+ }
+
+ System.out.println("\nDone! Sent " + messages.length + " messages to text-input/lines.");
+ System.out.println("Check Flink job logs to see word counts being processed.");
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/StreamTransformJob.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/StreamTransformJob.java
new file mode 100644
index 000000000..f8a692c9a
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/StreamTransformJob.java
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.flink.sink.IggySink;
+import org.apache.iggy.connector.flink.source.IggySource;
+import org.apache.iggy.connector.serialization.JsonDeserializationSchema;
+import org.apache.iggy.connector.serialization.JsonSerializationSchema;
+import org.apache.iggy.flink.example.model.Alert;
+import org.apache.iggy.flink.example.model.SensorReading;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+
+/**
+ * Example Flink job demonstrating stream-to-stream transformation using Iggy connector.
+ *
+ * <p>This job:
+ * <ul>
+ * <li>Reads sensor readings from Iggy stream "sensors" topic "readings"</li>
+ * <li>Detects anomalies (temperature > 30°C or humidity > 80%)</li>
+ * <li>Generates alerts for critical conditions</li>
+ * <li>Writes alerts back to Iggy stream "alerts" topic "critical"</li>
+ * </ul>
+ *
+ * <p>Usage:
+ * <pre>
+ * # Run locally
+ * ./gradlew :iggy-flink-examples:run
+ *
+ * # Submit to Flink cluster
+ * flink run build/libs/flink-iggy-examples.jar
+ * </pre>
+ */
+public class StreamTransformJob {
+
+ private static final String IGGY_SERVER = getEnv("IGGY_SERVER", "localhost:8090");
+ private static final String IGGY_USERNAME = getEnv("IGGY_USERNAME", "iggy");
+ private static final String IGGY_PASSWORD = getEnv("IGGY_PASSWORD", "iggy");
+
+ // Stream and topic IDs (numeric)
+ private static final String INPUT_STREAM = "1"; // sensors
+ private static final String INPUT_TOPIC = "1"; // readings
+ private static final String OUTPUT_STREAM = "2"; // alerts
+ private static final String OUTPUT_TOPIC = "1"; // critical
+
+ private static final double TEMP_THRESHOLD = 30.0;
+ private static final double HUMIDITY_THRESHOLD = 80.0;
+
+ public static void main(String[] args) throws Exception {
+ // Set up Flink execution environment
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ // Enable checkpointing for fault tolerance
+ env.enableCheckpointing(60000); // Checkpoint every 60 seconds
+
+ // Create Iggy connection configuration
+ IggyConnectionConfig connectionConfig = IggyConnectionConfig.builder()
+ .serverAddress(IGGY_SERVER)
+ .username(IGGY_USERNAME)
+ .password(IGGY_PASSWORD)
+ .connectionTimeout(Duration.ofSeconds(30))
+ .build();
+
+ // Read sensor readings from Iggy
+ DataStream<SensorReading> sensorReadings = env.fromSource(
+ IggySource.<SensorReading>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(INPUT_STREAM)
+ .setTopicId(INPUT_TOPIC)
+ .setConsumerGroup("flink-anomaly-detector")
+ .setDeserializer(new JsonDeserializationSchema<>(SensorReading.class))
+ .setPollBatchSize(100)
+ .build(),
+ WatermarkStrategy
+ .<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(10))
+ .withTimestampAssigner(new TimestampExtractor()),
+ "Iggy Sensor Source",
+ TypeInformation.of(SensorReading.class));
+
+ // Process sensor readings and generate alerts
+ DataStream<Alert> alerts = sensorReadings
+ .filter(new AnomalyFilter())
+ .map(new AlertMapper())
+ .name("Generate Alerts");
+
+ // Optional: Add windowed aggregation to count alerts per sensor
+ DataStream<Alert> windowedAlerts = alerts
+ .keyBy(Alert::getSensorId)
+ .window(TumblingEventTimeWindows.of(Duration.ofMinutes(5)))
+ .reduce(new AlertReducer())
+ .name("Window Alerts");
+
+ // Write alerts back to Iggy
+ windowedAlerts.sinkTo(
+ IggySink.<Alert>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(OUTPUT_STREAM)
+ .setTopicId(OUTPUT_TOPIC)
+ .setSerializer(new JsonSerializationSchema<>())
+ .setBatchSize(50)
+ .setFlushInterval(Duration.ofSeconds(5))
+ .withBalancedPartitioning() // Use balanced partitioning instead
+ .build())
+ .name("Iggy Alert Sink");
+
+ // Execute the job
+ env.execute("Sensor Anomaly Detection with Iggy");
+ }
+
+ private static String getEnv(String key, String defaultValue) {
+ String value = System.getenv(key);
+ return value != null ? value : defaultValue;
+ }
+
+ // Static serializable function classes
+ private static class AnomalyFilter implements FilterFunction<SensorReading> {
+ private static final Logger logger = LoggerFactory.getLogger(AnomalyFilter.class);
+
+ @Override
+ public boolean filter(SensorReading reading) {
+ boolean isAnomaly = reading.getTemperature() > TEMP_THRESHOLD
+ || reading.getHumidity() > HUMIDITY_THRESHOLD;
+ logger.info("AnomalyFilter: reading={}, temp={}, humidity={}, isAnomaly={}",
+ reading.getSensorId(), reading.getTemperature(), reading.getHumidity(), isAnomaly);
+ return isAnomaly;
+ }
+ }
+
+ private static class AlertMapper implements MapFunction<SensorReading, Alert> {
+ private static final Logger logger = LoggerFactory.getLogger(AlertMapper.class);
+
+ @Override
+ public Alert map(SensorReading reading) {
+ String alertType;
+ double value;
+ String message;
+
+ if (reading.getTemperature() > TEMP_THRESHOLD) {
+ alertType = "HIGH_TEMPERATURE";
+ value = reading.getTemperature();
+ message = String.format(
+ "Temperature %.1f°C exceeds threshold %.1f°C",
+ reading.getTemperature(), TEMP_THRESHOLD);
+ } else {
+ alertType = "HIGH_HUMIDITY";
+ value = reading.getHumidity();
+ message = String.format(
+ "Humidity %.1f%% exceeds threshold %.1f%%",
+ reading.getHumidity(), HUMIDITY_THRESHOLD);
+ }
+
+ Alert alert = new Alert(
+ reading.getSensorId(),
+ alertType,
+ value,
+ message,
+ reading.getTimestamp());
+ logger.info("AlertMapper: created alert={}, type={}, sensor={}",
+ alert, alertType, reading.getSensorId());
+ return alert;
+ }
+ }
+
+ private static class AlertReducer implements ReduceFunction<Alert> {
+ private static final Logger logger = LoggerFactory.getLogger(AlertReducer.class);
+
+ @Override
+ public Alert reduce(Alert alert1, Alert alert2) {
+ // Keep the latest alert in the window
+ Alert result = alert1.getTimestamp().isAfter(alert2.getTimestamp())
+ ? alert1
+ : alert2;
+ logger.info("AlertReducer: reducing alert1={} and alert2={}, result={}",
+ alert1.getSensorId(), alert2.getSensorId(), result.getSensorId());
+ return result;
+ }
+ }
+
+ private static class TimestampExtractor implements SerializableTimestampAssigner<SensorReading> {
+ @Override
+ public long extractTimestamp(SensorReading reading, long recordTimestamp) {
+ return reading.getTimestamp().toEpochMilli();
+ }
+ }
+
+ private static class PartitionKeyExtractor implements java.util.function.Function<Alert, Integer>, java.io.Serializable {
+ @Override
+ public Integer apply(Alert alert) {
+ return alert.getSensorId().hashCode();
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/WordCountJob.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/WordCountJob.java
new file mode 100644
index 000000000..7a6efa4e9
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/WordCountJob.java
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
+import org.apache.flink.util.Collector;
+import org.apache.iggy.connector.config.IggyConnectionConfig;
+import org.apache.iggy.connector.flink.sink.IggySink;
+import org.apache.iggy.connector.flink.source.IggySource;
+import org.apache.iggy.connector.serialization.JsonSerializationSchema;
+import org.apache.iggy.connector.serialization.StringDeserializationSchema;
+import org.apache.iggy.flink.example.model.WordCount;
+
+import java.time.Duration;
+
+/**
+ * Classic word count example demonstrating Iggy connector usage.
+ *
+ * <p>This job:
+ * <ul>
+ * <li>Reads text lines from Iggy stream "text-input" topic "lines"</li>
+ * <li>Tokenizes lines into words</li>
+ * <li>Counts word occurrences in 1-minute tumbling windows</li>
+ * <li>Writes word counts to Iggy stream "word-counts" topic "results"</li>
+ * </ul>
+ *
+ * <p>Usage:
+ * <pre>
+ * # Run locally
+ * ./gradlew :iggy-flink-examples:run -PmainClass=org.apache.iggy.flink.example.WordCountJob
+ *
+ * # Submit to Flink cluster
+ * flink run -c org.apache.iggy.flink.example.WordCountJob build/libs/flink-iggy-examples.jar
+ *
+ * # Send sample data (requires iggy CLI)
+ * echo "hello world hello flink" | iggy stream send text-input lines
+ * echo "apache flink connector for iggy" | iggy stream send text-input lines
+ * </pre>
+ */
+public class WordCountJob {
+
+ private static final String IGGY_SERVER = getEnv("IGGY_SERVER", "localhost:8090");
+ private static final String IGGY_USERNAME = getEnv("IGGY_USERNAME", "iggy");
+ private static final String IGGY_PASSWORD = getEnv("IGGY_PASSWORD", "iggy");
+
+ // Use stream/topic names (async TCP client requires names, not IDs)
+ private static final String INPUT_STREAM = "text-input";
+ private static final String INPUT_TOPIC = "lines";
+ private static final String OUTPUT_STREAM = "word-counts";
+ private static final String OUTPUT_TOPIC = "results";
+
+ public static void main(String[] args) throws Exception {
+ // Set up Flink execution environment
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment();
+
+ // Enable checkpointing for fault tolerance
+ env.enableCheckpointing(30000); // Checkpoint every 30 seconds
+
+ // Create Iggy connection configuration
+ IggyConnectionConfig connectionConfig = IggyConnectionConfig.builder()
+ .serverAddress(IGGY_SERVER)
+ .username(IGGY_USERNAME)
+ .password(IGGY_PASSWORD)
+ .connectionTimeout(Duration.ofSeconds(30))
+ .build();
+
+ // Read text lines from Iggy
+ DataStream<String> textLines = env.fromSource(
+ IggySource.<String>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(INPUT_STREAM)
+ .setTopicId(INPUT_TOPIC)
+ .setConsumerGroup("flink-word-counter")
+ .setDeserializer(new StringDeserializationSchema())
+ .setPollBatchSize(100)
+ .build(),
+ WatermarkStrategy.noWatermarks(),
+ "Iggy Text Source")
+ .returns(String.class);
+
+ // Tokenize, count, and aggregate words
+ DataStream<WordCount> wordCounts = textLines
+ .flatMap(new Tokenizer())
+ .returns(org.apache.flink.api.common.typeinfo.Types.TUPLE(
+ org.apache.flink.api.common.typeinfo.Types.STRING,
+ org.apache.flink.api.common.typeinfo.Types.LONG))
+ .name("Tokenize")
+ .keyBy(word -> word.f0)
+ .window(TumblingProcessingTimeWindows.of(Duration.ofMinutes(1)))
+ .sum(1)
+ .name("Count Words")
+ .map(tuple -> new WordCount(tuple.f0, tuple.f1))
+ .name("Create WordCount");
+
+ // Write word counts back to Iggy
+ wordCounts.sinkTo(
+ IggySink.<WordCount>builder()
+ .setConnectionConfig(connectionConfig)
+ .setStreamId(OUTPUT_STREAM)
+ .setTopicId(OUTPUT_TOPIC)
+ .setSerializer(new JsonSerializationSchema<>())
+ .setBatchSize(50)
+ .setFlushInterval(Duration.ofSeconds(5))
+ .withBalancedPartitioning() // Use balanced partitioning instead
+ .build())
+ .name("Iggy WordCount Sink");
+
+ // Execute the job
+ env.execute("Word Count with Iggy");
+ }
+
+ /**
+ * Tokenizer that splits lines into words.
+ */
+ public static final class Tokenizer
+ implements FlatMapFunction<String, org.apache.flink.api.java.tuple.Tuple2<String, Long>> {
+
+ @Override
+ public void flatMap(
+ String line,
+ Collector<org.apache.flink.api.java.tuple.Tuple2<String, Long>> out) {
+ if (line == null || line.trim().isEmpty()) {
+ return;
+ }
+
+ // Split line into words and emit each word with count 1
+ String[] words = line.toLowerCase()
+ .replaceAll("[^a-z0-9\\s]", "") // Remove punctuation
+ .trim()
+ .split("\\s+");
+
+ for (String word : words) {
+ if (!word.isEmpty()) {
+ out.collect(new org.apache.flink.api.java.tuple.Tuple2<>(word, 1L));
+ }
+ }
+ }
+ }
+
+ private static String getEnv(String key, String defaultValue) {
+ String value = System.getenv(key);
+ return value != null ? value : defaultValue;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/Alert.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/Alert.java
new file mode 100644
index 000000000..dd6cefdf3
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/Alert.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * Alert event generated from sensor readings.
+ */
+public class Alert implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String sensorId;
+ private final String alertType;
+ private final double value;
+ private final String message;
+ private final Instant timestamp;
+
+ @JsonCreator
+ public Alert(
+ @JsonProperty("sensorId") String sensorId,
+ @JsonProperty("alertType") String alertType,
+ @JsonProperty("value") double value,
+ @JsonProperty("message") String message,
+ @JsonProperty("timestamp") Instant timestamp) {
+ this.sensorId = sensorId;
+ this.alertType = alertType;
+ this.value = value;
+ this.message = message;
+ this.timestamp = timestamp != null ? timestamp : Instant.now();
+ }
+
+ public String getSensorId() {
+ return sensorId;
+ }
+
+ public String getAlertType() {
+ return alertType;
+ }
+
+ public double getValue() {
+ return value;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Alert alert = (Alert) o;
+ return Double.compare(alert.value, value) == 0
+ && Objects.equals(sensorId, alert.sensorId)
+ && Objects.equals(alertType, alert.alertType)
+ && Objects.equals(message, alert.message)
+ && Objects.equals(timestamp, alert.timestamp);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(sensorId, alertType, value, message, timestamp);
+ }
+
+ @Override
+ public String toString() {
+ return "Alert{"
+ + "sensorId='" + sensorId + '\''
+ + ", alertType='" + alertType + '\''
+ + ", value=" + value
+ + ", message='" + message + '\''
+ + ", timestamp=" + timestamp
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/EnrichedActivity.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/EnrichedActivity.java
new file mode 100644
index 000000000..69bc570d2
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/EnrichedActivity.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * User activity enriched with profile information.
+ */
+public class EnrichedActivity implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String userId;
+ private final String userName;
+ private final String userTier;
+ private final String country;
+ private final String activityType;
+ private final String resourceId;
+ private final Instant timestamp;
+
+ @JsonCreator
+ public EnrichedActivity(
+ @JsonProperty("userId") String userId,
+ @JsonProperty("userName") String userName,
+ @JsonProperty("userTier") String userTier,
+ @JsonProperty("country") String country,
+ @JsonProperty("activityType") String activityType,
+ @JsonProperty("resourceId") String resourceId,
+ @JsonProperty("timestamp") Instant timestamp) {
+ this.userId = userId;
+ this.userName = userName;
+ this.userTier = userTier;
+ this.country = country;
+ this.activityType = activityType;
+ this.resourceId = resourceId;
+ this.timestamp = timestamp;
+ }
+
+ public static EnrichedActivity from(UserActivity activity, UserProfile profile) {
+ return new EnrichedActivity(
+ activity.getUserId(),
+ profile.getUserName(),
+ profile.getUserTier(),
+ profile.getCountry(),
+ activity.getActivityType(),
+ activity.getResourceId(),
+ activity.getTimestamp());
+ }
+
+ public String getUserId() {
+ return userId;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public String getUserTier() {
+ return userTier;
+ }
+
+ public String getCountry() {
+ return country;
+ }
+
+ public String getActivityType() {
+ return activityType;
+ }
+
+ public String getResourceId() {
+ return resourceId;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ EnrichedActivity that = (EnrichedActivity) o;
+ return Objects.equals(userId, that.userId)
+ && Objects.equals(userName, that.userName)
+ && Objects.equals(userTier, that.userTier)
+ && Objects.equals(country, that.country)
+ && Objects.equals(activityType, that.activityType)
+ && Objects.equals(resourceId, that.resourceId)
+ && Objects.equals(timestamp, that.timestamp);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(userId, userName, userTier, country,
+ activityType, resourceId, timestamp);
+ }
+
+ @Override
+ public String toString() {
+ return "EnrichedActivity{"
+ + "userId='" + userId + '\''
+ + ", userName='" + userName + '\''
+ + ", userTier='" + userTier + '\''
+ + ", country='" + country + '\''
+ + ", activityType='" + activityType + '\''
+ + ", resourceId='" + resourceId + '\''
+ + ", timestamp=" + timestamp
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/SensorReading.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/SensorReading.java
new file mode 100644
index 000000000..932b5b4ad
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/SensorReading.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * Sensor reading event for stream processing examples.
+ */
+public class SensorReading implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String sensorId;
+ private final double temperature;
+ private final double humidity;
+ private final Instant timestamp;
+
+ @JsonCreator
+ public SensorReading(
+ @JsonProperty("sensorId") String sensorId,
+ @JsonProperty("temperature") double temperature,
+ @JsonProperty("humidity") double humidity,
+ @JsonProperty("timestamp") Instant timestamp) {
+ this.sensorId = sensorId;
+ this.temperature = temperature;
+ this.humidity = humidity;
+ this.timestamp = timestamp != null ? timestamp : Instant.now();
+ }
+
+ public String getSensorId() {
+ return sensorId;
+ }
+
+ public double getTemperature() {
+ return temperature;
+ }
+
+ public double getHumidity() {
+ return humidity;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SensorReading that = (SensorReading) o;
+ return Double.compare(that.temperature, temperature) == 0
+ && Double.compare(that.humidity, humidity) == 0
+ && Objects.equals(sensorId, that.sensorId)
+ && Objects.equals(timestamp, that.timestamp);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(sensorId, temperature, humidity, timestamp);
+ }
+
+ @Override
+ public String toString() {
+ return "SensorReading{"
+ + "sensorId='" + sensorId + '\''
+ + ", temperature=" + temperature
+ + ", humidity=" + humidity
+ + ", timestamp=" + timestamp
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/UserActivity.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/UserActivity.java
new file mode 100644
index 000000000..39bdb0d2e
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/UserActivity.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.Objects;
+
+/**
+ * User activity event (clicks, views, purchases, etc.).
+ */
+public class UserActivity implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String userId;
+ private final String activityType;
+ private final String resourceId;
+ private final Instant timestamp;
+
+ @JsonCreator
+ public UserActivity(
+ @JsonProperty("userId") String userId,
+ @JsonProperty("activityType") String activityType,
+ @JsonProperty("resourceId") String resourceId,
+ @JsonProperty("timestamp") Instant timestamp) {
+ this.userId = userId;
+ this.activityType = activityType;
+ this.resourceId = resourceId;
+ this.timestamp = timestamp != null ? timestamp : Instant.now();
+ }
+
+ public String getUserId() {
+ return userId;
+ }
+
+ public String getActivityType() {
+ return activityType;
+ }
+
+ public String getResourceId() {
+ return resourceId;
+ }
+
+ public Instant getTimestamp() {
+ return timestamp;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ UserActivity that = (UserActivity) o;
+ return Objects.equals(userId, that.userId)
+ && Objects.equals(activityType, that.activityType)
+ && Objects.equals(resourceId, that.resourceId)
+ && Objects.equals(timestamp, that.timestamp);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(userId, activityType, resourceId, timestamp);
+ }
+
+ @Override
+ public String toString() {
+ return "UserActivity{"
+ + "userId='" + userId + '\''
+ + ", activityType='" + activityType + '\''
+ + ", resourceId='" + resourceId + '\''
+ + ", timestamp=" + timestamp
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/UserProfile.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/UserProfile.java
new file mode 100644
index 000000000..e13bf85b4
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/UserProfile.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * User profile information.
+ */
+public class UserProfile implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String userId;
+ private final String userName;
+ private final String userTier;
+ private final String country;
+
+ @JsonCreator
+ public UserProfile(
+ @JsonProperty("userId") String userId,
+ @JsonProperty("userName") String userName,
+ @JsonProperty("userTier") String userTier,
+ @JsonProperty("country") String country) {
+ this.userId = userId;
+ this.userName = userName;
+ this.userTier = userTier;
+ this.country = country;
+ }
+
+ public String getUserId() {
+ return userId;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public String getUserTier() {
+ return userTier;
+ }
+
+ public String getCountry() {
+ return country;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ UserProfile that = (UserProfile) o;
+ return Objects.equals(userId, that.userId)
+ && Objects.equals(userName, that.userName)
+ && Objects.equals(userTier, that.userTier)
+ && Objects.equals(country, that.country);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(userId, userName, userTier, country);
+ }
+
+ @Override
+ public String toString() {
+ return "UserProfile{"
+ + "userId='" + userId + '\''
+ + ", userName='" + userName + '\''
+ + ", userTier='" + userTier + '\''
+ + ", country='" + country + '\''
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/WordCount.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/WordCount.java
new file mode 100644
index 000000000..2e0a0f91f
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/main/java/org/apache/iggy/flink/example/model/WordCount.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example.model;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Word count result for word count example.
+ */
+public class WordCount implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String word;
+ private final long count;
+
+ @JsonCreator
+ public WordCount(
+ @JsonProperty("word") String word,
+ @JsonProperty("count") long count) {
+ this.word = word;
+ this.count = count;
+ }
+
+ public String getWord() {
+ return word;
+ }
+
+ public long getCount() {
+ return count;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ WordCount wordCount = (WordCount) o;
+ return count == wordCount.count && Objects.equals(word, wordCount.word);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(word, count);
+ }
+
+ @Override
+ public String toString() {
+ return "WordCount{word='" + word + "', count=" + count + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/AsyncTcpMessagePollTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/AsyncTcpMessagePollTest.java
new file mode 100644
index 000000000..b5ca9b5fa
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/AsyncTcpMessagePollTest.java
@@ -0,0 +1,414 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.Partitioning;
+import org.apache.iggy.message.PolledMessages;
+import org.apache.iggy.message.PollingStrategy;
+import org.junit.jupiter.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test class demonstrating how to poll/read messages from Iggy using Async TCP Client.
+ *
+ * Prerequisites:
+ * - Iggy server running on localhost:8090
+ * - Stream 'text-input' exists
+ * - Topic 'lines' exists with at least 2 partitions
+ *
+ * Run docker-compose first:
+ * cd $IGGY_ROOT/foreign/java/external-processors/iggy-connector-flink
+ * docker-compose up -d
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+class AsyncTcpMessagePollTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncTcpMessagePollTest.class);
+
+ private static final String IGGY_SERVER_HOST = "localhost";
+ private static final int IGGY_SERVER_TCP_PORT = 8090;
+ private static final String USERNAME = "iggy";
+ private static final String PASSWORD = "iggy";
+
+ private AsyncIggyTcpClient client;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ LOG.info("Setting up Async TCP Client for polling...");
+
+ // Create client with credentials for auto-login
+ client = AsyncIggyTcpClient.builder()
+ .host(IGGY_SERVER_HOST)
+ .port(IGGY_SERVER_TCP_PORT)
+ .credentials(USERNAME, PASSWORD)
+ .build();
+
+ // Connect - this will also auto-login with the provided credentials
+ client.connect().join();
+ LOG.info("Connected to Iggy server at {}:{}", IGGY_SERVER_HOST, IGGY_SERVER_TCP_PORT);
+
+ // Send some test messages to ensure we have data to poll
+ seedTestMessages();
+ }
+
+ @AfterAll
+ void tearDown() {
+ if (client != null) {
+ try {
+ client.users().logoutAsync().join();
+ LOG.info("Logged out successfully");
+ } catch (Exception e) {
+ LOG.warn("Error during logout: {}", e.getMessage());
+ }
+
+ try {
+ client.close().join();
+ LOG.info("Disconnected from Iggy server");
+ } catch (Exception e) {
+ LOG.warn("Error during disconnect: {}", e.getMessage());
+ }
+ }
+ }
+
+ private void seedTestMessages() throws ExecutionException, InterruptedException {
+ LOG.info("Seeding test messages...");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ List<Message> messages = new ArrayList<>();
+ for (int i = 0; i < 20; i++) {
+ String content = "Test message #" + i + " at " + Instant.now();
+ messages.add(Message.of(content));
+ }
+
+ client.messages().sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages).get();
+ LOG.info("Seeded {} test messages", messages.size());
+ }
+
+ @Test
+ @Order(1)
+ @DisplayName("Poll messages from offset 0")
+ void testPollFromBeginning() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll messages from offset 0");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ // Create poll request starting from offset 0
+ PollingStrategy pollingStrategy = PollingStrategy.offset(BigInteger.ZERO);
+ Consumer consumer = Consumer.of(1L);
+
+ CompletableFuture<PolledMessages> pollFuture = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L), // partition 1
+ consumer,
+ pollingStrategy,
+ 10L, // count
+ false // autoCommit
+ );
+
+ PolledMessages polledMessages = pollFuture.get();
+
+ LOG.info("Polled {} messages from offset 0", polledMessages.messages().size());
+ assertThat(polledMessages).isNotNull();
+ assertThat(polledMessages.messages()).isNotEmpty();
+
+ // Display first message
+ if (!polledMessages.messages().isEmpty()) {
+ Message firstMessage = polledMessages.messages().get(0);
+ String payload = new String(firstMessage.payload(), StandardCharsets.UTF_8);
+ LOG.info("First message - Payload: {}", payload);
+ }
+ }
+
+ @Test
+ @Order(2)
+ @DisplayName("Poll messages using NEXT strategy")
+ void testPollNext() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll messages using NEXT strategy");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ // Use NEXT polling strategy to get messages after last consumed
+ PollingStrategy pollingStrategy = PollingStrategy.next();
+ Consumer consumer = Consumer.of(2L);
+
+ CompletableFuture<PolledMessages> pollFuture = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L), // partition 1
+ consumer,
+ pollingStrategy,
+ 5L, // count
+ true // autoCommit enabled
+ );
+
+ PolledMessages polledMessages = pollFuture.get();
+
+ LOG.info("Polled {} messages using NEXT strategy", polledMessages.messages().size());
+ assertThat(polledMessages).isNotNull();
+
+ for (Message msg : polledMessages.messages()) {
+ String payload = new String(msg.payload(), StandardCharsets.UTF_8);
+ LOG.info("Message - Payload: {}", payload);
+ }
+ }
+
+ @Test
+ @Order(3)
+ @DisplayName("Poll messages from all partitions")
+ void testPollAllPartitions() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll messages from all partitions");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ // Poll without specifying partition (all partitions)
+ PollingStrategy pollingStrategy = PollingStrategy.offset(BigInteger.ZERO);
+ Consumer consumer = Consumer.of(3L);
+
+ CompletableFuture<PolledMessages> pollFuture = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.empty(), // all partitions
+ consumer,
+ pollingStrategy,
+ 10L,
+ false
+ );
+
+ PolledMessages polledMessages = pollFuture.get();
+
+ LOG.info("Polled {} messages from all partitions", polledMessages.messages().size());
+ assertThat(polledMessages).isNotNull();
+ }
+
+ @Test
+ @Order(4)
+ @DisplayName("Poll messages from specific offset")
+ void testPollFromSpecificOffset() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll messages from specific offset");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ // Poll from a specific offset (e.g., offset 5)
+ long specificOffset = 5L;
+ PollingStrategy pollingStrategy = PollingStrategy.offset(BigInteger.valueOf(specificOffset));
+ Consumer consumer = Consumer.of(4L);
+
+ PolledMessages polledMessages = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L),
+ consumer,
+ pollingStrategy,
+ 3L,
+ false
+ ).get();
+
+ LOG.info("Polled {} messages from offset {}", polledMessages.messages().size(), specificOffset);
+
+ for (Message msg : polledMessages.messages()) {
+ String payload = new String(msg.payload(), StandardCharsets.UTF_8);
+ LOG.info("Payload: {}", payload);
+ }
+ }
+
+ @Test
+ @Order(5)
+ @DisplayName("Poll messages from multiple partitions in parallel")
+ void testPollFromMultiplePartitions() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll messages from multiple partitions in parallel");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ PollingStrategy strategy = PollingStrategy.offset(BigInteger.ZERO);
+ Consumer consumer = Consumer.of(5L);
+
+ // Poll from partition 1
+ CompletableFuture<PolledMessages> future1 = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L),
+ consumer,
+ strategy,
+ 5L,
+ false
+ );
+
+ // Poll from partition 2
+ CompletableFuture<PolledMessages> future2 = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(2L),
+ consumer,
+ strategy,
+ 5L,
+ false
+ );
+
+ // Wait for both
+ CompletableFuture.allOf(future1, future2).get();
+
+ PolledMessages messages1 = future1.get();
+ PolledMessages messages2 = future2.get();
+
+ LOG.info("Polled {} messages from partition 1", messages1.messages().size());
+ LOG.info("Polled {} messages from partition 2", messages2.messages().size());
+
+ assertThat(messages1).isNotNull();
+ assertThat(messages2).isNotNull();
+ }
+
+ @Test
+ @Order(6)
+ @DisplayName("Poll with FIRST strategy")
+ void testPollFirst() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll with FIRST strategy");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ // Use FIRST polling strategy to get earliest available messages
+ PollingStrategy pollingStrategy = PollingStrategy.first();
+ Consumer consumer = Consumer.of(6L);
+
+ PolledMessages polledMessages = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L),
+ consumer,
+ pollingStrategy,
+ 3L,
+ false
+ ).get();
+
+ LOG.info("Polled {} messages using FIRST strategy", polledMessages.messages().size());
+
+ if (!polledMessages.messages().isEmpty()) {
+ Message firstMessage = polledMessages.messages().get(0);
+ String payload = new String(firstMessage.payload(), StandardCharsets.UTF_8);
+ LOG.info("First available message - Payload: {}", payload);
+ }
+
+ assertThat(polledMessages).isNotNull();
+ }
+
+ @Test
+ @Order(7)
+ @DisplayName("Poll with LAST strategy")
+ void testPollLast() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Poll with LAST strategy");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ // Use LAST polling strategy to get most recent messages
+ PollingStrategy pollingStrategy = PollingStrategy.last();
+ Consumer consumer = Consumer.of(7L);
+
+ PolledMessages polledMessages = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L),
+ consumer,
+ pollingStrategy,
+ 3L,
+ false
+ ).get();
+
+ LOG.info("Polled {} messages using LAST strategy", polledMessages.messages().size());
+
+ if (!polledMessages.messages().isEmpty()) {
+ Message lastMessage = polledMessages.messages().get(polledMessages.messages().size() - 1);
+ String payload = new String(lastMessage.payload(), StandardCharsets.UTF_8);
+ LOG.info("Last message - Payload: {}", payload);
+ }
+
+ assertThat(polledMessages).isNotNull();
+ }
+
+ @Test
+ @Order(8)
+ @DisplayName("Continuous polling simulation")
+ void testContinuousPolling() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Continuous polling simulation");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ Consumer consumer = Consumer.of(8L);
+ int totalMessagesPolled = 0;
+ int maxIterations = 3;
+
+ for (int i = 0; i < maxIterations; i++) {
+ PollingStrategy pollingStrategy = PollingStrategy.next();
+
+ PolledMessages polledMessages = client.messages().pollMessagesAsync(
+ streamId,
+ topicId,
+ Optional.of(1L),
+ consumer,
+ pollingStrategy,
+ 5L,
+ true // autoCommit
+ ).get();
+
+ int messageCount = polledMessages.messages().size();
+ totalMessagesPolled += messageCount;
+
+ LOG.info("Iteration {}: Polled {} messages", i + 1, messageCount);
+
+ if (messageCount == 0) {
+ LOG.info("No more messages available, stopping");
+ break;
+ }
+
+ // Simulate processing time
+ Thread.sleep(100);
+ }
+
+ LOG.info("Total messages polled across all iterations: {}", totalMessagesPolled);
+ assertThat(totalMessagesPolled).isGreaterThanOrEqualTo(0);
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/AsyncTcpMessageSendTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/AsyncTcpMessageSendTest.java
new file mode 100644
index 000000000..f24650a96
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/AsyncTcpMessageSendTest.java
@@ -0,0 +1,308 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.Partitioning;
+import org.junit.jupiter.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test class demonstrating how to send messages to Iggy using Async TCP Client.
+ *
+ * Prerequisites:
+ * - Iggy server running on localhost:8090
+ * - Stream 'text-input' exists
+ * - Topic 'lines' exists
+ *
+ * Run docker-compose first:
+ * cd $IGGY_HOME/foreign/java/external-processors/iggy-connector-flink
+ * docker-compose up -d
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+class AsyncTcpMessageSendTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncTcpMessageSendTest.class);
+
+ private static final String IGGY_SERVER_HOST = "localhost";
+ private static final int IGGY_SERVER_TCP_PORT = 8090;
+ private static final String USERNAME = "iggy";
+ private static final String PASSWORD = "iggy";
+
+ private AsyncIggyTcpClient client;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ LOG.info("Setting up Async TCP Client...");
+
+ // Create client with credentials for auto-login
+ client = AsyncIggyTcpClient.builder()
+ .host(IGGY_SERVER_HOST)
+ .port(IGGY_SERVER_TCP_PORT)
+ .credentials(USERNAME, PASSWORD)
+ .build();
+
+ // Connect - this will also auto-login with the provided credentials
+ client.connect().join();
+ LOG.info("Connected to Iggy server at {}:{}", IGGY_SERVER_HOST, IGGY_SERVER_TCP_PORT);
+ }
+
+ @AfterAll
+ void tearDown() {
+ if (client != null) {
+ try {
+ client.users().logoutAsync().join();
+ LOG.info("Logged out successfully");
+ } catch (Exception e) {
+ LOG.warn("Error during logout: {}", e.getMessage());
+ }
+
+ try {
+ client.close().join();
+ LOG.info("Disconnected from Iggy server");
+ } catch (Exception e) {
+ LOG.warn("Error during disconnect: {}", e.getMessage());
+ }
+ }
+ }
+
+ @Test
+ @Order(1)
+ @DisplayName("Send single text message")
+ void testSendSingleMessage() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send single text message");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ String messageContent = "Hello from AsyncTcpClient at " + Instant.now();
+ Message message = Message.of(messageContent);
+
+ List<Message> messages = new ArrayList<>();
+ messages.add(message);
+
+ CompletableFuture<Void> sendFuture = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ messages
+ );
+
+ sendFuture.get();
+
+ LOG.info("Successfully sent message: {}", messageContent);
+ assertThat(sendFuture).isCompleted();
+ }
+
+ @Test
+ @Order(2)
+ @DisplayName("Send batch of messages")
+ void testSendBatchMessages() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send batch of messages");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ List<Message> messages = new ArrayList<>();
+ int batchSize = 10;
+
+ for (int i = 0; i < batchSize; i++) {
+ String content = String.format("Batch message #%d sent at %s", i, Instant.now());
+ messages.add(Message.of(content));
+ }
+
+ CompletableFuture<Void> sendFuture = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ messages
+ );
+ sendFuture.get();
+
+ LOG.info("Successfully sent batch of {} messages", batchSize);
+ assertThat(sendFuture).isCompleted();
+ }
+
+ @Test
+ @Order(3)
+ @DisplayName("Send message to specific partition")
+ void testSendToSpecificPartition() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send message to specific partition");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ Long targetPartition = 1L;
+ String messageContent = "Message to partition " + targetPartition;
+
+ Message message = Message.of(messageContent);
+ List<Message> messages = new ArrayList<>();
+ messages.add(message);
+
+ CompletableFuture<Void> sendFuture = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.partitionId(targetPartition),
+ messages
+ );
+ sendFuture.get();
+
+ LOG.info("Successfully sent message to partition {}", targetPartition);
+ assertThat(sendFuture).isCompleted();
+ }
+
+ @Test
+ @Order(4)
+ @DisplayName("Send messages using message key partitioning")
+ void testSendWithMessageKey() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send messages using message key partitioning");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ String messageContent = "Message with key partitioning";
+ Message message = Message.of(messageContent);
+
+ List<Message> messages = new ArrayList<>();
+ messages.add(message);
+
+ String messageKey = "test-key-123";
+ CompletableFuture<Void> sendFuture = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.messagesKey(messageKey),
+ messages
+ );
+ sendFuture.get();
+
+ LOG.info("Successfully sent message with key: {}", messageKey);
+ assertThat(sendFuture).isCompleted();
+ }
+
+ @Test
+ @Order(5)
+ @DisplayName("Send JSON-like messages")
+ void testSendJsonMessages() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send JSON messages");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ String jsonPayload = String.format(
+ "{\"sensor_id\":\"sensor-001\",\"temperature\":%.2f,\"timestamp\":\"%s\"}",
+ 23.5,
+ Instant.now()
+ );
+
+ Message message = Message.of(jsonPayload);
+ List<Message> messages = new ArrayList<>();
+ messages.add(message);
+
+ CompletableFuture<Void> sendFuture = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ messages
+ );
+ sendFuture.get();
+
+ LOG.info("Successfully sent JSON message: {}", jsonPayload);
+ assertThat(sendFuture).isCompleted();
+ }
+
+ @Test
+ @Order(6)
+ @DisplayName("Send multiple messages in parallel")
+ void testSendMultipleMessagesInParallel() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send multiple messages in parallel");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ int parallelRequests = 5;
+ List<CompletableFuture<Void>> futures = new ArrayList<>();
+
+ for (int i = 0; i < parallelRequests; i++) {
+ String content = "Parallel message #" + i;
+ Message message = Message.of(content);
+ List<Message> messages = new ArrayList<>();
+ messages.add(message);
+
+ CompletableFuture<Void> future = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ messages
+ );
+ futures.add(future);
+ }
+
+ CompletableFuture<Void> allFutures = CompletableFuture.allOf(
+ futures.toArray(new CompletableFuture[0])
+ );
+ allFutures.get();
+
+ LOG.info("Successfully sent {} messages in parallel", parallelRequests);
+ assertThat(allFutures).isCompleted();
+ }
+
+ @Test
+ @Order(7)
+ @DisplayName("Send large batch of messages")
+ void testSendLargeBatch() throws ExecutionException, InterruptedException {
+ LOG.info("Test: Send large batch of messages");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ List<Message> messages = new ArrayList<>();
+ int batchSize = 100;
+
+ for (int i = 0; i < batchSize; i++) {
+ messages.add(Message.of("Large batch message #" + i));
+ }
+
+ long startTime = System.currentTimeMillis();
+ CompletableFuture<Void> sendFuture = client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ messages
+ );
+ sendFuture.get();
+ long duration = System.currentTimeMillis() - startTime;
+
+ LOG.info("Successfully sent {} messages in {}ms", batchSize, duration);
+ assertThat(sendFuture).isCompleted();
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/SendTextDataTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/SendTextDataTest.java
new file mode 100644
index 000000000..313b9f14d
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-flink-examples/src/test/java/org/apache/iggy/flink/example/SendTextDataTest.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.flink.example;
+
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.Partitioning;
+import org.junit.jupiter.api.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Test to send sample text data to text-input stream for WordCountJob testing.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class SendTextDataTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SendTextDataTest.class);
+
+ private static final String IGGY_SERVER_HOST = "localhost";
+ private static final int IGGY_SERVER_TCP_PORT = 8090;
+ private static final String USERNAME = "iggy";
+ private static final String PASSWORD = "iggy";
+
+ private AsyncIggyTcpClient client;
+
+ @BeforeAll
+ void setUp() throws Exception {
+ LOG.info("Setting up Async TCP Client...");
+
+ client = AsyncIggyTcpClient.builder()
+ .host(IGGY_SERVER_HOST)
+ .port(IGGY_SERVER_TCP_PORT)
+ .credentials(USERNAME, PASSWORD)
+ .build();
+
+ client.connect().join();
+ LOG.info("Connected to Iggy server at {}:{}", IGGY_SERVER_HOST, IGGY_SERVER_TCP_PORT);
+ }
+
+ @AfterAll
+ void tearDown() {
+ if (client != null) {
+ try {
+ client.users().logoutAsync().join();
+ LOG.info("Logged out successfully");
+ } catch (Exception e) {
+ LOG.warn("Error during logout: {}", e.getMessage());
+ }
+
+ try {
+ client.close().join();
+ LOG.info("Disconnected from Iggy server");
+ } catch (Exception e) {
+ LOG.warn("Error during disconnect: {}", e.getMessage());
+ }
+ }
+ }
+
+ @Test
+ @DisplayName("Send sample text lines for word count")
+ void testSendSampleTextLines() throws Exception {
+ LOG.info("Sending sample text lines for WordCountJob testing...");
+
+ StreamId streamId = StreamId.of("text-input");
+ TopicId topicId = TopicId.of("lines");
+
+ List<String> sampleTexts = List.of(
+ "hello world hello flink",
+ "apache flink connector for iggy",
+ "flink streaming with iggy broker",
+ "hello flink hello world",
+ "real time stream processing",
+ "apache flink word count example",
+ "iggy message broker stream",
+ "flink apache iggy connector test",
+ "hello world stream processing",
+ "apache iggy flink integration"
+ );
+
+ List<Message> messages = new ArrayList<>();
+ for (String text : sampleTexts) {
+ messages.add(Message.of(text));
+ }
+
+ client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ messages
+ ).get();
+
+ LOG.info("Successfully sent {} text lines to stream 'text-input', topic 'lines'", messages.size());
+
+ // Send a few more batches to have continuous data
+ for (int batch = 1; batch <= 3; batch++) {
+ Thread.sleep(1000);
+
+ List<Message> batchMessages = new ArrayList<>();
+ for (String text : sampleTexts) {
+ batchMessages.add(Message.of(text + " batch" + batch));
+ }
+
+ client.messages().sendMessagesAsync(
+ streamId,
+ topicId,
+ Partitioning.balanced(),
+ batchMessages
+ ).get();
+
+ LOG.info("Sent batch {} with {} messages", batch, batchMessages.size());
+ }
+
+ LOG.info("Total messages sent: {}", sampleTexts.size() * 4);
+ }
+}
diff --git a/foreign/java/java-sdk/build.gradle.kts b/foreign/java/java-sdk/build.gradle.kts
index 9911e3e77..f7264dd97 100644
--- a/foreign/java/java-sdk/build.gradle.kts
+++ b/foreign/java/java-sdk/build.gradle.kts
@@ -33,6 +33,10 @@ repositories {
}
java {
+ // Keep Java 17 for SDK compatibility with broader ecosystem
+ sourceCompatibility = JavaVersion.VERSION_17
+ targetCompatibility = JavaVersion.VERSION_17
+
withJavadocJar()
withSourcesJar()
}