You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/10/22 06:14:52 UTC
(iggy) branch master updated: feat(proc): Implement Processor Infrastructure/Libraries (for flink to start with) (#2275)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk 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 3ab721d4e feat(proc): Implement Processor Infrastructure/Libraries (for flink to start with) (#2275)
3ab721d4e is described below
commit 3ab721d4e48590de3d4d44073578721761493cbe
Author: Chiradip Mandal <ch...@chiradip.com>
AuthorDate: Tue Oct 21 23:14:48 2025 -0700
feat(proc): Implement Processor Infrastructure/Libraries (for flink to start with) (#2275)
---
.../iggy-connector-library/README.md | 72 +++++++
.../iggy-connector-library/build.gradle.kts | 113 ++++++++++
.../connector/config/IggyConnectionConfig.java | 228 +++++++++++++++++++++
.../apache/iggy/connector/config/OffsetConfig.java | 153 ++++++++++++++
.../iggy/connector/error/ConnectorException.java | 94 +++++++++
.../iggy/connector/partition/PartitionInfo.java | 123 +++++++++++
.../serialization/DeserializationSchema.java | 63 ++++++
.../connector/serialization/RecordMetadata.java | 100 +++++++++
.../serialization/SerializationSchema.java | 65 ++++++
.../connector/serialization/TypeDescriptor.java | 88 ++++++++
.../connector/config/IggyConnectionConfigTest.java | 221 ++++++++++++++++++++
.../iggy/connector/config/OffsetConfigTest.java | 167 +++++++++++++++
.../connector/error/ConnectorExceptionTest.java | 168 +++++++++++++++
.../connector/partition/PartitionInfoTest.java | 157 ++++++++++++++
.../serialization/DeserializationSchemaTest.java | 171 ++++++++++++++++
.../serialization/RecordMetadataTest.java | 151 ++++++++++++++
.../serialization/SerializationSchemaTest.java | 208 +++++++++++++++++++
.../serialization/TypeDescriptorTest.java | 178 ++++++++++++++++
foreign/java/settings.gradle.kts | 7 +
19 files changed, 2527 insertions(+)
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md
new file mode 100644
index 000000000..b3c1f8c73
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/README.md
@@ -0,0 +1,72 @@
+# Apache Iggy Connector Library
+
+Core connector library for integrating Apache Iggy with stream processing engines.
+
+## Overview
+
+This library provides framework-agnostic abstractions and Flink-specific implementations for reading from and writing to Iggy streams.
+
+### Package Structure
+
+```text
+org.apache.iggy.connector/
+├── config/ - Configuration classes (connection, offset, retry)
+├── serialization/ - Serialization/deserialization interfaces
+├── offset/ - Offset tracking abstractions
+├── partition/ - Partition discovery and assignment
+├── error/ - Exception hierarchy
+├── util/ - Utility classes (client pooling, metrics)
+└── flink/ - Flink-specific implementations
+ ├── source/ - Flink Source API integration
+ ├── sink/ - Flink Sink API integration
+ └── config/ - Flink-specific configuration
+```
+
+## Design Philosophy
+
+### Framework-Agnostic Core
+
+The `org.apache.iggy.connector.*` packages (excluding `flink`) are designed to be framework-agnostic and reusable across different stream processing engines (Spark, Beam, etc.).
+
+**Design Principles:**
+
+- No Flink imports in common packages
+- Interface-based for extensibility
+- Serializable for distributed execution
+- Well-documented for external use
+
+### Flink Integration
+
+The `org.apache.iggy.connector.flink.*` packages provide Flink-specific implementations:
+
+- Implement Flink's Source and Sink APIs
+- Wrap common abstractions with Flink-specific adapters
+- Handle Flink checkpointing and state management
+
+## Usage
+
+### As a Dependency
+
+```kotlin
+dependencies {
+ implementation("org.apache.iggy:iggy-connector-library:0.5.0-SNAPSHOT")
+ compileOnly("org.apache.flink:flink-streaming-java:1.18.0")
+}
+```
+
+### Building from Source
+
+```bash
+cd iggy-connector-flink
+./gradlew :iggy-connector-library:build
+```
+
+## Future Evolution
+
+When Spark connector is added, common code may be extracted to a separate `iggy-connector-common` module. The current structure keeps Flink-first implementation simple while maintaining clear package boundaries for future refactoring.
+
+See [connector_library_placement_deliberation](https://github.com/apache/iggy/discussions/2236#discussioncomment-14702253) for architectural decisions.
+
+## License
+
+Apache License 2.0 - See LICENSE file for details.
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
new file mode 100644
index 000000000..ce4768fcf
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/build.gradle.kts
@@ -0,0 +1,113 @@
+/*
+ * 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.
+ */
+
+plugins {
+ id("java-library")
+ id("maven-publish")
+ id("checkstyle")
+}
+
+group = "org.apache.iggy"
+version = "0.5.0-SNAPSHOT"
+
+repositories {
+ mavenCentral()
+}
+
+java {
+ withJavadocJar()
+ withSourcesJar()
+}
+
+checkstyle {
+ toolVersion = "10.23.1"
+ configFile = file("../../../dev-support/checkstyle/checkstyle.xml")
+}
+
+val flinkVersion = "1.18.0"
+val iggyVersion = "0.5.0-SNAPSHOT"
+
+dependencies {
+ // Iggy SDK - use local project when building within Iggy repository
+ api(project(":iggy"))
+
+ // Flink dependencies (provided - not bundled with connector)
+ compileOnly("org.apache.flink:flink-connector-base:${flinkVersion}")
+ compileOnly("org.apache.flink:flink-streaming-java:${flinkVersion}")
+
+ // Serialization support
+ implementation("com.fasterxml.jackson.core:jackson-databind:2.18.0")
+
+ // Logging
+ compileOnly("org.slf4j:slf4j-api:2.0.16")
+
+ // Testing
+ testImplementation("org.apache.flink:flink-test-utils:${flinkVersion}")
+ testImplementation("org.apache.flink:flink-runtime:${flinkVersion}:tests")
+ 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()
+}
+
+tasks.withType<JavaCompile> {
+ options.encoding = "UTF-8"
+ options.compilerArgs.add("-parameters")
+}
+
+tasks.withType<Javadoc> {
+ options.encoding = "UTF-8"
+}
+
+publishing {
+ publications {
+ create<MavenPublication>("maven") {
+ groupId = "org.apache.iggy"
+ artifactId = "iggy-connector-library"
+ version = "0.5.0-SNAPSHOT"
+ from(components["java"])
+
+ pom {
+ name = "Apache Iggy Connector Library"
+ description = "Core connector library for Apache Iggy stream processors (Flink, Spark, etc.)"
+ url = "https://github.com/apache/iggy"
+ licenses {
+ license {
+ name = "Apache License, Version 2.0"
+ url = "https://www.apache.org/licenses/LICENSE-2.0.txt"
+ }
+ }
+ developers {
+ developer {
+ name = "Apache Iggy"
+ email = "dev@iggy.apache.org"
+ }
+ }
+ scm {
+ url = "https://github.com/apache/iggy"
+ connection = "scm:git:git://github.com/apache/iggy.git"
+ developerConnection = "scm:git:git://github.com/apache/iggy.git"
+ }
+ }
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/IggyConnectionConfig.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/IggyConnectionConfig.java
new file mode 100644
index 000000000..eff537a73
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/IggyConnectionConfig.java
@@ -0,0 +1,228 @@
+/*
+ * 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.config;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Configuration for connecting to an Iggy server.
+ * This class is framework-agnostic and can be reused across different
+ * stream processing engines (Flink, Spark, etc.).
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * IggyConnectionConfig config = IggyConnectionConfig.builder()
+ * .serverAddress("localhost:8080")
+ * .username("iggy")
+ * .password("iggy")
+ * .connectionTimeout(Duration.ofSeconds(30))
+ * .build();
+ * }</pre>
+ */
+public final class IggyConnectionConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String serverAddress;
+ private final String username;
+ private final String password;
+ private final Duration connectionTimeout;
+ private final Duration requestTimeout;
+ private final int maxRetries;
+ private final Duration retryBackoff;
+ private final boolean enableTls;
+
+ @SuppressWarnings("checkstyle:ParameterNumber")
+ private IggyConnectionConfig(
+ String serverAddress,
+ String username,
+ String password,
+ Duration connectionTimeout,
+ Duration requestTimeout,
+ int maxRetries,
+ Duration retryBackoff,
+ boolean enableTls) {
+ this.serverAddress = Objects.requireNonNull(serverAddress, "serverAddress must not be null");
+ this.username = Objects.requireNonNull(username, "username must not be null");
+ this.password = Objects.requireNonNull(password, "password must not be null");
+ this.connectionTimeout = Objects.requireNonNull(connectionTimeout, "connectionTimeout must not be null");
+ this.requestTimeout = Objects.requireNonNull(requestTimeout, "requestTimeout must not be null");
+ this.maxRetries = maxRetries;
+ this.retryBackoff = Objects.requireNonNull(retryBackoff, "retryBackoff must not be null");
+ this.enableTls = enableTls;
+
+ if (maxRetries < 0) {
+ throw new IllegalArgumentException("maxRetries must be non-negative");
+ }
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public String getServerAddress() {
+ return serverAddress;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public Duration getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ public Duration getRequestTimeout() {
+ return requestTimeout;
+ }
+
+ public int getMaxRetries() {
+ return maxRetries;
+ }
+
+ public Duration getRetryBackoff() {
+ return retryBackoff;
+ }
+
+ public boolean isEnableTls() {
+ return enableTls;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ IggyConnectionConfig that = (IggyConnectionConfig) o;
+ return maxRetries == that.maxRetries
+ && enableTls == that.enableTls
+ && Objects.equals(serverAddress, that.serverAddress)
+ && Objects.equals(username, that.username)
+ && Objects.equals(password, that.password)
+ && Objects.equals(connectionTimeout, that.connectionTimeout)
+ && Objects.equals(requestTimeout, that.requestTimeout)
+ && Objects.equals(retryBackoff, that.retryBackoff);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ serverAddress,
+ username,
+ password,
+ connectionTimeout,
+ requestTimeout,
+ maxRetries,
+ retryBackoff,
+ enableTls);
+ }
+
+ @Override
+ public String toString() {
+ return "IggyConnectionConfig{"
+ + "serverAddress='" + serverAddress + '\''
+ + ", username='" + username + '\''
+ + ", connectionTimeout=" + connectionTimeout
+ + ", requestTimeout=" + requestTimeout
+ + ", maxRetries=" + maxRetries
+ + ", retryBackoff=" + retryBackoff
+ + ", enableTls=" + enableTls
+ + '}';
+ }
+
+ /**
+ * Builder for {@link IggyConnectionConfig}.
+ */
+ @SuppressWarnings("checkstyle:HiddenField")
+ public static final class Builder {
+ private String serverAddress;
+ private String username;
+ private String password;
+ private Duration connectionTimeout = Duration.ofSeconds(30);
+ private Duration requestTimeout = Duration.ofSeconds(30);
+ private int maxRetries = 3;
+ private Duration retryBackoff = Duration.ofMillis(100);
+ private boolean enableTls = false;
+
+ private Builder() {
+ }
+
+ public Builder serverAddress(String serverAddress) {
+ this.serverAddress = serverAddress;
+ return this;
+ }
+
+ public Builder username(String username) {
+ this.username = username;
+ return this;
+ }
+
+ public Builder password(String password) {
+ this.password = password;
+ return this;
+ }
+
+ public Builder connectionTimeout(Duration connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ return this;
+ }
+
+ public Builder requestTimeout(Duration requestTimeout) {
+ this.requestTimeout = requestTimeout;
+ return this;
+ }
+
+ public Builder maxRetries(int maxRetries) {
+ this.maxRetries = maxRetries;
+ return this;
+ }
+
+ public Builder retryBackoff(Duration retryBackoff) {
+ this.retryBackoff = retryBackoff;
+ return this;
+ }
+
+ public Builder enableTls(boolean enableTls) {
+ this.enableTls = enableTls;
+ return this;
+ }
+
+ public IggyConnectionConfig build() {
+ return new IggyConnectionConfig(
+ serverAddress,
+ username,
+ password,
+ connectionTimeout,
+ requestTimeout,
+ maxRetries,
+ retryBackoff,
+ enableTls);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/OffsetConfig.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/OffsetConfig.java
new file mode 100644
index 000000000..e147b7df0
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/config/OffsetConfig.java
@@ -0,0 +1,153 @@
+/*
+ * 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.config;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Configuration for offset management strategies.
+ * Defines how offsets should be initialized, committed, and tracked.
+ */
+public final class OffsetConfig implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final OffsetResetStrategy resetStrategy;
+ private final Duration commitInterval;
+ private final boolean autoCommit;
+ private final int commitBatchSize;
+
+ private OffsetConfig(
+ OffsetResetStrategy resetStrategy,
+ Duration commitInterval,
+ boolean autoCommit,
+ int commitBatchSize) {
+ this.resetStrategy = Objects.requireNonNull(resetStrategy, "resetStrategy must not be null");
+ this.commitInterval = Objects.requireNonNull(commitInterval, "commitInterval must not be null");
+ this.autoCommit = autoCommit;
+ this.commitBatchSize = commitBatchSize;
+
+ if (commitBatchSize <= 0) {
+ throw new IllegalArgumentException("commitBatchSize must be positive");
+ }
+ }
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public OffsetResetStrategy getResetStrategy() {
+ return resetStrategy;
+ }
+
+ public Duration getCommitInterval() {
+ return commitInterval;
+ }
+
+ public boolean isAutoCommit() {
+ return autoCommit;
+ }
+
+ public int getCommitBatchSize() {
+ return commitBatchSize;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ OffsetConfig that = (OffsetConfig) o;
+ return autoCommit == that.autoCommit
+ && commitBatchSize == that.commitBatchSize
+ && resetStrategy == that.resetStrategy
+ && Objects.equals(commitInterval, that.commitInterval);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resetStrategy, commitInterval, autoCommit, commitBatchSize);
+ }
+
+ @Override
+ public String toString() {
+ return "OffsetConfig{"
+ + "resetStrategy=" + resetStrategy
+ + ", commitInterval=" + commitInterval
+ + ", autoCommit=" + autoCommit
+ + ", commitBatchSize=" + commitBatchSize
+ + '}';
+ }
+
+ /**
+ * Strategy for resetting offsets when no valid offset is found.
+ */
+ public enum OffsetResetStrategy {
+ /** Start reading from the earliest available offset. */
+ EARLIEST,
+ /** Start reading from the latest offset (skip existing messages). */
+ LATEST,
+ /** Fail if no valid offset is found. */
+ NONE
+ }
+
+ /**
+ * Builder for {@link OffsetConfig}.
+ */
+ @SuppressWarnings("checkstyle:HiddenField")
+ public static final class Builder {
+ private OffsetResetStrategy resetStrategy = OffsetResetStrategy.LATEST;
+ private Duration commitInterval = Duration.ofSeconds(5);
+ private boolean autoCommit = true;
+ private int commitBatchSize = 100;
+
+ private Builder() {
+ }
+
+ public Builder resetStrategy(OffsetResetStrategy resetStrategy) {
+ this.resetStrategy = resetStrategy;
+ return this;
+ }
+
+ public Builder commitInterval(Duration commitInterval) {
+ this.commitInterval = commitInterval;
+ return this;
+ }
+
+ public Builder autoCommit(boolean autoCommit) {
+ this.autoCommit = autoCommit;
+ return this;
+ }
+
+ public Builder commitBatchSize(int commitBatchSize) {
+ this.commitBatchSize = commitBatchSize;
+ return this;
+ }
+
+ public OffsetConfig build() {
+ return new OffsetConfig(resetStrategy, commitInterval, autoCommit, commitBatchSize);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/error/ConnectorException.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/error/ConnectorException.java
new file mode 100644
index 000000000..9c4273c06
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/error/ConnectorException.java
@@ -0,0 +1,94 @@
+/*
+ * 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.error;
+
+/**
+ * Base exception for all connector-related errors.
+ * This exception hierarchy is framework-agnostic and can be used
+ * across different stream processing integrations.
+ */
+public class ConnectorException extends RuntimeException {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ErrorCode errorCode;
+ private final boolean retryable;
+
+ public ConnectorException(String message) {
+ this(message, ErrorCode.UNKNOWN, false);
+ }
+
+ public ConnectorException(String message, Throwable cause) {
+ this(message, cause, ErrorCode.UNKNOWN, false);
+ }
+
+ public ConnectorException(String message, ErrorCode errorCode, boolean retryable) {
+ super(message);
+ this.errorCode = errorCode;
+ this.retryable = retryable;
+ }
+
+ public ConnectorException(String message, Throwable cause, ErrorCode errorCode, boolean retryable) {
+ super(message, cause);
+ this.errorCode = errorCode;
+ this.retryable = retryable;
+ }
+
+ /**
+ * Returns the error code associated with this exception.
+ */
+ public ErrorCode getErrorCode() {
+ return errorCode;
+ }
+
+ /**
+ * Returns whether this error is retryable.
+ * Retryable errors indicate transient failures that may succeed on retry.
+ */
+ public boolean isRetryable() {
+ return retryable;
+ }
+
+ /**
+ * Error codes for connector exceptions.
+ */
+ public enum ErrorCode {
+ /** Unknown or unclassified error. */
+ UNKNOWN,
+ /** Connection to Iggy server failed. */
+ CONNECTION_FAILED,
+ /** Authentication failed. */
+ AUTHENTICATION_FAILED,
+ /** Requested stream or topic not found. */
+ RESOURCE_NOT_FOUND,
+ /** Serialization or deserialization failed. */
+ SERIALIZATION_ERROR,
+ /** Offset management error. */
+ OFFSET_ERROR,
+ /** Partition discovery or assignment error. */
+ PARTITION_ERROR,
+ /** Configuration error. */
+ CONFIGURATION_ERROR,
+ /** Timeout waiting for operation. */
+ TIMEOUT,
+ /** Resource limit exceeded. */
+ RESOURCE_EXHAUSTED
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/partition/PartitionInfo.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/partition/PartitionInfo.java
new file mode 100644
index 000000000..50e1ea8dc
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/partition/PartitionInfo.java
@@ -0,0 +1,123 @@
+/*
+ * 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.partition;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Information about an Iggy partition.
+ * This is a framework-agnostic representation that can be used
+ * by different stream processing engines.
+ */
+public final class PartitionInfo implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String streamId;
+ private final String topicId;
+ private final int partitionId;
+ private final long currentOffset;
+ private final long endOffset;
+
+ private PartitionInfo(
+ String streamId, String topicId, int partitionId, long currentOffset, long endOffset) {
+ this.streamId = Objects.requireNonNull(streamId, "streamId must not be null");
+ this.topicId = Objects.requireNonNull(topicId, "topicId must not be null");
+ this.partitionId = partitionId;
+ this.currentOffset = currentOffset;
+ this.endOffset = endOffset;
+
+ if (partitionId < 0) {
+ throw new IllegalArgumentException("partitionId must be non-negative");
+ }
+ if (currentOffset < 0) {
+ throw new IllegalArgumentException("currentOffset must be non-negative");
+ }
+ if (endOffset < 0) {
+ throw new IllegalArgumentException("endOffset must be non-negative");
+ }
+ }
+
+ public static PartitionInfo of(
+ String streamId, String topicId, int partitionId, long currentOffset, long endOffset) {
+ return new PartitionInfo(streamId, topicId, partitionId, currentOffset, endOffset);
+ }
+
+ public String getStreamId() {
+ return streamId;
+ }
+
+ public String getTopicId() {
+ return topicId;
+ }
+
+ public int getPartitionId() {
+ return partitionId;
+ }
+
+ public long getCurrentOffset() {
+ return currentOffset;
+ }
+
+ public long getEndOffset() {
+ return endOffset;
+ }
+
+ /**
+ * Returns the number of messages available in this partition.
+ */
+ public long getAvailableMessages() {
+ return Math.max(0, endOffset - currentOffset);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PartitionInfo that = (PartitionInfo) o;
+ return partitionId == that.partitionId
+ && currentOffset == that.currentOffset
+ && endOffset == that.endOffset
+ && Objects.equals(streamId, that.streamId)
+ && Objects.equals(topicId, that.topicId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(streamId, topicId, partitionId, currentOffset, endOffset);
+ }
+
+ @Override
+ public String toString() {
+ return "PartitionInfo{"
+ + "streamId='" + streamId + '\''
+ + ", topicId='" + topicId + '\''
+ + ", partitionId=" + partitionId
+ + ", currentOffset=" + currentOffset
+ + ", endOffset=" + endOffset
+ + ", available=" + getAvailableMessages()
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/DeserializationSchema.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/DeserializationSchema.java
new file mode 100644
index 000000000..ec507bc82
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/DeserializationSchema.java
@@ -0,0 +1,63 @@
+/*
+ * 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.Serializable;
+
+/**
+ * Framework-agnostic interface for deserializing Iggy messages.
+ * This interface is designed to be wrapped by framework-specific
+ * deserialization adapters (e.g., Flink's DeserializationSchema,
+ * Spark's Decoder).
+ *
+ * @param <T> The type of the deserialized object
+ */
+public interface DeserializationSchema<T> extends Serializable {
+
+ /**
+ * Deserializes a message from byte array to the target type.
+ *
+ * @param data The message data as byte array
+ * @param metadata Metadata about the message (offset, partition, timestamp, etc.)
+ * @return The deserialized object
+ * @throws IOException If deserialization fails
+ */
+ T deserialize(byte[] data, RecordMetadata metadata) throws IOException;
+
+ /**
+ * Returns type information about the produced type.
+ * This is framework-agnostic type information that can be
+ * mapped to framework-specific type systems.
+ *
+ * @return Type descriptor for the produced type
+ */
+ TypeDescriptor<T> getProducedType();
+
+ /**
+ * Returns whether this schema can handle null values.
+ * If true, the deserialize method may be called with null data.
+ *
+ * @return true if null values are supported, false otherwise
+ */
+ default boolean isNullable() {
+ return false;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/RecordMetadata.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/RecordMetadata.java
new file mode 100644
index 000000000..34fa8a38d
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/RecordMetadata.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.serialization;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/**
+ * Metadata about a record from Iggy.
+ * Contains information like offset, partition, timestamp that may be
+ * useful during deserialization.
+ */
+public final class RecordMetadata implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private static final RecordMetadata EMPTY = new RecordMetadata(null, null, null, null);
+
+ private final String streamId;
+ private final String topicId;
+ private final Integer partitionId;
+ private final Long offset;
+
+ private RecordMetadata(String streamId, String topicId, Integer partitionId, Long offset) {
+ this.streamId = streamId;
+ this.topicId = topicId;
+ this.partitionId = partitionId;
+ this.offset = offset;
+ }
+
+ public static RecordMetadata empty() {
+ return EMPTY;
+ }
+
+ public static RecordMetadata of(String streamId, String topicId, Integer partitionId, Long offset) {
+ return new RecordMetadata(streamId, topicId, partitionId, offset);
+ }
+
+ public String getStreamId() {
+ return streamId;
+ }
+
+ public String getTopicId() {
+ return topicId;
+ }
+
+ public Integer getPartitionId() {
+ return partitionId;
+ }
+
+ public Long getOffset() {
+ return offset;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RecordMetadata that = (RecordMetadata) o;
+ return Objects.equals(streamId, that.streamId)
+ && Objects.equals(topicId, that.topicId)
+ && Objects.equals(partitionId, that.partitionId)
+ && Objects.equals(offset, that.offset);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(streamId, topicId, partitionId, offset);
+ }
+
+ @Override
+ public String toString() {
+ return "RecordMetadata{"
+ + "streamId='" + streamId + '\''
+ + ", topicId='" + topicId + '\''
+ + ", partitionId=" + partitionId
+ + ", offset=" + offset
+ + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/SerializationSchema.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/SerializationSchema.java
new file mode 100644
index 000000000..2e4ab6f54
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/SerializationSchema.java
@@ -0,0 +1,65 @@
+/*
+ * 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.Serializable;
+import java.util.Optional;
+
+/**
+ * Framework-agnostic interface for serializing objects to Iggy messages.
+ * This interface is designed to be wrapped by framework-specific
+ * serialization adapters.
+ *
+ * @param <T> The type of the object to serialize
+ */
+public interface SerializationSchema<T> extends Serializable {
+
+ /**
+ * Serializes an object to byte array for sending to Iggy.
+ *
+ * @param element The object to serialize
+ * @return The serialized message as byte array
+ * @throws IOException If serialization fails
+ */
+ byte[] serialize(T element) throws IOException;
+
+ /**
+ * Extracts a partition key from the element.
+ * If present, this key will be used to determine which partition
+ * the message should be sent to.
+ *
+ * @param element The object to extract key from
+ * @return Optional partition key
+ */
+ default Optional<Integer> extractPartitionKey(T element) {
+ return Optional.empty();
+ }
+
+ /**
+ * Returns whether this schema can handle null values.
+ * If true, the serialize method may be called with null element.
+ *
+ * @return true if null values are supported, false otherwise
+ */
+ default boolean isNullable() {
+ return false;
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/TypeDescriptor.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/TypeDescriptor.java
new file mode 100644
index 000000000..dcf14c020
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/main/java/org/apache/iggy/connector/serialization/TypeDescriptor.java
@@ -0,0 +1,88 @@
+/*
+ * 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.Serializable;
+import java.lang.reflect.Type;
+import java.util.Objects;
+
+/**
+ * Framework-agnostic type descriptor.
+ * This can be mapped to framework-specific type systems
+ * (e.g., Flink's TypeInformation, Spark's Encoder).
+ *
+ * @param <T> The described type
+ */
+public final class TypeDescriptor<T> implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Class<T> typeClass;
+ private final Type genericType;
+
+ private TypeDescriptor(Class<T> typeClass, Type genericType) {
+ this.typeClass = Objects.requireNonNull(typeClass, "typeClass must not be null");
+ this.genericType = genericType != null ? genericType : typeClass;
+ }
+
+ /**
+ * Creates a type descriptor for a simple class.
+ */
+ public static <T> TypeDescriptor<T> of(Class<T> typeClass) {
+ return new TypeDescriptor<>(typeClass, typeClass);
+ }
+
+ /**
+ * Creates a type descriptor with generic type information.
+ */
+ public static <T> TypeDescriptor<T> of(Class<T> typeClass, Type genericType) {
+ return new TypeDescriptor<>(typeClass, genericType);
+ }
+
+ public Class<T> getTypeClass() {
+ return typeClass;
+ }
+
+ public Type getGenericType() {
+ return genericType;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TypeDescriptor<?> that = (TypeDescriptor<?>) o;
+ return Objects.equals(typeClass, that.typeClass) && Objects.equals(genericType, that.genericType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(typeClass, genericType);
+ }
+
+ @Override
+ public String toString() {
+ return "TypeDescriptor{" + "typeClass=" + typeClass + ", genericType=" + genericType + '}';
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/IggyConnectionConfigTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/IggyConnectionConfigTest.java
new file mode 100644
index 000000000..467070c12
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/IggyConnectionConfigTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.config;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+
+class IggyConnectionConfigTest {
+
+ @Test
+ void shouldBuildValidConfig() {
+ IggyConnectionConfig config = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .connectionTimeout(Duration.ofSeconds(30))
+ .requestTimeout(Duration.ofSeconds(20))
+ .maxRetries(5)
+ .retryBackoff(Duration.ofMillis(200))
+ .enableTls(true)
+ .build();
+
+ assertThat(config.getServerAddress()).isEqualTo("localhost:8080");
+ assertThat(config.getUsername()).isEqualTo("iggy");
+ assertThat(config.getPassword()).isEqualTo("iggy");
+ assertThat(config.getConnectionTimeout()).isEqualTo(Duration.ofSeconds(30));
+ assertThat(config.getRequestTimeout()).isEqualTo(Duration.ofSeconds(20));
+ assertThat(config.getMaxRetries()).isEqualTo(5);
+ assertThat(config.getRetryBackoff()).isEqualTo(Duration.ofMillis(200));
+ assertThat(config.isEnableTls()).isTrue();
+ }
+
+ @Test
+ void shouldUseDefaultValues() {
+ IggyConnectionConfig config = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .build();
+
+ assertThat(config.getConnectionTimeout()).isEqualTo(Duration.ofSeconds(30));
+ assertThat(config.getRequestTimeout()).isEqualTo(Duration.ofSeconds(30));
+ assertThat(config.getMaxRetries()).isEqualTo(3);
+ assertThat(config.getRetryBackoff()).isEqualTo(Duration.ofMillis(100));
+ assertThat(config.isEnableTls()).isFalse();
+ }
+
+ @Test
+ void shouldThrowExceptionWhenServerAddressIsNull() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .username("iggy")
+ .password("iggy")
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("serverAddress must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenUsernameIsNull() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .password("iggy")
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("username must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenPasswordIsNull() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("password must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenConnectionTimeoutIsNull() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .connectionTimeout(null)
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("connectionTimeout must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenRequestTimeoutIsNull() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .requestTimeout(null)
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("requestTimeout must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenRetryBackoffIsNull() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .retryBackoff(null)
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("retryBackoff must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenMaxRetriesIsNegative() {
+ assertThatThrownBy(() -> IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .maxRetries(-1)
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("maxRetries must be non-negative");
+ }
+
+ @Test
+ void shouldAcceptZeroMaxRetries() {
+ IggyConnectionConfig config = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .maxRetries(0)
+ .build();
+
+ assertThat(config.getMaxRetries()).isZero();
+ }
+
+ @Test
+ void shouldImplementEqualsCorrectly() {
+ IggyConnectionConfig config1 = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .connectionTimeout(Duration.ofSeconds(30))
+ .requestTimeout(Duration.ofSeconds(20))
+ .maxRetries(3)
+ .retryBackoff(Duration.ofMillis(100))
+ .enableTls(true)
+ .build();
+
+ IggyConnectionConfig config2 = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .connectionTimeout(Duration.ofSeconds(30))
+ .requestTimeout(Duration.ofSeconds(20))
+ .maxRetries(3)
+ .retryBackoff(Duration.ofMillis(100))
+ .enableTls(true)
+ .build();
+
+ IggyConnectionConfig config3 = IggyConnectionConfig.builder()
+ .serverAddress("different:8080")
+ .username("iggy")
+ .password("iggy")
+ .build();
+
+ assertThat(config1).isEqualTo(config2);
+ assertThat(config1).hasSameHashCodeAs(config2);
+ assertThat(config1).isNotEqualTo(config3);
+ assertThat(config1).isEqualTo(config1);
+ assertThat(config1).isNotEqualTo(null);
+ assertThat(config1).isNotEqualTo(new Object());
+ }
+
+ @Test
+ void shouldImplementToStringCorrectly() {
+ IggyConnectionConfig config = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .build();
+
+ String toString = config.toString();
+ assertThat(toString).contains("IggyConnectionConfig");
+ assertThat(toString).contains("localhost:8080");
+ assertThat(toString).contains("iggy");
+ assertThat(toString).doesNotContain("password=iggy"); // Password should not be in toString
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ IggyConnectionConfig config = IggyConnectionConfig.builder()
+ .serverAddress("localhost:8080")
+ .username("iggy")
+ .password("iggy")
+ .build();
+
+ assertThat(config).isInstanceOf(java.io.Serializable.class);
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/OffsetConfigTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/OffsetConfigTest.java
new file mode 100644
index 000000000..dc2fbb9a6
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/config/OffsetConfigTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.config;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.time.Duration;
+import org.junit.jupiter.api.Test;
+
+class OffsetConfigTest {
+
+ @Test
+ void shouldBuildValidConfig() {
+ OffsetConfig config = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.EARLIEST)
+ .commitInterval(Duration.ofSeconds(10))
+ .autoCommit(false)
+ .commitBatchSize(200)
+ .build();
+
+ assertThat(config.getResetStrategy()).isEqualTo(OffsetConfig.OffsetResetStrategy.EARLIEST);
+ assertThat(config.getCommitInterval()).isEqualTo(Duration.ofSeconds(10));
+ assertThat(config.isAutoCommit()).isFalse();
+ assertThat(config.getCommitBatchSize()).isEqualTo(200);
+ }
+
+ @Test
+ void shouldUseDefaultValues() {
+ OffsetConfig config = OffsetConfig.builder().build();
+
+ assertThat(config.getResetStrategy()).isEqualTo(OffsetConfig.OffsetResetStrategy.LATEST);
+ assertThat(config.getCommitInterval()).isEqualTo(Duration.ofSeconds(5));
+ assertThat(config.isAutoCommit()).isTrue();
+ assertThat(config.getCommitBatchSize()).isEqualTo(100);
+ }
+
+ @Test
+ void shouldSupportAllResetStrategies() {
+ OffsetConfig earliestConfig = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.EARLIEST)
+ .build();
+ assertThat(earliestConfig.getResetStrategy()).isEqualTo(OffsetConfig.OffsetResetStrategy.EARLIEST);
+
+ OffsetConfig latestConfig = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.LATEST)
+ .build();
+ assertThat(latestConfig.getResetStrategy()).isEqualTo(OffsetConfig.OffsetResetStrategy.LATEST);
+
+ OffsetConfig noneConfig = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.NONE)
+ .build();
+ assertThat(noneConfig.getResetStrategy()).isEqualTo(OffsetConfig.OffsetResetStrategy.NONE);
+ }
+
+ @Test
+ void shouldThrowExceptionWhenResetStrategyIsNull() {
+ assertThatThrownBy(() -> OffsetConfig.builder()
+ .resetStrategy(null)
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("resetStrategy must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenCommitIntervalIsNull() {
+ assertThatThrownBy(() -> OffsetConfig.builder()
+ .commitInterval(null)
+ .build())
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("commitInterval must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenCommitBatchSizeIsZero() {
+ assertThatThrownBy(() -> OffsetConfig.builder()
+ .commitBatchSize(0)
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("commitBatchSize must be positive");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenCommitBatchSizeIsNegative() {
+ assertThatThrownBy(() -> OffsetConfig.builder()
+ .commitBatchSize(-1)
+ .build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("commitBatchSize must be positive");
+ }
+
+ @Test
+ void shouldAcceptMinimalCommitBatchSize() {
+ OffsetConfig config = OffsetConfig.builder()
+ .commitBatchSize(1)
+ .build();
+
+ assertThat(config.getCommitBatchSize()).isEqualTo(1);
+ }
+
+ @Test
+ void shouldImplementEqualsCorrectly() {
+ OffsetConfig config1 = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.EARLIEST)
+ .commitInterval(Duration.ofSeconds(10))
+ .autoCommit(false)
+ .commitBatchSize(200)
+ .build();
+
+ OffsetConfig config2 = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.EARLIEST)
+ .commitInterval(Duration.ofSeconds(10))
+ .autoCommit(false)
+ .commitBatchSize(200)
+ .build();
+
+ OffsetConfig config3 = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.LATEST)
+ .build();
+
+ assertThat(config1).isEqualTo(config2);
+ assertThat(config1).hasSameHashCodeAs(config2);
+ assertThat(config1).isNotEqualTo(config3);
+ assertThat(config1).isEqualTo(config1);
+ assertThat(config1).isNotEqualTo(null);
+ assertThat(config1).isNotEqualTo(new Object());
+ }
+
+ @Test
+ void shouldImplementToStringCorrectly() {
+ OffsetConfig config = OffsetConfig.builder()
+ .resetStrategy(OffsetConfig.OffsetResetStrategy.EARLIEST)
+ .commitInterval(Duration.ofSeconds(10))
+ .autoCommit(false)
+ .commitBatchSize(200)
+ .build();
+
+ String toString = config.toString();
+ assertThat(toString).contains("OffsetConfig");
+ assertThat(toString).contains("EARLIEST");
+ assertThat(toString).contains("autoCommit=false");
+ assertThat(toString).contains("commitBatchSize=200");
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ OffsetConfig config = OffsetConfig.builder().build();
+ assertThat(config).isInstanceOf(java.io.Serializable.class);
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/error/ConnectorExceptionTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/error/ConnectorExceptionTest.java
new file mode 100644
index 000000000..04dcc69a2
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/error/ConnectorExceptionTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.error;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+class ConnectorExceptionTest {
+
+ @Test
+ void shouldCreateExceptionWithMessage() {
+ ConnectorException exception = new ConnectorException("Test message");
+
+ assertThat(exception.getMessage()).isEqualTo("Test message");
+ assertThat(exception.getErrorCode()).isEqualTo(ConnectorException.ErrorCode.UNKNOWN);
+ assertThat(exception.isRetryable()).isFalse();
+ assertThat(exception.getCause()).isNull();
+ }
+
+ @Test
+ void shouldCreateExceptionWithMessageAndCause() {
+ Throwable cause = new RuntimeException("Root cause");
+ ConnectorException exception = new ConnectorException("Test message", cause);
+
+ assertThat(exception.getMessage()).isEqualTo("Test message");
+ assertThat(exception.getCause()).isEqualTo(cause);
+ assertThat(exception.getErrorCode()).isEqualTo(ConnectorException.ErrorCode.UNKNOWN);
+ assertThat(exception.isRetryable()).isFalse();
+ }
+
+ @Test
+ void shouldCreateExceptionWithMessageErrorCodeAndRetryable() {
+ ConnectorException exception = new ConnectorException(
+ "Connection failed",
+ ConnectorException.ErrorCode.CONNECTION_FAILED,
+ true
+ );
+
+ assertThat(exception.getMessage()).isEqualTo("Connection failed");
+ assertThat(exception.getErrorCode()).isEqualTo(ConnectorException.ErrorCode.CONNECTION_FAILED);
+ assertThat(exception.isRetryable()).isTrue();
+ assertThat(exception.getCause()).isNull();
+ }
+
+ @Test
+ void shouldCreateExceptionWithAllParameters() {
+ Throwable cause = new RuntimeException("Root cause");
+ ConnectorException exception = new ConnectorException(
+ "Connection failed",
+ cause,
+ ConnectorException.ErrorCode.CONNECTION_FAILED,
+ true
+ );
+
+ assertThat(exception.getMessage()).isEqualTo("Connection failed");
+ assertThat(exception.getCause()).isEqualTo(cause);
+ assertThat(exception.getErrorCode()).isEqualTo(ConnectorException.ErrorCode.CONNECTION_FAILED);
+ assertThat(exception.isRetryable()).isTrue();
+ }
+
+ @Test
+ void shouldSupportAllErrorCodes() {
+ assertThat(ConnectorException.ErrorCode.UNKNOWN).isNotNull();
+ assertThat(ConnectorException.ErrorCode.CONNECTION_FAILED).isNotNull();
+ assertThat(ConnectorException.ErrorCode.AUTHENTICATION_FAILED).isNotNull();
+ assertThat(ConnectorException.ErrorCode.RESOURCE_NOT_FOUND).isNotNull();
+ assertThat(ConnectorException.ErrorCode.SERIALIZATION_ERROR).isNotNull();
+ assertThat(ConnectorException.ErrorCode.OFFSET_ERROR).isNotNull();
+ assertThat(ConnectorException.ErrorCode.PARTITION_ERROR).isNotNull();
+ assertThat(ConnectorException.ErrorCode.CONFIGURATION_ERROR).isNotNull();
+ assertThat(ConnectorException.ErrorCode.TIMEOUT).isNotNull();
+ assertThat(ConnectorException.ErrorCode.RESOURCE_EXHAUSTED).isNotNull();
+ }
+
+ @Test
+ void shouldMarkConnectionErrorsAsRetryable() {
+ ConnectorException exception = new ConnectorException(
+ "Connection failed",
+ ConnectorException.ErrorCode.CONNECTION_FAILED,
+ true
+ );
+
+ assertThat(exception.isRetryable()).isTrue();
+ }
+
+ @Test
+ void shouldMarkAuthenticationErrorsAsNonRetryable() {
+ ConnectorException exception = new ConnectorException(
+ "Authentication failed",
+ ConnectorException.ErrorCode.AUTHENTICATION_FAILED,
+ false
+ );
+
+ assertThat(exception.isRetryable()).isFalse();
+ }
+
+ @Test
+ void shouldMarkSerializationErrorsAsNonRetryable() {
+ ConnectorException exception = new ConnectorException(
+ "Serialization failed",
+ ConnectorException.ErrorCode.SERIALIZATION_ERROR,
+ false
+ );
+
+ assertThat(exception.isRetryable()).isFalse();
+ }
+
+ @Test
+ void shouldMarkTimeoutErrorsAsRetryable() {
+ ConnectorException exception = new ConnectorException(
+ "Operation timed out",
+ ConnectorException.ErrorCode.TIMEOUT,
+ true
+ );
+
+ assertThat(exception.isRetryable()).isTrue();
+ }
+
+ @Test
+ void shouldMarkConfigurationErrorsAsNonRetryable() {
+ ConnectorException exception = new ConnectorException(
+ "Configuration error",
+ ConnectorException.ErrorCode.CONFIGURATION_ERROR,
+ false
+ );
+
+ assertThat(exception.isRetryable()).isFalse();
+ }
+
+ @Test
+ void shouldBeInstanceOfRuntimeException() {
+ ConnectorException exception = new ConnectorException("Test");
+ assertThat(exception).isInstanceOf(RuntimeException.class);
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ ConnectorException exception = new ConnectorException("Test");
+ assertThat(exception).isInstanceOf(java.io.Serializable.class);
+ }
+
+ @Test
+ void shouldPreserveStackTrace() {
+ Throwable cause = new RuntimeException("Root cause");
+ ConnectorException exception = new ConnectorException("Wrapper", cause);
+
+ assertThat(exception.getStackTrace()).isNotEmpty();
+ assertThat(exception.getCause().getStackTrace()).isNotEmpty();
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/partition/PartitionInfoTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/partition/PartitionInfoTest.java
new file mode 100644
index 000000000..13703c503
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/partition/PartitionInfoTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.partition;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+class PartitionInfoTest {
+
+ @Test
+ void shouldCreatePartitionInfoWithValidParameters() {
+ PartitionInfo partitionInfo = PartitionInfo.of("stream1", "topic1", 0, 100, 500);
+
+ assertThat(partitionInfo.getStreamId()).isEqualTo("stream1");
+ assertThat(partitionInfo.getTopicId()).isEqualTo("topic1");
+ assertThat(partitionInfo.getPartitionId()).isZero();
+ assertThat(partitionInfo.getCurrentOffset()).isEqualTo(100);
+ assertThat(partitionInfo.getEndOffset()).isEqualTo(500);
+ assertThat(partitionInfo.getAvailableMessages()).isEqualTo(400);
+ }
+
+ @Test
+ void shouldCalculateAvailableMessagesCorrectly() {
+ PartitionInfo partitionInfo = PartitionInfo.of("stream1", "topic1", 0, 100, 500);
+ assertThat(partitionInfo.getAvailableMessages()).isEqualTo(400);
+
+ PartitionInfo emptyPartition = PartitionInfo.of("stream1", "topic1", 0, 500, 500);
+ assertThat(emptyPartition.getAvailableMessages()).isZero();
+
+ PartitionInfo newPartition = PartitionInfo.of("stream1", "topic1", 0, 0, 1000);
+ assertThat(newPartition.getAvailableMessages()).isEqualTo(1000);
+ }
+
+ @Test
+ void shouldReturnZeroWhenCurrentOffsetExceedsEndOffset() {
+ // This can happen in edge cases during offset management
+ PartitionInfo partitionInfo = PartitionInfo.of("stream1", "topic1", 0, 600, 500);
+ assertThat(partitionInfo.getAvailableMessages()).isZero();
+ }
+
+ @Test
+ void shouldThrowExceptionWhenStreamIdIsNull() {
+ assertThatThrownBy(() -> PartitionInfo.of(null, "topic1", 0, 100, 500))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("streamId must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenTopicIdIsNull() {
+ assertThatThrownBy(() -> PartitionInfo.of("stream1", null, 0, 100, 500))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("topicId must not be null");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenPartitionIdIsNegative() {
+ assertThatThrownBy(() -> PartitionInfo.of("stream1", "topic1", -1, 100, 500))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("partitionId must be non-negative");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenCurrentOffsetIsNegative() {
+ assertThatThrownBy(() -> PartitionInfo.of("stream1", "topic1", 0, -1, 500))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("currentOffset must be non-negative");
+ }
+
+ @Test
+ void shouldThrowExceptionWhenEndOffsetIsNegative() {
+ assertThatThrownBy(() -> PartitionInfo.of("stream1", "topic1", 0, 100, -1))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("endOffset must be non-negative");
+ }
+
+ @Test
+ void shouldAcceptZeroValues() {
+ PartitionInfo partitionInfo = PartitionInfo.of("stream1", "topic1", 0, 0, 0);
+
+ assertThat(partitionInfo.getPartitionId()).isZero();
+ assertThat(partitionInfo.getCurrentOffset()).isZero();
+ assertThat(partitionInfo.getEndOffset()).isZero();
+ assertThat(partitionInfo.getAvailableMessages()).isZero();
+ }
+
+ @Test
+ void shouldSupportMultiplePartitions() {
+ PartitionInfo partition0 = PartitionInfo.of("stream1", "topic1", 0, 100, 500);
+ PartitionInfo partition1 = PartitionInfo.of("stream1", "topic1", 1, 200, 600);
+ PartitionInfo partition99 = PartitionInfo.of("stream1", "topic1", 99, 0, 1000);
+
+ assertThat(partition0.getPartitionId()).isZero();
+ assertThat(partition1.getPartitionId()).isEqualTo(1);
+ assertThat(partition99.getPartitionId()).isEqualTo(99);
+ }
+
+ @Test
+ void shouldImplementEqualsCorrectly() {
+ PartitionInfo partition1 = PartitionInfo.of("stream1", "topic1", 0, 100, 500);
+ PartitionInfo partition2 = PartitionInfo.of("stream1", "topic1", 0, 100, 500);
+ PartitionInfo partition3 = PartitionInfo.of("stream2", "topic1", 0, 100, 500);
+ PartitionInfo partition4 = PartitionInfo.of("stream1", "topic2", 0, 100, 500);
+ PartitionInfo partition5 = PartitionInfo.of("stream1", "topic1", 1, 100, 500);
+ PartitionInfo partition6 = PartitionInfo.of("stream1", "topic1", 0, 200, 500);
+ PartitionInfo partition7 = PartitionInfo.of("stream1", "topic1", 0, 100, 600);
+
+ assertThat(partition1).isEqualTo(partition2);
+ assertThat(partition1).hasSameHashCodeAs(partition2);
+ assertThat(partition1).isNotEqualTo(partition3);
+ assertThat(partition1).isNotEqualTo(partition4);
+ assertThat(partition1).isNotEqualTo(partition5);
+ assertThat(partition1).isNotEqualTo(partition6);
+ assertThat(partition1).isNotEqualTo(partition7);
+ assertThat(partition1).isEqualTo(partition1);
+ assertThat(partition1).isNotEqualTo(null);
+ assertThat(partition1).isNotEqualTo(new Object());
+ }
+
+ @Test
+ void shouldImplementToStringCorrectly() {
+ PartitionInfo partitionInfo = PartitionInfo.of("stream1", "topic1", 5, 100, 500);
+
+ String toString = partitionInfo.toString();
+ assertThat(toString).contains("PartitionInfo");
+ assertThat(toString).contains("stream1");
+ assertThat(toString).contains("topic1");
+ assertThat(toString).contains("partitionId=5");
+ assertThat(toString).contains("currentOffset=100");
+ assertThat(toString).contains("endOffset=500");
+ assertThat(toString).contains("available=400");
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ PartitionInfo partitionInfo = PartitionInfo.of("stream1", "topic1", 0, 100, 500);
+ assertThat(partitionInfo).isInstanceOf(java.io.Serializable.class);
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/DeserializationSchemaTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/DeserializationSchemaTest.java
new file mode 100644
index 000000000..5f52c3de6
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/DeserializationSchemaTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+
+class DeserializationSchemaTest {
+
+ @Test
+ void shouldDeserializeData() throws IOException {
+ DeserializationSchema<String> schema = new TestDeserializationSchema();
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 0, 100L);
+
+ String result = schema.deserialize("test-data".getBytes(StandardCharsets.UTF_8), metadata);
+
+ assertThat(result).isEqualTo("test-data");
+ }
+
+ @Test
+ void shouldReturnCorrectTypeDescriptor() {
+ DeserializationSchema<String> schema = new TestDeserializationSchema();
+
+ TypeDescriptor<String> typeDescriptor = schema.getProducedType();
+
+ assertThat(typeDescriptor.getTypeClass()).isEqualTo(String.class);
+ }
+
+ @Test
+ void shouldReturnFalseForIsNullableByDefault() {
+ DeserializationSchema<String> schema = new TestDeserializationSchema();
+
+ assertThat(schema.isNullable()).isFalse();
+ }
+
+ @Test
+ void shouldAllowOverridingIsNullable() {
+ DeserializationSchema<String> schema = new DeserializationSchema<String>() {
+ @Override
+ public String deserialize(byte[] data, RecordMetadata metadata) {
+ return null;
+ }
+
+ @Override
+ public TypeDescriptor<String> getProducedType() {
+ return TypeDescriptor.of(String.class);
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+ };
+
+ assertThat(schema.isNullable()).isTrue();
+ }
+
+ @Test
+ void shouldHandleEmptyData() throws IOException {
+ DeserializationSchema<String> schema = new TestDeserializationSchema();
+ RecordMetadata metadata = RecordMetadata.empty();
+
+ String result = schema.deserialize(new byte[0], metadata);
+
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void shouldPropagateIOException() {
+ DeserializationSchema<String> schema = new DeserializationSchema<String>() {
+ @Override
+ public String deserialize(byte[] data, RecordMetadata metadata) throws IOException {
+ throw new IOException("Deserialization failed");
+ }
+
+ @Override
+ public TypeDescriptor<String> getProducedType() {
+ return TypeDescriptor.of(String.class);
+ }
+ };
+
+ assertThatThrownBy(() -> schema.deserialize("test".getBytes(), RecordMetadata.empty()))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Deserialization failed");
+ }
+
+ @Test
+ void shouldSupportDifferentTypes() throws IOException {
+ DeserializationSchema<Integer> intSchema = new DeserializationSchema<Integer>() {
+ @Override
+ public Integer deserialize(byte[] data, RecordMetadata metadata) {
+ return Integer.parseInt(new String(data, StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public TypeDescriptor<Integer> getProducedType() {
+ return TypeDescriptor.of(Integer.class);
+ }
+ };
+
+ Integer result = intSchema.deserialize("42".getBytes(StandardCharsets.UTF_8), RecordMetadata.empty());
+ assertThat(result).isEqualTo(42);
+ assertThat(intSchema.getProducedType().getTypeClass()).isEqualTo(Integer.class);
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ DeserializationSchema<String> schema = new TestDeserializationSchema();
+ assertThat(schema).isInstanceOf(java.io.Serializable.class);
+ }
+
+ @Test
+ void shouldAccessMetadataInDeserialization() throws IOException {
+ DeserializationSchema<String> schema = new DeserializationSchema<String>() {
+ @Override
+ public String deserialize(byte[] data, RecordMetadata metadata) {
+ String content = new String(data, StandardCharsets.UTF_8);
+ return String.format("%s:%s:%d:%d:%s",
+ metadata.getStreamId(),
+ metadata.getTopicId(),
+ metadata.getPartitionId(),
+ metadata.getOffset(),
+ content);
+ }
+
+ @Override
+ public TypeDescriptor<String> getProducedType() {
+ return TypeDescriptor.of(String.class);
+ }
+ };
+
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 5, 1000L);
+ String result = schema.deserialize("data".getBytes(StandardCharsets.UTF_8), metadata);
+
+ assertThat(result).isEqualTo("stream1:topic1:5:1000:data");
+ }
+
+ // Helper implementation for testing
+ private static final class TestDeserializationSchema implements DeserializationSchema<String> {
+ @Override
+ public String deserialize(byte[] data, RecordMetadata metadata) {
+ return new String(data, StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public TypeDescriptor<String> getProducedType() {
+ return TypeDescriptor.of(String.class);
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/RecordMetadataTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/RecordMetadataTest.java
new file mode 100644
index 000000000..729b27186
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/RecordMetadataTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Test;
+
+class RecordMetadataTest {
+
+ @Test
+ void shouldCreateMetadataWithAllParameters() {
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 5, 1000L);
+
+ assertThat(metadata.getStreamId()).isEqualTo("stream1");
+ assertThat(metadata.getTopicId()).isEqualTo("topic1");
+ assertThat(metadata.getPartitionId()).isEqualTo(5);
+ assertThat(metadata.getOffset()).isEqualTo(1000L);
+ }
+
+ @Test
+ void shouldCreateEmptyMetadata() {
+ RecordMetadata metadata = RecordMetadata.empty();
+
+ assertThat(metadata.getStreamId()).isNull();
+ assertThat(metadata.getTopicId()).isNull();
+ assertThat(metadata.getPartitionId()).isNull();
+ assertThat(metadata.getOffset()).isNull();
+ }
+
+ @Test
+ void shouldAllowNullValues() {
+ RecordMetadata metadata = RecordMetadata.of(null, null, null, null);
+
+ assertThat(metadata.getStreamId()).isNull();
+ assertThat(metadata.getTopicId()).isNull();
+ assertThat(metadata.getPartitionId()).isNull();
+ assertThat(metadata.getOffset()).isNull();
+ }
+
+ @Test
+ void shouldSupportPartialMetadata() {
+ RecordMetadata streamOnly = RecordMetadata.of("stream1", null, null, null);
+ assertThat(streamOnly.getStreamId()).isEqualTo("stream1");
+ assertThat(streamOnly.getTopicId()).isNull();
+
+ RecordMetadata streamAndTopic = RecordMetadata.of("stream1", "topic1", null, null);
+ assertThat(streamAndTopic.getStreamId()).isEqualTo("stream1");
+ assertThat(streamAndTopic.getTopicId()).isEqualTo("topic1");
+ assertThat(streamAndTopic.getPartitionId()).isNull();
+ }
+
+ @Test
+ void shouldSupportZeroValues() {
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 0, 0L);
+
+ assertThat(metadata.getPartitionId()).isZero();
+ assertThat(metadata.getOffset()).isZero();
+ }
+
+ @Test
+ void shouldSupportLargeOffsets() {
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 999, Long.MAX_VALUE);
+
+ assertThat(metadata.getPartitionId()).isEqualTo(999);
+ assertThat(metadata.getOffset()).isEqualTo(Long.MAX_VALUE);
+ }
+
+ @Test
+ void shouldImplementEqualsCorrectly() {
+ RecordMetadata metadata1 = RecordMetadata.of("stream1", "topic1", 5, 1000L);
+ RecordMetadata metadata2 = RecordMetadata.of("stream1", "topic1", 5, 1000L);
+ RecordMetadata metadata3 = RecordMetadata.of("stream2", "topic1", 5, 1000L);
+ RecordMetadata metadata4 = RecordMetadata.of("stream1", "topic2", 5, 1000L);
+ RecordMetadata metadata5 = RecordMetadata.of("stream1", "topic1", 6, 1000L);
+ RecordMetadata metadata6 = RecordMetadata.of("stream1", "topic1", 5, 2000L);
+
+ assertThat(metadata1).isEqualTo(metadata2);
+ assertThat(metadata1).hasSameHashCodeAs(metadata2);
+ assertThat(metadata1).isNotEqualTo(metadata3);
+ assertThat(metadata1).isNotEqualTo(metadata4);
+ assertThat(metadata1).isNotEqualTo(metadata5);
+ assertThat(metadata1).isNotEqualTo(metadata6);
+ assertThat(metadata1).isEqualTo(metadata1);
+ assertThat(metadata1).isNotEqualTo(null);
+ assertThat(metadata1).isNotEqualTo(new Object());
+ }
+
+ @Test
+ void shouldHandleEqualsWithNullFields() {
+ RecordMetadata empty1 = RecordMetadata.empty();
+ RecordMetadata empty2 = RecordMetadata.empty();
+ RecordMetadata partial = RecordMetadata.of("stream1", null, null, null);
+
+ assertThat(empty1).isEqualTo(empty2);
+ assertThat(empty1).hasSameHashCodeAs(empty2);
+ assertThat(empty1).isNotEqualTo(partial);
+ }
+
+ @Test
+ void shouldImplementToStringCorrectly() {
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 5, 1000L);
+
+ String toString = metadata.toString();
+ assertThat(toString).contains("RecordMetadata");
+ assertThat(toString).contains("stream1");
+ assertThat(toString).contains("topic1");
+ assertThat(toString).contains("partitionId=5");
+ assertThat(toString).contains("offset=1000");
+ }
+
+ @Test
+ void shouldHandleToStringWithNullFields() {
+ RecordMetadata metadata = RecordMetadata.empty();
+
+ String toString = metadata.toString();
+ assertThat(toString).contains("RecordMetadata");
+ assertThat(toString).contains("null");
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ RecordMetadata metadata = RecordMetadata.of("stream1", "topic1", 5, 1000L);
+ assertThat(metadata).isInstanceOf(java.io.Serializable.class);
+ }
+
+ @Test
+ void shouldReuseEmptyInstance() {
+ RecordMetadata empty1 = RecordMetadata.empty();
+ RecordMetadata empty2 = RecordMetadata.empty();
+
+ assertThat(empty1).isSameAs(empty2);
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/SerializationSchemaTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/SerializationSchemaTest.java
new file mode 100644
index 000000000..c9eff24bb
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/SerializationSchemaTest.java
@@ -0,0 +1,208 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import org.junit.jupiter.api.Test;
+
+class SerializationSchemaTest {
+
+ @Test
+ void shouldSerializeData() throws IOException {
+ SerializationSchema<String> schema = new TestSerializationSchema();
+
+ byte[] result = schema.serialize("test-data");
+
+ assertThat(new String(result, StandardCharsets.UTF_8)).isEqualTo("test-data");
+ }
+
+ @Test
+ void shouldReturnEmptyOptionalForPartitionKeyByDefault() {
+ SerializationSchema<String> schema = new TestSerializationSchema();
+
+ Optional<Integer> partitionKey = schema.extractPartitionKey("test-data");
+
+ assertThat(partitionKey).isEmpty();
+ }
+
+ @Test
+ void shouldAllowOverridingExtractPartitionKey() {
+ SerializationSchema<String> schema = new SerializationSchema<String>() {
+ @Override
+ public byte[] serialize(String element) {
+ return element.getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public Optional<Integer> extractPartitionKey(String element) {
+ return Optional.of(element.hashCode());
+ }
+ };
+
+ Optional<Integer> partitionKey = schema.extractPartitionKey("test-data");
+
+ assertThat(partitionKey).isPresent();
+ assertThat(partitionKey.get()).isEqualTo("test-data".hashCode());
+ }
+
+ @Test
+ void shouldReturnFalseForIsNullableByDefault() {
+ SerializationSchema<String> schema = new TestSerializationSchema();
+
+ assertThat(schema.isNullable()).isFalse();
+ }
+
+ @Test
+ void shouldAllowOverridingIsNullable() {
+ SerializationSchema<String> schema = new SerializationSchema<String>() {
+ @Override
+ public byte[] serialize(String element) {
+ return element != null ? element.getBytes(StandardCharsets.UTF_8) : new byte[0];
+ }
+
+ @Override
+ public boolean isNullable() {
+ return true;
+ }
+ };
+
+ assertThat(schema.isNullable()).isTrue();
+ }
+
+ @Test
+ void shouldHandleEmptyString() throws IOException {
+ SerializationSchema<String> schema = new TestSerializationSchema();
+
+ byte[] result = schema.serialize("");
+
+ assertThat(result).isEmpty();
+ }
+
+ @Test
+ void shouldPropagateIOException() {
+ SerializationSchema<String> schema = new SerializationSchema<String>() {
+ @Override
+ public byte[] serialize(String element) throws IOException {
+ throw new IOException("Serialization failed");
+ }
+ };
+
+ assertThatThrownBy(() -> schema.serialize("test"))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Serialization failed");
+ }
+
+ @Test
+ void shouldSupportDifferentTypes() throws IOException {
+ SerializationSchema<Integer> intSchema = new SerializationSchema<Integer>() {
+ @Override
+ public byte[] serialize(Integer element) {
+ return element.toString().getBytes(StandardCharsets.UTF_8);
+ }
+ };
+
+ byte[] result = intSchema.serialize(42);
+ assertThat(new String(result, StandardCharsets.UTF_8)).isEqualTo("42");
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ SerializationSchema<String> schema = new TestSerializationSchema();
+ assertThat(schema).isInstanceOf(java.io.Serializable.class);
+ }
+
+ @Test
+ void shouldSupportCustomPartitionKeyExtraction() {
+ SerializationSchema<TestRecord> schema = new SerializationSchema<TestRecord>() {
+ @Override
+ public byte[] serialize(TestRecord element) {
+ return element.data.getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public Optional<Integer> extractPartitionKey(TestRecord element) {
+ return Optional.of(element.partitionKey);
+ }
+ };
+
+ TestRecord record = new TestRecord("test-data", 5);
+ Optional<Integer> partitionKey = schema.extractPartitionKey(record);
+
+ assertThat(partitionKey).isPresent();
+ assertThat(partitionKey.get()).isEqualTo(5);
+ }
+
+ @Test
+ void shouldSupportConditionalPartitionKeyExtraction() {
+ SerializationSchema<TestRecord> schema = new SerializationSchema<TestRecord>() {
+ @Override
+ public byte[] serialize(TestRecord element) {
+ return element.data.getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public Optional<Integer> extractPartitionKey(TestRecord element) {
+ // Only return partition key if it's positive
+ return element.partitionKey > 0 ? Optional.of(element.partitionKey) : Optional.empty();
+ }
+ };
+
+ TestRecord recordWithKey = new TestRecord("test-data", 5);
+ TestRecord recordWithoutKey = new TestRecord("test-data", -1);
+
+ assertThat(schema.extractPartitionKey(recordWithKey)).isPresent();
+ assertThat(schema.extractPartitionKey(recordWithoutKey)).isEmpty();
+ }
+
+ @Test
+ void shouldHandleLargeData() throws IOException {
+ SerializationSchema<String> schema = new TestSerializationSchema();
+ String largeData = "x".repeat(10000);
+
+ byte[] result = schema.serialize(largeData);
+
+ assertThat(result).hasSize(10000);
+ assertThat(new String(result, StandardCharsets.UTF_8)).isEqualTo(largeData);
+ }
+
+ // Helper implementation for testing
+ private static final class TestSerializationSchema implements SerializationSchema<String> {
+ @Override
+ public byte[] serialize(String element) {
+ return element.getBytes(StandardCharsets.UTF_8);
+ }
+ }
+
+ // Helper class for testing custom types
+ private static final class TestRecord {
+ private final String data;
+ private final int partitionKey;
+
+ TestRecord(String data, int partitionKey) {
+ this.data = data;
+ this.partitionKey = partitionKey;
+ }
+ }
+}
diff --git a/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/TypeDescriptorTest.java b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/TypeDescriptorTest.java
new file mode 100644
index 000000000..ab76c8e4f
--- /dev/null
+++ b/foreign/java/external-processors/iggy-connector-flink/iggy-connector-library/src/test/java/org/apache/iggy/connector/serialization/TypeDescriptorTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class TypeDescriptorTest {
+
+ @Test
+ void shouldCreateDescriptorWithSimpleClass() {
+ TypeDescriptor<String> descriptor = TypeDescriptor.of(String.class);
+
+ assertThat(descriptor.getTypeClass()).isEqualTo(String.class);
+ assertThat(descriptor.getGenericType()).isEqualTo(String.class);
+ }
+
+ @Test
+ void shouldCreateDescriptorWithGenericType() {
+ Type genericType = new ParameterizedType() {
+ @Override
+ public Type[] getActualTypeArguments() {
+ return new Type[]{String.class};
+ }
+
+ @Override
+ public Type getRawType() {
+ return List.class;
+ }
+
+ @Override
+ public Type getOwnerType() {
+ return null;
+ }
+ };
+
+ @SuppressWarnings("unchecked")
+ TypeDescriptor<List> descriptor = TypeDescriptor.of((Class<List>) List.class, genericType);
+
+ assertThat(descriptor.getTypeClass()).isEqualTo(List.class);
+ assertThat(descriptor.getGenericType()).isEqualTo(genericType);
+ }
+
+ @Test
+ void shouldUseTypeClassAsGenericTypeWhenGenericTypeIsNull() {
+ TypeDescriptor<String> descriptor = TypeDescriptor.of(String.class, null);
+
+ assertThat(descriptor.getTypeClass()).isEqualTo(String.class);
+ assertThat(descriptor.getGenericType()).isEqualTo(String.class);
+ }
+
+ @Test
+ void shouldThrowExceptionWhenTypeClassIsNull() {
+ assertThatThrownBy(() -> TypeDescriptor.of(null))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("typeClass must not be null");
+ }
+
+ @Test
+ void shouldSupportPrimitiveTypes() {
+ TypeDescriptor<Integer> intDescriptor = TypeDescriptor.of(Integer.class);
+ TypeDescriptor<Long> longDescriptor = TypeDescriptor.of(Long.class);
+ TypeDescriptor<Double> doubleDescriptor = TypeDescriptor.of(Double.class);
+ TypeDescriptor<Boolean> booleanDescriptor = TypeDescriptor.of(Boolean.class);
+
+ assertThat(intDescriptor.getTypeClass()).isEqualTo(Integer.class);
+ assertThat(longDescriptor.getTypeClass()).isEqualTo(Long.class);
+ assertThat(doubleDescriptor.getTypeClass()).isEqualTo(Double.class);
+ assertThat(booleanDescriptor.getTypeClass()).isEqualTo(Boolean.class);
+ }
+
+ @Test
+ void shouldSupportCustomClasses() {
+ TypeDescriptor<CustomClass> descriptor = TypeDescriptor.of(CustomClass.class);
+
+ assertThat(descriptor.getTypeClass()).isEqualTo(CustomClass.class);
+ }
+
+ @Test
+ void shouldImplementEqualsCorrectly() {
+ TypeDescriptor<String> descriptor1 = TypeDescriptor.of(String.class);
+ TypeDescriptor<String> descriptor2 = TypeDescriptor.of(String.class);
+ TypeDescriptor<Integer> descriptor3 = TypeDescriptor.of(Integer.class);
+
+ assertThat(descriptor1).isEqualTo(descriptor2);
+ assertThat(descriptor1).hasSameHashCodeAs(descriptor2);
+ assertThat(descriptor1).isNotEqualTo(descriptor3);
+ assertThat(descriptor1).isEqualTo(descriptor1);
+ assertThat(descriptor1).isNotEqualTo(null);
+ assertThat(descriptor1).isNotEqualTo(new Object());
+ }
+
+ @Test
+ void shouldConsiderGenericTypeInEquals() {
+ Type genericType1 = new ParameterizedType() {
+ @Override
+ public Type[] getActualTypeArguments() {
+ return new Type[]{String.class};
+ }
+
+ @Override
+ public Type getRawType() {
+ return List.class;
+ }
+
+ @Override
+ public Type getOwnerType() {
+ return null;
+ }
+ };
+
+ Type genericType2 = new ParameterizedType() {
+ @Override
+ public Type[] getActualTypeArguments() {
+ return new Type[]{Integer.class};
+ }
+
+ @Override
+ public Type getRawType() {
+ return List.class;
+ }
+
+ @Override
+ public Type getOwnerType() {
+ return null;
+ }
+ };
+
+ @SuppressWarnings("unchecked")
+ TypeDescriptor<List> descriptor1 = TypeDescriptor.of((Class<List>) List.class, genericType1);
+ @SuppressWarnings("unchecked")
+ TypeDescriptor<List> descriptor2 = TypeDescriptor.of((Class<List>) List.class, genericType2);
+
+ assertThat(descriptor1).isNotEqualTo(descriptor2);
+ }
+
+ @Test
+ void shouldImplementToStringCorrectly() {
+ TypeDescriptor<String> descriptor = TypeDescriptor.of(String.class);
+
+ String toString = descriptor.toString();
+ assertThat(toString).contains("TypeDescriptor");
+ assertThat(toString).contains("String");
+ }
+
+ @Test
+ void shouldBeSerializable() {
+ TypeDescriptor<String> descriptor = TypeDescriptor.of(String.class);
+ assertThat(descriptor).isInstanceOf(java.io.Serializable.class);
+ }
+
+ // Helper class for testing custom types
+ private static final class CustomClass {
+ private String value;
+ }
+}
diff --git a/foreign/java/settings.gradle.kts b/foreign/java/settings.gradle.kts
index 653ef334e..0af0da426 100644
--- a/foreign/java/settings.gradle.kts
+++ b/foreign/java/settings.gradle.kts
@@ -24,3 +24,10 @@ project(":iggy").projectDir = file("java-sdk")
include("iggy-example")
project(":iggy-example").projectDir = file("examples")
+
+// External processors - Stream processing integrations
+include("iggy-connector-library")
+project(":iggy-connector-library").projectDir = file("external-processors/iggy-connector-flink/iggy-connector-library")
+
+include("iggy-flink-examples")
+project(":iggy-flink-examples").projectDir = file("external-processors/iggy-connector-flink/iggy-flink-examples")