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 2026/01/27 19:05:49 UTC
(iggy) branch java-sdk-improvements created (now 8a7f5be98)
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a change to branch java-sdk-improvements
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 8a7f5be98 feat(java): refactor SDK with fluent builder API and exception hierarchy
This branch includes the following new commits:
new 8a7f5be98 feat(java): refactor SDK with fluent builder API and exception hierarchy
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(java): refactor SDK with fluent builder API and exception hierarchy
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
maciej pushed a commit to branch java-sdk-improvements
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8a7f5be98d8153a196ded1b734c2938f6085172b
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Tue Jan 27 13:00:28 2026 +0100
feat(java): refactor SDK with fluent builder API and exception hierarchy
- Add protocol-first fluent API: Iggy.tcp().blocking(), Iggy.http().blocking()
- Extract builder logic into dedicated builder classes
- Add semantic exceptions: Authentication, Authorization, NotFound, Conflict
- Add RetryPolicy with exponential backoff support
- Add IggyVersion for build-time version info
---
.../flink/source/IggyPartitionSplitReader.java | 2 +-
.../iggy/connector/flink/source/IggySource.java | 14 +-
.../flink/source/IggySourceSplitEnumerator.java | 2 +-
.../flink/example/AsyncTcpMessagePollTest.java | 28 +-
.../flink/example/AsyncTcpMessageSendTest.java | 24 +-
.../iggy/flink/example/SendTextDataTest.java | 8 +-
.../pinot/consumer/IggyPartitionGroupConsumer.java | 8 +-
.../pinot/metadata/IggyStreamMetadataProvider.java | 8 +-
foreign/java/java-sdk/build.gradle.kts | 24 ++
.../src/main/java/org/apache/iggy/Iggy.java | 128 +++++++-
.../src/main/java/org/apache/iggy/IggyVersion.java | 131 ++++++++
.../org/apache/iggy/builder/HttpClientBuilder.java | 66 ++++
.../org/apache/iggy/builder/TcpClientBuilder.java | 86 ++++++
.../package-info.java} | 31 +-
.../apache/iggy/client/async/MessagesClient.java | 12 +-
.../apache/iggy/client/async/StreamsClient.java | 22 +-
.../org/apache/iggy/client/async/TopicsClient.java | 10 +-
.../org/apache/iggy/client/async/UsersClient.java | 4 +-
.../iggy/client/async/tcp/AsyncIggyTcpClient.java | 333 +++-----------------
.../async/tcp/AsyncIggyTcpClientBuilder.java | 186 ++++++++++++
.../iggy/client/async/tcp/AsyncTcpConnection.java | 17 +-
.../client/async/tcp/ConsumerGroupsTcpClient.java | 4 +-
.../iggy/client/async/tcp/MessagesTcpClient.java | 30 +-
.../iggy/client/async/tcp/StreamsTcpClient.java | 34 +--
.../iggy/client/async/tcp/TopicsTcpClient.java | 60 ++--
.../iggy/client/async/tcp/UsersTcpClient.java | 8 +-
.../blocking/http/ConsumerGroupsHttpClient.java | 5 +-
.../iggy/client/blocking/http/IggyHttpClient.java | 37 ++-
.../blocking/http/IggyHttpClientBuilder.java | 210 +++++++++++++
.../client/blocking/http/InternalHttpClient.java | 24 +-
.../client/blocking/http/SystemHttpClient.java | 3 +-
.../iggy/client/blocking/tcp/IggyTcpClient.java | 338 +++++----------------
.../client/blocking/tcp/IggyTcpClientBuilder.java | 245 +++++++++++++++
.../client/blocking/tcp/InternalTcpClient.java | 15 +-
.../java/org/apache/iggy/config/RetryPolicy.java | 130 ++++++++
.../IggyClient.java => config/package-info.java} | 24 +-
.../exception/IggyAuthenticationException.java | 49 +++
.../iggy/exception/IggyAuthorizationException.java | 49 +++
.../IggyClientException.java} | 32 +-
.../iggy/exception/IggyConflictException.java | 49 +++
.../exception/IggyConnectionClosedException.java | 52 ++++
.../IggyConnectionException.java} | 32 +-
.../IggyEmptyResponseException.java} | 31 +-
.../org/apache/iggy/exception/IggyErrorCode.java | 163 ++++++++++
.../org/apache/iggy/exception/IggyException.java | 58 ++++
.../IggyInvalidArgumentException.java} | 32 +-
.../IggyMalformedResponseException.java} | 29 +-
.../IggyMissingCredentialsException.java} | 34 ++-
.../IggyNotConnectedException.java} | 25 +-
.../IggyOperationNotSupportedException.java | 72 +++++
.../IggyProtocolException.java} | 32 +-
.../exception/IggyResourceNotFoundException.java | 49 +++
.../apache/iggy/exception/IggyServerException.java | 255 ++++++++++++++++
.../IggyTimeoutException.java} | 36 ++-
.../IggyTlsException.java} | 32 +-
.../iggy/exception/IggyValidationException.java | 49 +++
.../package-info.java} | 17 +-
.../org/apache/iggy/identifier/Identifier.java | 5 +-
.../org/apache/iggy/message/BytesMessageId.java | 3 +-
.../java/org/apache/iggy/message/HeaderKind.java | 4 +-
.../java/org/apache/iggy/message/MessageState.java | 4 +-
.../java/org/apache/iggy/message/Partitioning.java | 3 +-
.../org/apache/iggy/serde/BytesSerializer.java | 11 +-
.../apache/iggy/topic/CompressionAlgorithm.java | 4 +-
.../main/java/org/apache/iggy/user/UserStatus.java | 3 +-
.../src/main/resources/iggy-version.properties | 3 +
.../src/test/java/org/apache/iggy/IggyTest.java | 74 ++++-
.../client/async/AsyncClientIntegrationTest.java | 35 ++-
.../iggy/client/async/AsyncPollMessageTest.java | 26 +-
.../async/tcp/AsyncIggyTcpClientBuilderTest.java | 153 +---------
.../blocking/tcp/IggyTcpClientBuilderTest.java | 91 +++---
.../iggy/client/blocking/tcp/TcpClientFactory.java | 12 +-
72 files changed, 2768 insertions(+), 1151 deletions(-)
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
index 0262b1d75..61f279f36 100644
--- 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
@@ -143,7 +143,7 @@ public class IggyPartitionSplitReader<T> {
// Poll messages from Iggy (async with blocking)
PolledMessages polledMessages = asyncClient
.messages()
- .pollMessagesAsync(streamId, topicId, partitionId, consumer, strategy, pollBatchSize, true)
+ .pollMessages(streamId, topicId, partitionId, consumer, strategy, pollBatchSize, true)
.join();
log.info(
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
index 7569ff917..209d4c384 100644
--- 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
@@ -168,15 +168,15 @@ public class IggySource<T> implements Source<T, IggySourceSplit, IggySourceEnume
}
// Create async TCP client using builder pattern
- AsyncIggyTcpClient asyncClient = AsyncIggyTcpClient.builder()
- .host(host)
- .port(port)
- .credentials(connectionConfig.getUsername(), connectionConfig.getPassword())
- .connectionPoolSize(4)
- .build();
+ AsyncIggyTcpClient asyncClient =
+ AsyncIggyTcpClient.builder().host(host).port(port).build();
- // Connect and block (auto-login happens during connect)
+ // Connect and login
asyncClient.connect().join();
+ asyncClient
+ .users()
+ .login(connectionConfig.getUsername(), connectionConfig.getPassword())
+ .join();
return asyncClient;
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
index de24a2af5..2566453a7 100644
--- 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
@@ -157,7 +157,7 @@ public class IggySourceSplitEnumerator implements SplitEnumerator<IggySourceSpli
// Get topic details including partitions
Optional<TopicDetails> topicDetailsOpt =
- asyncClient.topics().getTopicAsync(stream, topic).join();
+ asyncClient.topics().getTopic(stream, topic).join();
if (topicDetailsOpt.isPresent()) {
TopicDetails topicDetails = topicDetailsOpt.get();
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
index 3cf1b6e95..912f57140 100644
--- 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
@@ -78,15 +78,15 @@ class AsyncTcpMessagePollTest {
void setUp() throws Exception {
log.info("Setting up Async TCP Client for polling...");
- // Create client with credentials for auto-login
+ // Create client
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
+ // Connect and login
client.connect().join();
+ client.users().login(USERNAME, PASSWORD).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
@@ -97,7 +97,7 @@ class AsyncTcpMessagePollTest {
void tearDown() {
if (client != null) {
try {
- client.users().logoutAsync().join();
+ client.users().logout().join();
log.info("Logged out successfully");
} catch (RuntimeException e) {
log.warn("Error during logout: {}", e.getMessage());
@@ -125,7 +125,7 @@ class AsyncTcpMessagePollTest {
}
client.messages()
- .sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages)
+ .sendMessages(streamId, topicId, Partitioning.balanced(), messages)
.get();
log.info("Seeded {} test messages", messages.size());
}
@@ -144,7 +144,7 @@ class AsyncTcpMessagePollTest {
Consumer consumer = Consumer.of(1L);
CompletableFuture<PolledMessages> pollFuture = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
streamId,
topicId,
Optional.of(1L), // partition 1
@@ -182,7 +182,7 @@ class AsyncTcpMessagePollTest {
Consumer consumer = Consumer.of(2L);
CompletableFuture<PolledMessages> pollFuture = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
streamId,
topicId,
Optional.of(1L), // partition 1
@@ -219,7 +219,7 @@ class AsyncTcpMessagePollTest {
Consumer consumer = Consumer.of(3L);
CompletableFuture<PolledMessages> pollFuture = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
streamId,
topicId,
Optional.empty(), // all partitions
@@ -251,7 +251,7 @@ class AsyncTcpMessagePollTest {
Consumer consumer = Consumer.of(4L);
PolledMessages polledMessages = client.messages()
- .pollMessagesAsync(streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 3L, false)
+ .pollMessages(streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 3L, false)
.get();
log.info("Polled {} messages from offset {}", polledMessages.messages().size(), specificOffset);
@@ -276,11 +276,11 @@ class AsyncTcpMessagePollTest {
// Poll from partition 1
CompletableFuture<PolledMessages> future1 =
- client.messages().pollMessagesAsync(streamId, topicId, Optional.of(1L), consumer, strategy, 5L, false);
+ client.messages().pollMessages(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);
+ client.messages().pollMessages(streamId, topicId, Optional.of(2L), consumer, strategy, 5L, false);
// Wait for both
CompletableFuture.allOf(future1, future2).get();
@@ -309,7 +309,7 @@ class AsyncTcpMessagePollTest {
Consumer consumer = Consumer.of(6L);
PolledMessages polledMessages = client.messages()
- .pollMessagesAsync(streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 3L, false)
+ .pollMessages(streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 3L, false)
.get();
log.info(
@@ -339,7 +339,7 @@ class AsyncTcpMessagePollTest {
Consumer consumer = Consumer.of(7L);
PolledMessages polledMessages = client.messages()
- .pollMessagesAsync(streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 3L, false)
+ .pollMessages(streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 3L, false)
.get();
log.info(
@@ -373,7 +373,7 @@ class AsyncTcpMessagePollTest {
PollingStrategy pollingStrategy = PollingStrategy.next();
PolledMessages polledMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
streamId, topicId, Optional.of(1L), consumer, pollingStrategy, 5L, true // autoCommit
)
.get();
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
index 6527e2b6b..6e31bef4f 100644
--- 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
@@ -72,15 +72,15 @@ class AsyncTcpMessageSendTest {
void setUp() {
log.info("Setting up Async TCP Client...");
- // Create client with credentials for auto-login
+ // Create client
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
+ // Connect and login
client.connect().join();
+ client.users().login(USERNAME, PASSWORD).join();
log.info("Connected to Iggy server at {}:{}", IGGY_SERVER_HOST, IGGY_SERVER_TCP_PORT);
}
@@ -88,7 +88,7 @@ class AsyncTcpMessageSendTest {
void tearDown() {
if (client != null) {
try {
- client.users().logoutAsync().join();
+ client.users().logout().join();
log.info("Logged out successfully");
} catch (RuntimeException e) {
log.warn("Error during logout: {}", e.getMessage());
@@ -119,7 +119,7 @@ class AsyncTcpMessageSendTest {
messages.add(message);
CompletableFuture<Void> sendFuture =
- client.messages().sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages);
+ client.messages().sendMessages(streamId, topicId, Partitioning.balanced(), messages);
sendFuture.get();
@@ -145,7 +145,7 @@ class AsyncTcpMessageSendTest {
}
CompletableFuture<Void> sendFuture =
- client.messages().sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages);
+ client.messages().sendMessages(streamId, topicId, Partitioning.balanced(), messages);
sendFuture.get();
log.info("Successfully sent batch of {} messages", batchSize);
@@ -168,8 +168,8 @@ class AsyncTcpMessageSendTest {
List<Message> messages = new ArrayList<>();
messages.add(message);
- CompletableFuture<Void> sendFuture = client.messages()
- .sendMessagesAsync(streamId, topicId, Partitioning.partitionId(targetPartition), messages);
+ CompletableFuture<Void> sendFuture =
+ client.messages().sendMessages(streamId, topicId, Partitioning.partitionId(targetPartition), messages);
sendFuture.get();
log.info("Successfully sent message to partition {}", targetPartition);
@@ -193,7 +193,7 @@ class AsyncTcpMessageSendTest {
String messageKey = "test-key-123";
CompletableFuture<Void> sendFuture =
- client.messages().sendMessagesAsync(streamId, topicId, Partitioning.messagesKey(messageKey), messages);
+ client.messages().sendMessages(streamId, topicId, Partitioning.messagesKey(messageKey), messages);
sendFuture.get();
log.info("Successfully sent message with key: {}", messageKey);
@@ -217,7 +217,7 @@ class AsyncTcpMessageSendTest {
messages.add(message);
CompletableFuture<Void> sendFuture =
- client.messages().sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages);
+ client.messages().sendMessages(streamId, topicId, Partitioning.balanced(), messages);
sendFuture.get();
log.info("Successfully sent JSON message: {}", jsonPayload);
@@ -243,7 +243,7 @@ class AsyncTcpMessageSendTest {
messages.add(message);
CompletableFuture<Void> future =
- client.messages().sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages);
+ client.messages().sendMessages(streamId, topicId, Partitioning.balanced(), messages);
futures.add(future);
}
@@ -272,7 +272,7 @@ class AsyncTcpMessageSendTest {
long startTime = System.currentTimeMillis();
CompletableFuture<Void> sendFuture =
- client.messages().sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages);
+ client.messages().sendMessages(streamId, topicId, Partitioning.balanced(), messages);
sendFuture.get();
long duration = System.currentTimeMillis() - startTime;
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
index ad9d82465..193aa814c 100644
--- 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
@@ -57,10 +57,10 @@ class SendTextDataTest {
client = AsyncIggyTcpClient.builder()
.host(IGGY_SERVER_HOST)
.port(IGGY_SERVER_TCP_PORT)
- .credentials(USERNAME, PASSWORD)
.build();
client.connect().join();
+ client.users().login(USERNAME, PASSWORD).join();
log.info("Connected to Iggy server at {}:{}", IGGY_SERVER_HOST, IGGY_SERVER_TCP_PORT);
}
@@ -68,7 +68,7 @@ class SendTextDataTest {
void tearDown() {
if (client != null) {
try {
- client.users().logoutAsync().join();
+ client.users().logout().join();
log.info("Logged out successfully");
} catch (RuntimeException e) {
log.warn("Error during logout: {}", e.getMessage());
@@ -109,7 +109,7 @@ class SendTextDataTest {
}
client.messages()
- .sendMessagesAsync(streamId, topicId, Partitioning.balanced(), messages)
+ .sendMessages(streamId, topicId, Partitioning.balanced(), messages)
.get();
log.info("Successfully sent {} text lines to stream 'text-input', topic 'lines'", messages.size());
@@ -124,7 +124,7 @@ class SendTextDataTest {
}
client.messages()
- .sendMessagesAsync(streamId, topicId, Partitioning.balanced(), batchMessages)
+ .sendMessages(streamId, topicId, Partitioning.balanced(), batchMessages)
.get();
log.info("Sent batch {} with {} messages", batch, batchMessages.size());
diff --git a/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/consumer/IggyPartitionGroupConsumer.java b/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/consumer/IggyPartitionGroupConsumer.java
index 37ba72fc0..788aded8b 100644
--- a/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/consumer/IggyPartitionGroupConsumer.java
+++ b/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/consumer/IggyPartitionGroupConsumer.java
@@ -128,12 +128,14 @@ public class IggyPartitionGroupConsumer implements PartitionGroupConsumer {
asyncClient = AsyncIggyTcpClient.builder()
.host(config.getHost())
.port(config.getPort())
- .credentials(config.getUsername(), config.getPassword())
- .connectionPoolSize(config.getConnectionPoolSize())
.build();
// Connect and authenticate
asyncClient.connect().join();
+ asyncClient
+ .users()
+ .login(config.getUsername(), config.getPassword())
+ .join();
// Parse stream and topic IDs
streamId = parseStreamId(config.getStreamId());
@@ -180,7 +182,7 @@ public class IggyPartitionGroupConsumer implements PartitionGroupConsumer {
// Poll with auto-commit disabled (we'll manage offsets via Pinot)
PolledMessages polledMessages = asyncClient
.messages()
- .pollMessagesAsync(
+ .pollMessages(
streamId,
topicId,
partition,
diff --git a/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/metadata/IggyStreamMetadataProvider.java b/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/metadata/IggyStreamMetadataProvider.java
index 36ef0aa6e..750f2885f 100644
--- a/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/metadata/IggyStreamMetadataProvider.java
+++ b/foreign/java/external-processors/iggy-connector-pinot/src/main/java/org/apache/iggy/connector/pinot/metadata/IggyStreamMetadataProvider.java
@@ -158,12 +158,14 @@ public class IggyStreamMetadataProvider implements StreamMetadataProvider {
asyncClient = AsyncIggyTcpClient.builder()
.host(config.getHost())
.port(config.getPort())
- .credentials(config.getUsername(), config.getPassword())
- .connectionPoolSize(config.getConnectionPoolSize())
.build();
// Connect and authenticate
asyncClient.connect().join();
+ asyncClient
+ .users()
+ .login(config.getUsername(), config.getPassword())
+ .join();
// Parse stream and topic IDs
streamId = parseStreamId(config.getStreamId());
@@ -181,7 +183,7 @@ public class IggyStreamMetadataProvider implements StreamMetadataProvider {
if (cachedTopicDetails == null || (now - lastDetailsRefresh) > DETAILS_CACHE_MS) {
try {
Optional<TopicDetails> details =
- asyncClient.topics().getTopicAsync(streamId, topicId).join();
+ asyncClient.topics().getTopic(streamId, topicId).join();
cachedTopicDetails =
details.orElseThrow(() -> new RuntimeException("Topic not found: " + config.getTopicId()));
lastDetailsRefresh = now;
diff --git a/foreign/java/java-sdk/build.gradle.kts b/foreign/java/java-sdk/build.gradle.kts
index 6197e5e07..4434f5336 100644
--- a/foreign/java/java-sdk/build.gradle.kts
+++ b/foreign/java/java-sdk/build.gradle.kts
@@ -17,10 +17,34 @@
* under the License.
*/
+import java.time.Instant
+
plugins {
id("iggy.java-library-conventions")
}
+val gitCommit: String by lazy {
+ try {
+ val process = ProcessBuilder("git", "rev-parse", "--short", "HEAD")
+ .directory(projectDir)
+ .redirectErrorStream(true)
+ .start()
+ process.inputStream.bufferedReader().readText().trim()
+ } catch (e: Exception) {
+ "unknown"
+ }
+}
+
+tasks.processResources {
+ filesMatching("iggy-version.properties") {
+ expand(
+ "version" to project.version,
+ "buildTime" to Instant.now().toString(),
+ "gitCommit" to gitCommit
+ )
+ }
+}
+
dependencies {
implementation(libs.httpclient5)
implementation(libs.jackson.databind)
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/Iggy.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/Iggy.java
index ac983f564..492847fa8 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/Iggy.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/Iggy.java
@@ -19,13 +19,135 @@
package org.apache.iggy;
-import org.apache.iggy.client.blocking.IggyClientBuilder;
+import org.apache.iggy.builder.HttpClientBuilder;
+import org.apache.iggy.builder.TcpClientBuilder;
+import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
+/**
+ * Main entry point for creating Iggy clients.
+ *
+ * <p>Iggy provides a fluent API for creating clients with protocol-first design:
+ *
+ * <h2>TCP Clients (recommended for performance)</h2>
+ * <pre>{@code
+ * // Blocking TCP client
+ * var client = Iggy.tcp().blocking()
+ * .host("localhost")
+ * .port(8090)
+ * .build();
+ * client.connect();
+ * client.users().login("iggy", "iggy");
+ *
+ * // Async TCP client
+ * var asyncClient = Iggy.tcp().async()
+ * .host("localhost")
+ * .build();
+ * asyncClient.connect().join();
+ * asyncClient.users().login("iggy", "iggy").join();
+ * }</pre>
+ *
+ * <h2>HTTP Clients</h2>
+ * <pre>{@code
+ * var httpClient = Iggy.http().blocking()
+ * .url("http://localhost:3000")
+ * .build();
+ *
+ * // Login after creating the client
+ * httpClient.users().login("iggy", "iggy");
+ * }</pre>
+ *
+ * <h2>Quick Factory Methods</h2>
+ * <pre>{@code
+ * // Local TCP client (localhost:8090)
+ * var client = Iggy.localTcp();
+ *
+ * // Local HTTP client (localhost:3000)
+ * var httpClient = Iggy.localHttp();
+ * }</pre>
+ *
+ * <h2>Version Information</h2>
+ * <pre>{@code
+ * String version = Iggy.version(); // e.g., "1.0.0"
+ * IggyVersion info = Iggy.versionInfo(); // Full version details
+ * }</pre>
+ *
+ * @see org.apache.iggy.builder.TcpClientBuilder
+ * @see org.apache.iggy.builder.HttpClientBuilder
+ * @see IggyVersion
+ */
public final class Iggy {
+ private static final String DEFAULT_HOST = "localhost";
+ private static final int DEFAULT_TCP_PORT = 8090;
+ private static final int DEFAULT_HTTP_PORT = 3000;
+
private Iggy() {}
- public static IggyClientBuilder clientBuilder() {
- return new IggyClientBuilder();
+ /**
+ * Creates a builder for TCP clients.
+ *
+ * <p>TCP provides the best performance and is recommended for most use cases.
+ *
+ * @return a TCP client builder
+ */
+ public static TcpClientBuilder tcp() {
+ return new TcpClientBuilder();
+ }
+
+ /**
+ * Creates a builder for HTTP clients.
+ *
+ * <p>HTTP is useful when TCP is blocked by firewalls or when HTTP semantics are preferred.
+ *
+ * @return an HTTP client builder
+ */
+ public static HttpClientBuilder http() {
+ return new HttpClientBuilder();
+ }
+
+ /**
+ * Creates a local blocking TCP client connected to localhost:8090.
+ *
+ * <p>This is a convenience method for local development and testing.
+ * Call {@code client.users().login(username, password)} to authenticate.
+ *
+ * @return a connected IggyTcpClient
+ */
+ public static IggyTcpClient localTcp() {
+ IggyTcpClient client =
+ tcp().blocking().host(DEFAULT_HOST).port(DEFAULT_TCP_PORT).build();
+ client.connect();
+ return client;
+ }
+
+ /**
+ * Creates a local blocking HTTP client connected to localhost:3000.
+ *
+ * <p>This is a convenience method for local development and testing.
+ * Call {@code client.users().login(username, password)} to authenticate.
+ *
+ * @return an IggyHttpClient
+ */
+ public static IggyHttpClient localHttp() {
+ return http().blocking().host(DEFAULT_HOST).port(DEFAULT_HTTP_PORT).build();
+ }
+
+ /**
+ * Returns the SDK version string.
+ *
+ * @return the version string (e.g., "1.0.0")
+ */
+ public static String version() {
+ return IggyVersion.getInstance().getVersion();
+ }
+
+ /**
+ * Returns detailed version information.
+ *
+ * @return the version information object
+ */
+ public static IggyVersion versionInfo() {
+ return IggyVersion.getInstance();
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/IggyVersion.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/IggyVersion.java
new file mode 100644
index 000000000..a2c4a308d
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/IggyVersion.java
@@ -0,0 +1,131 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+/**
+ * Provides version information for the Iggy Java SDK.
+ *
+ * <p>Version information is read from a properties file generated at build time.
+ */
+public final class IggyVersion {
+
+ private static final String PROPERTIES_FILE = "/iggy-version.properties";
+ private static final String UNKNOWN = "unknown";
+
+ private static final IggyVersion INSTANCE;
+
+ static {
+ String version = UNKNOWN;
+ String buildTime = UNKNOWN;
+ String gitCommit = UNKNOWN;
+
+ try (InputStream is = IggyVersion.class.getResourceAsStream(PROPERTIES_FILE)) {
+ if (is != null) {
+ Properties props = new Properties();
+ props.load(is);
+ version = props.getProperty("version", UNKNOWN);
+ buildTime = props.getProperty("buildTime", UNKNOWN);
+ gitCommit = props.getProperty("gitCommit", UNKNOWN);
+ }
+ } catch (IOException e) {
+ // Use default values
+ }
+
+ INSTANCE = new IggyVersion(version, buildTime, gitCommit);
+ }
+
+ private final String version;
+ private final String buildTime;
+ private final String gitCommit;
+
+ private IggyVersion(String version, String buildTime, String gitCommit) {
+ this.version = version;
+ this.buildTime = buildTime;
+ this.gitCommit = gitCommit;
+ }
+
+ /**
+ * Gets the singleton IggyVersion instance.
+ *
+ * @return the version information instance
+ */
+ public static IggyVersion getInstance() {
+ return INSTANCE;
+ }
+
+ /**
+ * Gets the SDK version string.
+ *
+ * @return the version string (e.g., "1.0.0")
+ */
+ public String getVersion() {
+ return version;
+ }
+
+ /**
+ * Gets the build timestamp.
+ *
+ * @return the build time as ISO-8601 string, or "unknown" if not available
+ */
+ public String getBuildTime() {
+ return buildTime;
+ }
+
+ /**
+ * Gets the Git commit hash.
+ *
+ * @return the short Git commit hash, or "unknown" if not available
+ */
+ public String getGitCommit() {
+ return gitCommit;
+ }
+
+ /**
+ * Gets a User-Agent string suitable for HTTP requests.
+ *
+ * @return the User-Agent string (e.g., "iggy-java-sdk/1.0.0")
+ */
+ public String getUserAgent() {
+ return "iggy-java-sdk/" + version;
+ }
+
+ /**
+ * Returns a formatted version string including all available information.
+ *
+ * @return formatted version string
+ */
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Iggy Java SDK ").append(version);
+ if (!UNKNOWN.equals(buildTime)) {
+ sb.append(" (built: ").append(buildTime);
+ if (!UNKNOWN.equals(gitCommit)) {
+ sb.append(", commit: ").append(gitCommit);
+ }
+ sb.append(")");
+ }
+ return sb.toString();
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/HttpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/HttpClientBuilder.java
new file mode 100644
index 000000000..8dfefb9d5
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/HttpClientBuilder.java
@@ -0,0 +1,66 @@
+/*
+ * 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.builder;
+
+import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.client.blocking.http.IggyHttpClientBuilder;
+
+/**
+ * Entry point for building HTTP clients.
+ *
+ * <p>Use this builder to create blocking HTTP clients:
+ * <pre>{@code
+ * // HTTP client - manual login
+ * var client = Iggy.http().blocking()
+ * .url("http://localhost:3000")
+ * .build();
+ * client.users().login("iggy", "iggy");
+ *
+ * // HTTP client - convenience method with auto-login
+ * var client = Iggy.http().blocking()
+ * .host("localhost")
+ * .port(3000)
+ * .credentials("iggy", "iggy")
+ * .buildAndLogin();
+ *
+ * // HTTPS client with auto-login
+ * var secureClient = Iggy.http().blocking()
+ * .host("iggy-server.example.com")
+ * .port(443)
+ * .enableTls()
+ * .credentials("admin", "secret")
+ * .buildAndLogin();
+ * }</pre>
+ *
+ * @see IggyHttpClientBuilder
+ */
+public final class HttpClientBuilder {
+
+ public HttpClientBuilder() {}
+
+ /**
+ * Returns a builder for creating a blocking HTTP client.
+ *
+ * @return a new IggyHttpClientBuilder instance
+ */
+ public IggyHttpClientBuilder blocking() {
+ return IggyHttpClient.builder();
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/TcpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/TcpClientBuilder.java
new file mode 100644
index 000000000..235be3938
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/TcpClientBuilder.java
@@ -0,0 +1,86 @@
+/*
+ * 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.builder;
+
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClient;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClientBuilder;
+import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
+import org.apache.iggy.client.blocking.tcp.IggyTcpClientBuilder;
+
+/**
+ * Entry point for building TCP clients.
+ *
+ * <p>Use this builder to choose between blocking and async TCP clients:
+ * <pre>{@code
+ * // Blocking TCP client - manual connect and login
+ * var client = Iggy.tcp().blocking()
+ * .host("localhost")
+ * .port(8090)
+ * .build();
+ * client.connect();
+ * client.users().login("iggy", "iggy");
+ *
+ * // Blocking TCP client - convenience method with auto-login
+ * var client = Iggy.tcp().blocking()
+ * .host("localhost")
+ * .port(8090)
+ * .credentials("iggy", "iggy")
+ * .buildAndLogin();
+ *
+ * // Async TCP client - manual connect and login
+ * var asyncClient = Iggy.tcp().async()
+ * .host("localhost")
+ * .build();
+ * asyncClient.connect().join();
+ * asyncClient.users().login("iggy", "iggy").join();
+ *
+ * // Async TCP client - convenience method with auto-login
+ * var asyncClient = Iggy.tcp().async()
+ * .host("localhost")
+ * .credentials("iggy", "iggy")
+ * .buildAndLogin()
+ * .join();
+ * }</pre>
+ *
+ * @see IggyTcpClientBuilder
+ * @see AsyncIggyTcpClientBuilder
+ */
+public final class TcpClientBuilder {
+
+ public TcpClientBuilder() {}
+
+ /**
+ * Returns a builder for creating a blocking TCP client.
+ *
+ * @return a new IggyTcpClientBuilder instance
+ */
+ public IggyTcpClientBuilder blocking() {
+ return IggyTcpClient.builder();
+ }
+
+ /**
+ * Returns a builder for creating an async TCP client.
+ *
+ * @return a new AsyncIggyTcpClientBuilder instance
+ */
+ public AsyncIggyTcpClientBuilder async() {
+ return AsyncIggyTcpClient.builder();
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/package-info.java
similarity index 63%
rename from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyClientBuilder.java
rename to foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/package-info.java
index f7393f756..5348e0d1d 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyClientBuilder.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/builder/package-info.java
@@ -17,21 +17,18 @@
* under the License.
*/
-package org.apache.iggy.client.blocking;
-
-public class IggyClientBuilder {
-
- private IggyBaseClient client;
-
- public IggyClientBuilder withBaseClient(IggyBaseClient client) {
- this.client = client;
- return this;
- }
+/**
+ * Fluent builders for creating Iggy client instances.
+ *
+ * <p>This package provides the builder API accessed through {@link org.apache.iggy.Iggy}:
+ * <ul>
+ * <li>{@link org.apache.iggy.builder.TcpClientBuilder} - Entry point for TCP clients</li>
+ * <li>{@link org.apache.iggy.builder.HttpClientBuilder} - Entry point for HTTP clients</li>
+ * </ul>
+ *
+ * @see org.apache.iggy.Iggy
+ */
+@NonNullApi
+package org.apache.iggy.builder;
- public IggyClient build() {
- if (client == null) {
- throw new IllegalArgumentException("Base client not provided");
- }
- return new IggyClient(client);
- }
-}
+import org.apache.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
index 78415dc1b..2654e87c6 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
@@ -49,7 +49,7 @@ public interface MessagesClient {
* @param autoCommit whether to auto-commit offsets
* @return CompletableFuture that will complete with the polled messages
*/
- CompletableFuture<PolledMessages> pollMessagesAsync(
+ CompletableFuture<PolledMessages> pollMessages(
StreamId streamId,
TopicId topicId,
Optional<Long> partitionId,
@@ -70,7 +70,7 @@ public interface MessagesClient {
* @param autoCommit whether to auto-commit offsets
* @return CompletableFuture that will complete with the polled messages
*/
- default CompletableFuture<PolledMessages> pollMessagesAsync(
+ default CompletableFuture<PolledMessages> pollMessages(
Long streamId,
Long topicId,
Optional<Long> partitionId,
@@ -78,7 +78,7 @@ public interface MessagesClient {
PollingStrategy strategy,
Long count,
boolean autoCommit) {
- return pollMessagesAsync(
+ return pollMessages(
StreamId.of(streamId),
TopicId.of(topicId),
partitionId,
@@ -97,7 +97,7 @@ public interface MessagesClient {
* @param messages the messages to send
* @return CompletableFuture that will complete when messages are sent
*/
- CompletableFuture<Void> sendMessagesAsync(
+ CompletableFuture<Void> sendMessages(
StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages);
/**
@@ -109,8 +109,8 @@ public interface MessagesClient {
* @param messages the messages to send
* @return CompletableFuture that will complete when messages are sent
*/
- default CompletableFuture<Void> sendMessagesAsync(
+ default CompletableFuture<Void> sendMessages(
Long streamId, Long topicId, Partitioning partitioning, List<Message> messages) {
- return sendMessagesAsync(StreamId.of(streamId), TopicId.of(topicId), partitioning, messages);
+ return sendMessages(StreamId.of(streamId), TopicId.of(topicId), partitioning, messages);
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/StreamsClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/StreamsClient.java
index 73eabe7ed..f2cb77e10 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/StreamsClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/StreamsClient.java
@@ -29,25 +29,25 @@ import java.util.concurrent.CompletableFuture;
public interface StreamsClient {
- default CompletableFuture<Optional<StreamDetails>> getStreamAsync(Long streamId) {
- return getStreamAsync(StreamId.of(streamId));
+ default CompletableFuture<Optional<StreamDetails>> getStream(Long streamId) {
+ return getStream(StreamId.of(streamId));
}
- CompletableFuture<Optional<StreamDetails>> getStreamAsync(StreamId streamId);
+ CompletableFuture<Optional<StreamDetails>> getStream(StreamId streamId);
- CompletableFuture<List<StreamBase>> getStreamsAsync();
+ CompletableFuture<List<StreamBase>> getStreams();
- CompletableFuture<StreamDetails> createStreamAsync(String name);
+ CompletableFuture<StreamDetails> createStream(String name);
- default CompletableFuture<Void> updateStreamAsync(Long streamId, String name) {
- return updateStreamAsync(StreamId.of(streamId), name);
+ default CompletableFuture<Void> updateStream(Long streamId, String name) {
+ return updateStream(StreamId.of(streamId), name);
}
- CompletableFuture<Void> updateStreamAsync(StreamId streamId, String name);
+ CompletableFuture<Void> updateStream(StreamId streamId, String name);
- default CompletableFuture<Void> deleteStreamAsync(Long streamId) {
- return deleteStreamAsync(StreamId.of(streamId));
+ default CompletableFuture<Void> deleteStream(Long streamId) {
+ return deleteStream(StreamId.of(streamId));
}
- CompletableFuture<Void> deleteStreamAsync(StreamId streamId);
+ CompletableFuture<Void> deleteStream(StreamId streamId);
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/TopicsClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/TopicsClient.java
index 5b7ffce4b..a976ad65c 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/TopicsClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/TopicsClient.java
@@ -42,7 +42,7 @@ public interface TopicsClient {
* @param topicId The topic identifier
* @return CompletableFuture with Optional TopicDetails
*/
- CompletableFuture<Optional<TopicDetails>> getTopicAsync(StreamId streamId, TopicId topicId);
+ CompletableFuture<Optional<TopicDetails>> getTopic(StreamId streamId, TopicId topicId);
/**
* Gets all topics in a stream.
@@ -50,7 +50,7 @@ public interface TopicsClient {
* @param streamId The stream identifier
* @return CompletableFuture with list of Topics
*/
- CompletableFuture<List<Topic>> getTopicsAsync(StreamId streamId);
+ CompletableFuture<List<Topic>> getTopics(StreamId streamId);
/**
* Creates a new topic.
@@ -64,7 +64,7 @@ public interface TopicsClient {
* @param name Topic name
* @return CompletableFuture with created TopicDetails
*/
- CompletableFuture<TopicDetails> createTopicAsync(
+ CompletableFuture<TopicDetails> createTopic(
StreamId streamId,
Long partitionsCount,
CompressionAlgorithm compressionAlgorithm,
@@ -85,7 +85,7 @@ public interface TopicsClient {
* @param name Topic name
* @return CompletableFuture that completes when update is done
*/
- CompletableFuture<Void> updateTopicAsync(
+ CompletableFuture<Void> updateTopic(
StreamId streamId,
TopicId topicId,
CompressionAlgorithm compressionAlgorithm,
@@ -101,5 +101,5 @@ public interface TopicsClient {
* @param topicId The topic identifier
* @return CompletableFuture that completes when deletion is done
*/
- CompletableFuture<Void> deleteTopicAsync(StreamId streamId, TopicId topicId);
+ CompletableFuture<Void> deleteTopic(StreamId streamId, TopicId topicId);
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
index 3dfc40792..db39f7d2c 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
@@ -35,12 +35,12 @@ public interface UsersClient {
* @param password The password to login with
* @return A CompletableFuture that completes with the user's identity information
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ CompletableFuture<IdentityInfo> login(String username, String password);
/**
* Logs out from the server.
*
* @return A CompletableFuture that completes when logout is successful
*/
- CompletableFuture<Void> logoutAsync();
+ CompletableFuture<Void> logout();
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
index b0b1d30ea..c75e0b472 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
@@ -19,15 +19,16 @@
package org.apache.iggy.client.async.tcp;
-import org.apache.commons.lang3.StringUtils;
import org.apache.iggy.client.async.ConsumerGroupsClient;
import org.apache.iggy.client.async.MessagesClient;
import org.apache.iggy.client.async.StreamsClient;
import org.apache.iggy.client.async.TopicsClient;
import org.apache.iggy.client.async.UsersClient;
+import org.apache.iggy.exception.IggyMissingCredentialsException;
+import org.apache.iggy.exception.IggyNotConnectedException;
+import org.apache.iggy.user.IdentityInfo;
import java.io.File;
-import java.time.Duration;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -39,14 +40,10 @@ public class AsyncIggyTcpClient {
private final String host;
private final int port;
- private final Optional<String> username;
- private final Optional<String> password;
- private final Optional<Duration> connectionTimeout;
- private final Optional<Duration> requestTimeout;
- private final Optional<Integer> connectionPoolSize;
- private final Optional<RetryPolicy> retryPolicy;
private final boolean enableTls;
private final Optional<File> tlsCertificate;
+ private final Optional<String> username;
+ private final Optional<String> password;
private AsyncTcpConnection connection;
private MessagesClient messagesClient;
private ConsumerGroupsClient consumerGroupsClient;
@@ -55,29 +52,15 @@ public class AsyncIggyTcpClient {
private UsersClient usersClient;
public AsyncIggyTcpClient(String host, int port) {
- this(host, port, null, null, null, null, null, null, false, Optional.empty());
+ this(host, port, null, null, false, Optional.empty());
}
- @SuppressWarnings("checkstyle:ParameterNumber")
- private AsyncIggyTcpClient(
- String host,
- int port,
- String username,
- String password,
- Duration connectionTimeout,
- Duration requestTimeout,
- Integer connectionPoolSize,
- RetryPolicy retryPolicy,
- boolean enableTls,
- Optional<File> tlsCertificate) {
+ AsyncIggyTcpClient(
+ String host, int port, String username, String password, boolean enableTls, Optional<File> tlsCertificate) {
this.host = host;
this.port = port;
this.username = Optional.ofNullable(username);
this.password = Optional.ofNullable(password);
- this.connectionTimeout = Optional.ofNullable(connectionTimeout);
- this.requestTimeout = Optional.ofNullable(requestTimeout);
- this.connectionPoolSize = Optional.ofNullable(connectionPoolSize);
- this.retryPolicy = Optional.ofNullable(retryPolicy);
this.enableTls = enableTls;
this.tlsCertificate = tlsCertificate;
}
@@ -87,33 +70,41 @@ public class AsyncIggyTcpClient {
*
* @return a new Builder instance
*/
- public static Builder builder() {
- return new Builder();
+ public static AsyncIggyTcpClientBuilder builder() {
+ return new AsyncIggyTcpClientBuilder();
}
/**
* Connects to the Iggy server asynchronously.
+ *
+ * @return a CompletableFuture that completes when connected
*/
public CompletableFuture<Void> connect() {
connection = new AsyncTcpConnection(host, port, enableTls, tlsCertificate);
- return connection
- .connect()
- .thenRun(() -> {
- messagesClient = new MessagesTcpClient(connection);
- consumerGroupsClient = new ConsumerGroupsTcpClient(connection);
- streamsClient = new StreamsTcpClient(connection);
- topicsClient = new TopicsTcpClient(connection);
- usersClient = new UsersTcpClient(connection);
- })
- .thenCompose(v -> {
- // Auto-login if credentials are provided
- if (username.isPresent() && password.isPresent()) {
- return usersClient
- .loginAsync(username.get(), password.get())
- .thenApply(identity -> null);
- }
- return CompletableFuture.completedFuture(null);
- });
+ return connection.connect().thenRun(() -> {
+ messagesClient = new MessagesTcpClient(connection);
+ consumerGroupsClient = new ConsumerGroupsTcpClient(connection);
+ streamsClient = new StreamsTcpClient(connection);
+ topicsClient = new TopicsTcpClient(connection);
+ usersClient = new UsersTcpClient(connection);
+ });
+ }
+
+ /**
+ * Logs in using the credentials provided during client construction.
+ *
+ * @return a CompletableFuture that completes when logged in
+ * @throws IggyMissingCredentialsException if no credentials were provided
+ * @throws IggyNotConnectedException if client is not connected
+ */
+ public CompletableFuture<IdentityInfo> login() {
+ if (usersClient == null) {
+ throw new IggyNotConnectedException();
+ }
+ if (username.isEmpty() || password.isEmpty()) {
+ throw new IggyMissingCredentialsException();
+ }
+ return usersClient.login(username.get(), password.get());
}
/**
@@ -121,7 +112,7 @@ public class AsyncIggyTcpClient {
*/
public UsersClient users() {
if (usersClient == null) {
- throw new IllegalStateException("Client not connected. Call connect() first.");
+ throw new IggyNotConnectedException();
}
return usersClient;
}
@@ -131,7 +122,7 @@ public class AsyncIggyTcpClient {
*/
public MessagesClient messages() {
if (messagesClient == null) {
- throw new IllegalStateException("Client not connected. Call connect() first.");
+ throw new IggyNotConnectedException();
}
return messagesClient;
}
@@ -141,7 +132,7 @@ public class AsyncIggyTcpClient {
*/
public ConsumerGroupsClient consumerGroups() {
if (consumerGroupsClient == null) {
- throw new IllegalStateException("Client not connected. Call connect() first.");
+ throw new IggyNotConnectedException();
}
return consumerGroupsClient;
}
@@ -151,7 +142,7 @@ public class AsyncIggyTcpClient {
*/
public StreamsClient streams() {
if (streamsClient == null) {
- throw new IllegalStateException("Client not connected. Call connect() first.");
+ throw new IggyNotConnectedException();
}
return streamsClient;
}
@@ -161,7 +152,7 @@ public class AsyncIggyTcpClient {
*/
public TopicsClient topics() {
if (topicsClient == null) {
- throw new IllegalStateException("Client not connected. Call connect() first.");
+ throw new IggyNotConnectedException();
}
return topicsClient;
}
@@ -175,246 +166,4 @@ public class AsyncIggyTcpClient {
}
return CompletableFuture.completedFuture(null);
}
-
- /**
- * Builder for creating configured AsyncIggyTcpClient instances.
- */
- public static final class Builder {
- private String host = "localhost";
- private Integer port = 8090;
- private String username;
- private String password;
- private Duration connectionTimeout;
- private Duration requestTimeout;
- private Integer connectionPoolSize;
- private RetryPolicy retryPolicy;
- private boolean enableTls = false;
- private File tlsTrustedCertificatePem;
-
- private Builder() {}
-
- /**
- * Sets the host address for the Iggy server.
- *
- * @param host the host address
- * @return this builder
- */
- public Builder host(String host) {
- this.host = host;
- return this;
- }
-
- /**
- * Sets the port for the Iggy server.
- *
- * @param port the port number
- * @return this builder
- */
- public Builder port(Integer port) {
- this.port = port;
- return this;
- }
-
- /**
- * Sets the credentials for authentication.
- *
- * @param username the username
- * @param password the password
- * @return this builder
- */
- public Builder credentials(String username, String password) {
- this.username = username;
- this.password = password;
- return this;
- }
-
- /**
- * Sets the connection timeout.
- *
- * @param connectionTimeout the connection timeout duration
- * @return this builder
- */
- public Builder connectionTimeout(Duration connectionTimeout) {
- this.connectionTimeout = connectionTimeout;
- return this;
- }
-
- /**
- * Sets the request timeout.
- *
- * @param requestTimeout the request timeout duration
- * @return this builder
- */
- public Builder requestTimeout(Duration requestTimeout) {
- this.requestTimeout = requestTimeout;
- return this;
- }
-
- /**
- * Sets the connection pool size.
- *
- * @param connectionPoolSize the size of the connection pool
- * @return this builder
- */
- public Builder connectionPoolSize(Integer connectionPoolSize) {
- this.connectionPoolSize = connectionPoolSize;
- return this;
- }
-
- /**
- * Sets the retry policy.
- *
- * @param retryPolicy the retry policy to use
- * @return this builder
- */
- public Builder retryPolicy(RetryPolicy retryPolicy) {
- this.retryPolicy = retryPolicy;
- return this;
- }
-
- /**
- * Enables or disables TLS for the TCP connection.
- *
- * @param enableTls whether to enable TLS
- * @return this builder
- */
- public Builder tls(boolean enableTls) {
- this.enableTls = enableTls;
- return this;
- }
-
- /**
- * Enables TLS for the TCP connection.
- *
- * @return this builder
- */
- public Builder enableTls() {
- this.enableTls = true;
- return this;
- }
-
- /**
- * Sets a custom trusted certificate (PEM file) to validate the server certificate.
- *
- * @param certificate the PEM file containing the certificate or CA chain
- * @return this builder
- */
- public Builder tlsTrustedCertificate(File certificate) {
- this.tlsTrustedCertificatePem = certificate;
- return this;
- }
-
- /**
- * Sets a custom trusted certificate (PEM file path) to validate the server certificate.
- *
- * @param certificatePath the PEM file path containing the certificate or CA chain
- * @return this builder
- */
- public Builder tlsTrustedCertificate(String certificatePath) {
- this.tlsTrustedCertificatePem = StringUtils.isBlank(certificatePath) ? null : new File(certificatePath);
- return this;
- }
-
- /**
- * Builds and returns a configured AsyncIggyTcpClient instance.
- * Note: You still need to call connect() on the returned client.
- *
- * @return a new AsyncIggyTcpClient instance
- */
- public AsyncIggyTcpClient build() {
- if (host == null || host.isEmpty()) {
- throw new IllegalArgumentException("Host cannot be null or empty");
- }
- if (port == null || port <= 0) {
- throw new IllegalArgumentException("Port must be a positive integer");
- }
- return new AsyncIggyTcpClient(
- host,
- port,
- username,
- password,
- connectionTimeout,
- requestTimeout,
- connectionPoolSize,
- retryPolicy,
- enableTls,
- Optional.ofNullable(tlsTrustedCertificatePem));
- }
- }
-
- /**
- * Retry policy for client operations.
- */
- public static final class RetryPolicy {
- private final int maxRetries;
- private final Duration initialDelay;
- private final Duration maxDelay;
- private final double multiplier;
-
- private RetryPolicy(int maxRetries, Duration initialDelay, Duration maxDelay, double multiplier) {
- this.maxRetries = maxRetries;
- this.initialDelay = initialDelay;
- this.maxDelay = maxDelay;
- this.multiplier = multiplier;
- }
-
- /**
- * Creates a retry policy with exponential backoff.
- *
- * @return a RetryPolicy with exponential backoff configuration
- */
- public static RetryPolicy exponentialBackoff() {
- return new RetryPolicy(3, Duration.ofMillis(100), Duration.ofSeconds(5), 2.0);
- }
-
- /**
- * Creates a retry policy with exponential backoff and custom parameters.
- *
- * @param maxRetries the maximum number of retries
- * @param initialDelay the initial delay before the first retry
- * @param maxDelay the maximum delay between retries
- * @param multiplier the multiplier for exponential backoff
- * @return a RetryPolicy with custom exponential backoff configuration
- */
- public static RetryPolicy exponentialBackoff(
- int maxRetries, Duration initialDelay, Duration maxDelay, double multiplier) {
- return new RetryPolicy(maxRetries, initialDelay, maxDelay, multiplier);
- }
-
- /**
- * Creates a retry policy with fixed delay.
- *
- * @param maxRetries the maximum number of retries
- * @param delay the fixed delay between retries
- * @return a RetryPolicy with fixed delay configuration
- */
- public static RetryPolicy fixedDelay(int maxRetries, Duration delay) {
- return new RetryPolicy(maxRetries, delay, delay, 1.0);
- }
-
- /**
- * Creates a no-retry policy.
- *
- * @return a RetryPolicy that does not retry
- */
- public static RetryPolicy noRetry() {
- return new RetryPolicy(0, Duration.ZERO, Duration.ZERO, 1.0);
- }
-
- public int getMaxRetries() {
- return maxRetries;
- }
-
- public Duration getInitialDelay() {
- return initialDelay;
- }
-
- public Duration getMaxDelay() {
- return maxDelay;
- }
-
- public double getMultiplier() {
- return multiplier;
- }
- }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java
new file mode 100644
index 000000000..4c36270bf
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilder.java
@@ -0,0 +1,186 @@
+/*
+ * 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.client.async.tcp;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+import org.apache.iggy.exception.IggyMissingCredentialsException;
+
+import java.io.File;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Builder for creating configured AsyncIggyTcpClient instances.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * // Basic usage with explicit connect and login
+ * var client = AsyncIggyTcpClient.builder()
+ * .host("localhost")
+ * .port(8090)
+ * .build();
+ * client.connect().join();
+ * client.users().login("iggy", "iggy").join();
+ *
+ * // Convenience method with auto-login
+ * var client = AsyncIggyTcpClient.builder()
+ * .host("localhost")
+ * .port(8090)
+ * .credentials("iggy", "iggy")
+ * .buildAndLogin()
+ * .join();
+ *
+ * // With TLS enabled
+ * var client = AsyncIggyTcpClient.builder()
+ * .host("iggy-server.example.com")
+ * .port(8090)
+ * .enableTls()
+ * .credentials("admin", "secret")
+ * .buildAndLogin()
+ * .join();
+ * }</pre>
+ *
+ * @see AsyncIggyTcpClient#builder()
+ */
+public final class AsyncIggyTcpClientBuilder {
+ private String host = "localhost";
+ private Integer port = 8090;
+ private String username;
+ private String password;
+ private boolean enableTls = false;
+ private File tlsCertificate;
+
+ AsyncIggyTcpClientBuilder() {}
+
+ /**
+ * Sets the host address for the Iggy server.
+ *
+ * @param host the host address
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder host(String host) {
+ this.host = host;
+ return this;
+ }
+
+ /**
+ * Sets the port for the Iggy server.
+ *
+ * @param port the port number
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder port(Integer port) {
+ this.port = port;
+ return this;
+ }
+
+ /**
+ * Sets the credentials for authentication.
+ * These credentials are stored and can be used with {@link AsyncIggyTcpClient#login()}.
+ *
+ * @param username the username
+ * @param password the password
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder credentials(String username, String password) {
+ this.username = username;
+ this.password = password;
+ return this;
+ }
+
+ /**
+ * Enables or disables TLS for the TCP connection.
+ *
+ * @param enableTls whether to enable TLS
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder tls(boolean enableTls) {
+ this.enableTls = enableTls;
+ return this;
+ }
+
+ /**
+ * Enables TLS for the TCP connection.
+ *
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder enableTls() {
+ this.enableTls = true;
+ return this;
+ }
+
+ /**
+ * Sets a custom trusted certificate (PEM file) to validate the server certificate.
+ *
+ * @param tlsCertificate the PEM file containing the certificate or CA chain
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder tlsCertificate(File tlsCertificate) {
+ this.tlsCertificate = tlsCertificate;
+ return this;
+ }
+
+ /**
+ * Sets a custom trusted certificate (PEM file path) to validate the server certificate.
+ *
+ * @param tlsCertificatePath the PEM file path containing the certificate or CA chain
+ * @return this builder
+ */
+ public AsyncIggyTcpClientBuilder tlsCertificate(String tlsCertificatePath) {
+ this.tlsCertificate = StringUtils.isBlank(tlsCertificatePath) ? null : new File(tlsCertificatePath);
+ return this;
+ }
+
+ /**
+ * Builds and returns a configured AsyncIggyTcpClient instance.
+ * Note: You still need to call {@link AsyncIggyTcpClient#connect()} on the returned client.
+ *
+ * @return a new AsyncIggyTcpClient instance
+ * @throws IggyInvalidArgumentException if the host is null or empty, or if the port is not positive
+ */
+ public AsyncIggyTcpClient build() {
+ if (host == null || host.isEmpty()) {
+ throw new IggyInvalidArgumentException("Host cannot be null or empty");
+ }
+ if (port == null || port <= 0) {
+ throw new IggyInvalidArgumentException("Port must be a positive integer");
+ }
+ return new AsyncIggyTcpClient(host, port, username, password, enableTls, Optional.ofNullable(tlsCertificate));
+ }
+
+ /**
+ * Builds, connects, and logs in using the provided credentials.
+ * This is a convenience method equivalent to calling {@code build()}, {@code connect()},
+ * and {@code login()}.
+ *
+ * @return a CompletableFuture that completes with the connected and logged in client
+ * @throws IggyMissingCredentialsException if no credentials were provided
+ * @throws IggyInvalidArgumentException if the host is null or empty, or if the port is not positive
+ */
+ public CompletableFuture<AsyncIggyTcpClient> buildAndLogin() {
+ if (username == null || password == null) {
+ throw new IggyMissingCredentialsException(
+ "Credentials must be provided to use buildAndLogin(). Use credentials(username, password).");
+ }
+ AsyncIggyTcpClient client = build();
+ return client.connect().thenCompose(v -> client.login()).thenApply(v -> client);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
index 76539d044..d316fcf1c 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
@@ -34,6 +34,9 @@ import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
+import org.apache.iggy.exception.IggyNotConnectedException;
+import org.apache.iggy.exception.IggyServerException;
+import org.apache.iggy.exception.IggyTlsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,7 +82,7 @@ public class AsyncTcpConnection {
this.tlsCertificate.ifPresent(builder::trustManager);
this.sslContext = builder.build();
} catch (SSLException e) {
- throw new RuntimeException("Failed to build SSL context for AsyncTcpConnection", e);
+ throw new IggyTlsException("Failed to build SSL context for AsyncTcpConnection", e);
}
} else {
this.sslContext = null;
@@ -137,10 +140,11 @@ public class AsyncTcpConnection {
* Sends a command asynchronously and returns the response.
* Uses Netty's EventLoop to ensure thread-safe sequential request processing with FIFO response matching.
*/
- public CompletableFuture<ByteBuf> sendAsync(int commandCode, ByteBuf payload) {
+ public CompletableFuture<ByteBuf> send(int commandCode, ByteBuf payload) {
if (channel == null || !channel.isActive()) {
payload.release();
- return CompletableFuture.failedFuture(new IllegalStateException("Connection not established or closed"));
+ return CompletableFuture.failedFuture(
+ new IggyNotConnectedException("Connection not established or closed"));
}
CompletableFuture<ByteBuf> responseFuture = new CompletableFuture<>();
@@ -252,14 +256,11 @@ public class AsyncTcpConnection {
future.complete(msg.retainedSlice());
} else {
// Error - the payload contains the error message
+ byte[] errorBytes = length > 0 ? new byte[length] : new byte[0];
if (length > 0) {
- byte[] errorBytes = new byte[length];
msg.readBytes(errorBytes);
- future.completeExceptionally(
- new RuntimeException("Server error: " + new String(errorBytes)));
- } else {
- future.completeExceptionally(new RuntimeException("Server error with status: " + status));
}
+ future.completeExceptionally(IggyServerException.fromTcpResponse(status, errorBytes));
}
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
index e671539f8..cd0da30c8 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
@@ -59,7 +59,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient {
log.debug("Joining consumer group - Stream: {}, Topic: {}, Group: {}", streamId, topicId, groupId);
return connection
- .sendAsync(CommandCode.ConsumerGroup.JOIN.getValue(), payload)
+ .send(CommandCode.ConsumerGroup.JOIN.getValue(), payload)
.thenAccept(response -> {
log.debug("Successfully joined consumer group");
response.release();
@@ -82,7 +82,7 @@ public class ConsumerGroupsTcpClient implements ConsumerGroupsClient {
log.debug("Leaving consumer group - Stream: {}, Topic: {}, Group: {}", streamId, topicId, groupId);
return connection
- .sendAsync(CommandCode.ConsumerGroup.LEAVE.getValue(), payload)
+ .send(CommandCode.ConsumerGroup.LEAVE.getValue(), payload)
.thenAccept(response -> {
log.debug("Successfully left consumer group");
response.release();
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
index 74fdc49c2..51c84f375 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
@@ -49,7 +49,7 @@ public class MessagesTcpClient implements MessagesClient {
}
@Override
- public CompletableFuture<PolledMessages> pollMessagesAsync(
+ public CompletableFuture<PolledMessages> pollMessages(
StreamId streamId,
TopicId topicId,
Optional<Long> partitionId,
@@ -79,19 +79,17 @@ public class MessagesTcpClient implements MessagesClient {
payload.writeByte(autoCommit ? 1 : 0);
// Send async request and transform response
- return connection
- .sendAsync(CommandCode.Messages.POLL.getValue(), payload)
- .thenApply(response -> {
- try {
- return BytesDeserializer.readPolledMessages(response);
- } finally {
- response.release();
- }
- });
+ return connection.send(CommandCode.Messages.POLL.getValue(), payload).thenApply(response -> {
+ try {
+ return BytesDeserializer.readPolledMessages(response);
+ } finally {
+ response.release();
+ }
+ });
}
@Override
- public CompletableFuture<Void> sendMessagesAsync(
+ public CompletableFuture<Void> sendMessages(
StreamId streamId, TopicId topicId, Partitioning partitioning, List<Message> messages) {
// Build metadata section following the blocking client pattern
@@ -125,11 +123,9 @@ public class MessagesTcpClient implements MessagesClient {
}
// Send async request (no response data expected for send)
- return connection
- .sendAsync(CommandCode.Messages.SEND.getValue(), payload)
- .thenAccept(response -> {
- // Response received, messages sent successfully
- response.release(); // Release the buffer
- });
+ return connection.send(CommandCode.Messages.SEND.getValue(), payload).thenAccept(response -> {
+ // Response received, messages sent successfully
+ response.release(); // Release the buffer
+ });
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
index c9e10ee26..0a0466881 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
@@ -49,26 +49,24 @@ public class StreamsTcpClient implements StreamsClient {
}
@Override
- public CompletableFuture<StreamDetails> createStreamAsync(String name) {
+ public CompletableFuture<StreamDetails> createStream(String name) {
var payloadSize = 1 + name.length();
var payload = Unpooled.buffer(payloadSize);
payload.writeBytes(BytesSerializer.toBytes(name));
- return connection
- .sendAsync(CommandCode.Stream.CREATE.getValue(), payload)
- .thenApply(response -> {
- StreamDetails details = readStreamDetails(response);
- response.release();
- return details;
- });
+ return connection.send(CommandCode.Stream.CREATE.getValue(), payload).thenApply(response -> {
+ StreamDetails details = readStreamDetails(response);
+ response.release();
+ return details;
+ });
}
@Override
- public CompletableFuture<Optional<StreamDetails>> getStreamAsync(StreamId streamId) {
+ public CompletableFuture<Optional<StreamDetails>> getStream(StreamId streamId) {
var payload = toBytes(streamId);
- return connection.sendAsync(CommandCode.Stream.GET.getValue(), payload).thenApply(response -> {
+ return connection.send(CommandCode.Stream.GET.getValue(), payload).thenApply(response -> {
Optional<StreamDetails> result;
if (response.isReadable()) {
result = Optional.of(readStreamDetails(response));
@@ -81,9 +79,9 @@ public class StreamsTcpClient implements StreamsClient {
}
@Override
- public CompletableFuture<List<StreamBase>> getStreamsAsync() {
+ public CompletableFuture<List<StreamBase>> getStreams() {
return connection
- .sendAsync(CommandCode.Stream.GET_ALL.getValue(), Unpooled.EMPTY_BUFFER)
+ .send(CommandCode.Stream.GET_ALL.getValue(), Unpooled.EMPTY_BUFFER)
.thenApply(response -> {
List<StreamBase> streams = new ArrayList<>();
while (response.isReadable()) {
@@ -95,7 +93,7 @@ public class StreamsTcpClient implements StreamsClient {
}
@Override
- public CompletableFuture<Void> updateStreamAsync(StreamId streamId, String name) {
+ public CompletableFuture<Void> updateStream(StreamId streamId, String name) {
var payloadSize = 1 + name.length();
var idBytes = toBytes(streamId);
var payload = Unpooled.buffer(payloadSize + idBytes.capacity());
@@ -103,17 +101,13 @@ public class StreamsTcpClient implements StreamsClient {
payload.writeBytes(idBytes);
payload.writeBytes(BytesSerializer.toBytes(name));
- return connection
- .sendAsync(CommandCode.Stream.UPDATE.getValue(), payload)
- .thenAccept(ReferenceCounted::release);
+ return connection.send(CommandCode.Stream.UPDATE.getValue(), payload).thenAccept(ReferenceCounted::release);
}
@Override
- public CompletableFuture<Void> deleteStreamAsync(StreamId streamId) {
+ public CompletableFuture<Void> deleteStream(StreamId streamId) {
var payload = toBytes(streamId);
- return connection
- .sendAsync(CommandCode.Stream.DELETE.getValue(), payload)
- .thenAccept(ReferenceCounted::release);
+ return connection.send(CommandCode.Stream.DELETE.getValue(), payload).thenAccept(ReferenceCounted::release);
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
index 80e73b0e4..dce5b70f3 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
@@ -51,12 +51,12 @@ public class TopicsTcpClient implements TopicsClient {
}
@Override
- public CompletableFuture<Optional<TopicDetails>> getTopicAsync(StreamId streamId, TopicId topicId) {
+ public CompletableFuture<Optional<TopicDetails>> getTopic(StreamId streamId, TopicId topicId) {
var payload = Unpooled.buffer();
payload.writeBytes(toBytes(streamId));
payload.writeBytes(toBytes(topicId));
- return connection.sendAsync(CommandCode.Topic.GET.getValue(), payload).thenApply(response -> {
+ return connection.send(CommandCode.Topic.GET.getValue(), payload).thenApply(response -> {
try {
if (response.isReadable()) {
return Optional.of(BytesDeserializer.readTopicDetails(response));
@@ -69,26 +69,24 @@ public class TopicsTcpClient implements TopicsClient {
}
@Override
- public CompletableFuture<List<Topic>> getTopicsAsync(StreamId streamId) {
+ public CompletableFuture<List<Topic>> getTopics(StreamId streamId) {
var payload = toBytes(streamId);
- return connection
- .sendAsync(CommandCode.Topic.GET_ALL.getValue(), payload)
- .thenApply(response -> {
- try {
- List<Topic> topics = new ArrayList<>();
- while (response.isReadable()) {
- topics.add(BytesDeserializer.readTopic(response));
- }
- return topics;
- } finally {
- response.release();
- }
- });
+ return connection.send(CommandCode.Topic.GET_ALL.getValue(), payload).thenApply(response -> {
+ try {
+ List<Topic> topics = new ArrayList<>();
+ while (response.isReadable()) {
+ topics.add(BytesDeserializer.readTopic(response));
+ }
+ return topics;
+ } finally {
+ response.release();
+ }
+ });
}
@Override
- public CompletableFuture<TopicDetails> createTopicAsync(
+ public CompletableFuture<TopicDetails> createTopic(
StreamId streamId,
Long partitionsCount,
CompressionAlgorithm compressionAlgorithm,
@@ -108,19 +106,17 @@ public class TopicsTcpClient implements TopicsClient {
payload.writeByte(replicationFactor.orElse((short) 0));
payload.writeBytes(BytesSerializer.toBytes(name));
- return connection
- .sendAsync(CommandCode.Topic.CREATE.getValue(), payload)
- .thenApply(response -> {
- try {
- return BytesDeserializer.readTopicDetails(response);
- } finally {
- response.release();
- }
- });
+ return connection.send(CommandCode.Topic.CREATE.getValue(), payload).thenApply(response -> {
+ try {
+ return BytesDeserializer.readTopicDetails(response);
+ } finally {
+ response.release();
+ }
+ });
}
@Override
- public CompletableFuture<Void> updateTopicAsync(
+ public CompletableFuture<Void> updateTopic(
StreamId streamId,
TopicId topicId,
CompressionAlgorithm compressionAlgorithm,
@@ -138,19 +134,15 @@ public class TopicsTcpClient implements TopicsClient {
payload.writeByte(replicationFactor.orElse((short) 0));
payload.writeBytes(BytesSerializer.toBytes(name));
- return connection
- .sendAsync(CommandCode.Topic.UPDATE.getValue(), payload)
- .thenAccept(response -> response.release());
+ return connection.send(CommandCode.Topic.UPDATE.getValue(), payload).thenAccept(response -> response.release());
}
@Override
- public CompletableFuture<Void> deleteTopicAsync(StreamId streamId, TopicId topicId) {
+ public CompletableFuture<Void> deleteTopic(StreamId streamId, TopicId topicId) {
var payload = Unpooled.buffer();
payload.writeBytes(toBytes(streamId));
payload.writeBytes(toBytes(topicId));
- return connection
- .sendAsync(CommandCode.Topic.DELETE.getValue(), payload)
- .thenAccept(response -> response.release());
+ return connection.send(CommandCode.Topic.DELETE.getValue(), payload).thenAccept(response -> response.release());
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java
index 8f7ded490..3857d5455 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java
@@ -43,7 +43,7 @@ public class UsersTcpClient implements UsersClient {
}
@Override
- public CompletableFuture<IdentityInfo> loginAsync(String username, String password) {
+ public CompletableFuture<IdentityInfo> login(String username, String password) {
String version = "0.6.30";
String context = "java-sdk";
@@ -60,7 +60,7 @@ public class UsersTcpClient implements UsersClient {
log.debug("Logging in user: {}", username);
- return connection.sendAsync(CommandCode.User.LOGIN.getValue(), payload).thenApply(response -> {
+ return connection.send(CommandCode.User.LOGIN.getValue(), payload).thenApply(response -> {
try {
// Read the user ID from response (4-byte unsigned int LE)
var userId = response.readUnsignedIntLE();
@@ -72,12 +72,12 @@ public class UsersTcpClient implements UsersClient {
}
@Override
- public CompletableFuture<Void> logoutAsync() {
+ public CompletableFuture<Void> logout() {
var payload = Unpooled.buffer(0); // Empty payload for logout
log.debug("Logging out");
- return connection.sendAsync(CommandCode.User.LOGOUT.getValue(), payload).thenAccept(response -> {
+ return connection.send(CommandCode.User.LOGOUT.getValue(), payload).thenAccept(response -> {
response.release();
log.debug("Logged out successfully");
});
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/ConsumerGroupsHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/ConsumerGroupsHttpClient.java
index fec8f8ddf..1d73f56f6 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/ConsumerGroupsHttpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/ConsumerGroupsHttpClient.java
@@ -22,6 +22,7 @@ package org.apache.iggy.client.blocking.http;
import org.apache.iggy.client.blocking.ConsumerGroupsClient;
import org.apache.iggy.consumergroup.ConsumerGroup;
import org.apache.iggy.consumergroup.ConsumerGroupDetails;
+import org.apache.iggy.exception.IggyOperationNotSupportedException;
import org.apache.iggy.identifier.ConsumerId;
import org.apache.iggy.identifier.StreamId;
import org.apache.iggy.identifier.TopicId;
@@ -64,12 +65,12 @@ class ConsumerGroupsHttpClient implements ConsumerGroupsClient {
@Override
public void joinConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
- throw new UnsupportedOperationException("Method not available in HTTP client");
+ throw new IggyOperationNotSupportedException("joinConsumerGroup", "HTTP");
}
@Override
public void leaveConsumerGroup(StreamId streamId, TopicId topicId, ConsumerId groupId) {
- throw new UnsupportedOperationException("Method not available in HTTP client");
+ throw new IggyOperationNotSupportedException("leaveConsumerGroup", "HTTP");
}
private static String path(StreamId streamId, TopicId topicId) {
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java
index a24166c1b..dfef3de96 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java
@@ -29,9 +29,15 @@ import org.apache.iggy.client.blocking.StreamsClient;
import org.apache.iggy.client.blocking.SystemClient;
import org.apache.iggy.client.blocking.TopicsClient;
import org.apache.iggy.client.blocking.UsersClient;
+import org.apache.iggy.exception.IggyMissingCredentialsException;
+
+import java.time.Duration;
+import java.util.Optional;
public class IggyHttpClient implements IggyBaseClient {
+ static final int DEFAULT_HTTP_PORT = 3000;
+
private final SystemHttpClient systemClient;
private final StreamsHttpClient streamsClient;
private final UsersHttpClient usersClient;
@@ -41,9 +47,17 @@ public class IggyHttpClient implements IggyBaseClient {
private final ConsumerOffsetsHttpClient consumerOffsetsClient;
private final MessagesHttpClient messagesClient;
private final PersonalAccessTokensHttpClient personalAccessTokensHttpClient;
+ private final Optional<String> username;
+ private final Optional<String> password;
public IggyHttpClient(String url) {
- InternalHttpClient httpClient = new InternalHttpClient(url);
+ this(url, null, null, null, null);
+ }
+
+ IggyHttpClient(String url, String username, String password, Duration connectionTimeout, Duration requestTimeout) {
+ this.username = Optional.ofNullable(username);
+ this.password = Optional.ofNullable(password);
+ InternalHttpClient httpClient = new InternalHttpClient(url, connectionTimeout, requestTimeout);
systemClient = new SystemHttpClient(httpClient);
streamsClient = new StreamsHttpClient(httpClient);
usersClient = new UsersHttpClient(httpClient);
@@ -55,6 +69,27 @@ public class IggyHttpClient implements IggyBaseClient {
personalAccessTokensHttpClient = new PersonalAccessTokensHttpClient(httpClient);
}
+ /**
+ * Creates a new builder for configuring IggyHttpClient.
+ *
+ * @return a new Builder instance
+ */
+ public static IggyHttpClientBuilder builder() {
+ return new IggyHttpClientBuilder();
+ }
+
+ /**
+ * Logs in using the credentials provided during client construction.
+ *
+ * @throws IggyMissingCredentialsException if no credentials were provided
+ */
+ public void login() {
+ if (username.isEmpty() || password.isEmpty()) {
+ throw new IggyMissingCredentialsException();
+ }
+ usersClient.login(username.get(), password.get());
+ }
+
@Override
public SystemClient system() {
return systemClient;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClientBuilder.java
new file mode 100644
index 000000000..0d999612b
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClientBuilder.java
@@ -0,0 +1,210 @@
+/*
+ * 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.client.blocking.http;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+import org.apache.iggy.exception.IggyMissingCredentialsException;
+
+import java.time.Duration;
+
+/**
+ * Builder for creating configured IggyHttpClient instances.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * // Basic usage with explicit login
+ * var client = IggyHttpClient.builder()
+ * .url("http://localhost:3000")
+ * .build();
+ * client.users().login("iggy", "iggy");
+ *
+ * // Using host/port instead of full URL
+ * var client = IggyHttpClient.builder()
+ * .host("localhost")
+ * .port(3000)
+ * .build();
+ * client.users().login("iggy", "iggy");
+ *
+ * // Convenience method with auto-login
+ * var client = IggyHttpClient.builder()
+ * .host("localhost")
+ * .port(3000)
+ * .credentials("iggy", "iggy")
+ * .buildAndLogin();
+ *
+ * // HTTPS with auto-login
+ * var client = IggyHttpClient.builder()
+ * .host("iggy-server.example.com")
+ * .port(443)
+ * .enableTls()
+ * .credentials("admin", "secret")
+ * .buildAndLogin();
+ * }</pre>
+ *
+ * @see IggyHttpClient#builder()
+ */
+public final class IggyHttpClientBuilder {
+ private String url;
+ private String host = "localhost";
+ private Integer port = IggyHttpClient.DEFAULT_HTTP_PORT;
+ private boolean enableTls = false;
+ private Duration connectionTimeout;
+ private Duration requestTimeout;
+ private String username;
+ private String password;
+
+ IggyHttpClientBuilder() {}
+
+ /**
+ * Sets the full URL for the Iggy server.
+ * If set, this takes precedence over host/port/tls settings.
+ *
+ * @param url the full URL (e.g., "http://localhost:3000")
+ * @return this builder
+ */
+ public IggyHttpClientBuilder url(String url) {
+ this.url = url;
+ return this;
+ }
+
+ /**
+ * Sets the host address for the Iggy server.
+ *
+ * @param host the host address
+ * @return this builder
+ */
+ public IggyHttpClientBuilder host(String host) {
+ this.host = host;
+ return this;
+ }
+
+ /**
+ * Sets the port for the Iggy server.
+ *
+ * @param port the port number
+ * @return this builder
+ */
+ public IggyHttpClientBuilder port(Integer port) {
+ this.port = port;
+ return this;
+ }
+
+ /**
+ * Enables TLS for the HTTP connection.
+ *
+ * @return this builder
+ */
+ public IggyHttpClientBuilder enableTls() {
+ this.enableTls = true;
+ return this;
+ }
+
+ /**
+ * Enables or disables TLS for the HTTP connection.
+ *
+ * @param enableTls whether to enable TLS
+ * @return this builder
+ */
+ public IggyHttpClientBuilder tls(boolean enableTls) {
+ this.enableTls = enableTls;
+ return this;
+ }
+
+ /**
+ * Sets the connection timeout.
+ *
+ * @param connectionTimeout the connection timeout duration
+ * @return this builder
+ */
+ public IggyHttpClientBuilder connectionTimeout(Duration connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ return this;
+ }
+
+ /**
+ * Sets the request timeout.
+ *
+ * @param requestTimeout the request timeout duration
+ * @return this builder
+ */
+ public IggyHttpClientBuilder requestTimeout(Duration requestTimeout) {
+ this.requestTimeout = requestTimeout;
+ return this;
+ }
+
+ /**
+ * Sets the credentials for authentication.
+ * These credentials are stored and can be used with {@link IggyHttpClient#login()}.
+ *
+ * @param username the username
+ * @param password the password
+ * @return this builder
+ */
+ public IggyHttpClientBuilder credentials(String username, String password) {
+ this.username = username;
+ this.password = password;
+ return this;
+ }
+
+ /**
+ * Builds and returns a configured IggyHttpClient instance.
+ * The client will not be logged in.
+ *
+ * @return a new IggyHttpClient instance
+ * @throws IggyInvalidArgumentException if the host is null or empty (when url is not set),
+ * or if the port is not positive (when url is not set)
+ */
+ public IggyHttpClient build() {
+ String finalUrl;
+ if (StringUtils.isNotBlank(url)) {
+ finalUrl = url;
+ } else {
+ if (StringUtils.isBlank(host)) {
+ throw new IggyInvalidArgumentException("Host cannot be null or empty");
+ }
+ if (port == null || port <= 0) {
+ throw new IggyInvalidArgumentException("Port must be a positive integer");
+ }
+ String protocol = enableTls ? "https" : "http";
+ finalUrl = protocol + "://" + host + ":" + port;
+ }
+ return new IggyHttpClient(finalUrl, username, password, connectionTimeout, requestTimeout);
+ }
+
+ /**
+ * Builds and logs in using the provided credentials.
+ * This is a convenience method equivalent to calling {@code build()} followed by {@code login()}.
+ *
+ * @return a new IggyHttpClient instance that is logged in
+ * @throws IggyMissingCredentialsException if no credentials were provided
+ * @throws IggyInvalidArgumentException if the host is null or empty (when url is not set),
+ * or if the port is not positive (when url is not set)
+ */
+ public IggyHttpClient buildAndLogin() {
+ if (username == null || password == null) {
+ throw new IggyMissingCredentialsException(
+ "Credentials must be provided to use buildAndLogin(). Use credentials(username, password).");
+ }
+ IggyHttpClient client = build();
+ client.login();
+ return client;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/InternalHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/InternalHttpClient.java
index 70a2379a0..4965716fb 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/InternalHttpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/InternalHttpClient.java
@@ -25,15 +25,17 @@ import org.apache.hc.core5.http.ClassicHttpResponse;
import org.apache.hc.core5.http.NameValuePair;
import org.apache.hc.core5.http.io.HttpClientResponseHandler;
import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
-import org.apache.iggy.client.blocking.http.error.IggyHttpError;
-import org.apache.iggy.client.blocking.http.error.IggyHttpException;
+import org.apache.iggy.exception.IggyConnectionException;
+import org.apache.iggy.exception.IggyServerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tools.jackson.core.type.TypeReference;
import tools.jackson.databind.JavaType;
import tools.jackson.databind.ObjectMapper;
+import tools.jackson.databind.node.ObjectNode;
import java.io.IOException;
+import java.time.Duration;
import java.util.Optional;
final class InternalHttpClient {
@@ -43,10 +45,18 @@ final class InternalHttpClient {
private static final String AUTHORIZATION = "Authorization";
private final String url;
private final ObjectMapper objectMapper = ObjectMapperFactory.getInstance();
+ private final Duration connectionTimeout;
+ private final Duration requestTimeout;
private Optional<String> token = Optional.empty();
InternalHttpClient(String url) {
+ this(url, null, null);
+ }
+
+ InternalHttpClient(String url, Duration connectionTimeout, Duration requestTimeout) {
this.url = url;
+ this.connectionTimeout = connectionTimeout;
+ this.requestTimeout = requestTimeout;
}
void setToken(Optional<String> token) {
@@ -96,7 +106,7 @@ final class InternalHttpClient {
try (var client = HttpClients.createDefault()) {
return client.execute(request, responseHandler);
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new IggyConnectionException("HTTP request failed", e);
}
}
@@ -140,8 +150,12 @@ final class InternalHttpClient {
private void handleErrorResponse(ClassicHttpResponse response) throws IOException {
if (!isSuccessful(response.getCode())) {
- var error = objectMapper.readValue(response.getEntity().getContent(), IggyHttpError.class);
- throw new IggyHttpException(error);
+ var errorNode = objectMapper.readValue(response.getEntity().getContent(), ObjectNode.class);
+ String id = errorNode.has("id") ? errorNode.get("id").asText() : null;
+ String code = errorNode.has("code") ? errorNode.get("code").asText() : null;
+ String reason = errorNode.has("reason") ? errorNode.get("reason").asText() : null;
+ String field = errorNode.has("field") ? errorNode.get("field").asText() : null;
+ throw IggyServerException.fromHttpResponse(id, code, reason, field);
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java
index af6edb30c..3507919dc 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/SystemHttpClient.java
@@ -20,6 +20,7 @@
package org.apache.iggy.client.blocking.http;
import org.apache.iggy.client.blocking.SystemClient;
+import org.apache.iggy.exception.IggyOperationNotSupportedException;
import org.apache.iggy.system.ClientInfo;
import org.apache.iggy.system.ClientInfoDetails;
import org.apache.iggy.system.Stats;
@@ -46,7 +47,7 @@ class SystemHttpClient implements SystemClient {
@Override
public ClientInfoDetails getMe() {
- throw new UnsupportedOperationException("Method not available in HTTP client");
+ throw new IggyOperationNotSupportedException("getMe", "HTTP");
}
@Override
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java
index 6d47d039b..1f150d328 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java
@@ -19,7 +19,6 @@
package org.apache.iggy.client.blocking.tcp;
-import org.apache.commons.lang3.StringUtils;
import org.apache.iggy.client.blocking.ConsumerGroupsClient;
import org.apache.iggy.client.blocking.ConsumerOffsetsClient;
import org.apache.iggy.client.blocking.IggyBaseClient;
@@ -30,6 +29,10 @@ import org.apache.iggy.client.blocking.StreamsClient;
import org.apache.iggy.client.blocking.SystemClient;
import org.apache.iggy.client.blocking.TopicsClient;
import org.apache.iggy.client.blocking.UsersClient;
+import org.apache.iggy.config.RetryPolicy;
+import org.apache.iggy.exception.IggyMissingCredentialsException;
+import org.apache.iggy.exception.IggyNotConnectedException;
+import org.apache.iggy.user.IdentityInfo;
import java.io.File;
import java.time.Duration;
@@ -37,32 +40,29 @@ import java.util.Optional;
public class IggyTcpClient implements IggyBaseClient {
- private final UsersTcpClient usersClient;
- private final StreamsTcpClient streamsClient;
- private final TopicsTcpClient topicsClient;
- private final PartitionsTcpClient partitionsClient;
- private final ConsumerGroupsTcpClient consumerGroupsClient;
- private final ConsumerOffsetTcpClient consumerOffsetsClient;
- private final MessagesTcpClient messagesClient;
- private final SystemTcpClient systemClient;
- private final PersonalAccessTokensTcpClient personalAccessTokensClient;
private final String host;
- private final Integer port;
- private final Optional<String> username;
- private final Optional<String> password;
- private final Optional<Duration> connectionTimeout;
- private final Optional<Duration> requestTimeout;
- private final Optional<Integer> connectionPoolSize;
- private final Optional<RetryPolicy> retryPolicy;
+ private final int port;
private final boolean enableTls;
private final Optional<File> tlsCertificate;
+ private final Optional<String> username;
+ private final Optional<String> password;
+
+ private UsersTcpClient usersClient;
+ private StreamsTcpClient streamsClient;
+ private TopicsTcpClient topicsClient;
+ private PartitionsTcpClient partitionsClient;
+ private ConsumerGroupsTcpClient consumerGroupsClient;
+ private ConsumerOffsetTcpClient consumerOffsetsClient;
+ private MessagesTcpClient messagesClient;
+ private SystemTcpClient systemClient;
+ private PersonalAccessTokensTcpClient personalAccessTokensClient;
public IggyTcpClient(String host, Integer port) {
this(host, port, null, null, null, null, null, null, false, Optional.empty());
}
@SuppressWarnings("checkstyle:ParameterNumber")
- private IggyTcpClient(
+ IggyTcpClient(
String host,
Integer port,
String username,
@@ -77,13 +77,14 @@ public class IggyTcpClient implements IggyBaseClient {
this.port = port;
this.username = Optional.ofNullable(username);
this.password = Optional.ofNullable(password);
- this.connectionTimeout = Optional.ofNullable(connectionTimeout);
- this.requestTimeout = Optional.ofNullable(requestTimeout);
- this.connectionPoolSize = Optional.ofNullable(connectionPoolSize);
- this.retryPolicy = Optional.ofNullable(retryPolicy);
this.enableTls = enableTls;
this.tlsCertificate = tlsCertificate;
+ }
+ /**
+ * Connects to the Iggy server.
+ */
+ public void connect() {
InternalTcpClient tcpClient = new InternalTcpClient(host, port, enableTls, tlsCertificate);
tcpClient.connect();
usersClient = new UsersTcpClient(tcpClient);
@@ -95,11 +96,6 @@ public class IggyTcpClient implements IggyBaseClient {
messagesClient = new MessagesTcpClient(tcpClient);
systemClient = new SystemTcpClient(tcpClient);
personalAccessTokensClient = new PersonalAccessTokensTcpClient(tcpClient);
-
- // Auto-login if credentials are provided
- if (this.username.isPresent() && this.password.isPresent()) {
- usersClient.login(this.username.get(), this.password.get());
- }
}
/**
@@ -107,293 +103,95 @@ public class IggyTcpClient implements IggyBaseClient {
*
* @return a new Builder instance
*/
- public static Builder builder() {
- return new Builder();
+ public static IggyTcpClientBuilder builder() {
+ return new IggyTcpClientBuilder();
+ }
+
+ /**
+ * Logs in using the credentials provided during client construction.
+ *
+ * @throws IggyMissingCredentialsException if no credentials were provided
+ * @throws IggyNotConnectedException if client is not connected
+ */
+ public IdentityInfo login() {
+ if (usersClient == null) {
+ throw new IggyNotConnectedException();
+ }
+ if (username.isEmpty() || password.isEmpty()) {
+ throw new IggyMissingCredentialsException();
+ }
+ return usersClient.login(username.get(), password.get());
}
@Override
public SystemClient system() {
+ if (systemClient == null) {
+ throw new IggyNotConnectedException();
+ }
return systemClient;
}
@Override
public StreamsClient streams() {
+ if (streamsClient == null) {
+ throw new IggyNotConnectedException();
+ }
return streamsClient;
}
@Override
public UsersClient users() {
+ if (usersClient == null) {
+ throw new IggyNotConnectedException();
+ }
return usersClient;
}
@Override
public TopicsClient topics() {
+ if (topicsClient == null) {
+ throw new IggyNotConnectedException();
+ }
return topicsClient;
}
@Override
public PartitionsClient partitions() {
+ if (partitionsClient == null) {
+ throw new IggyNotConnectedException();
+ }
return partitionsClient;
}
@Override
public ConsumerGroupsClient consumerGroups() {
+ if (consumerGroupsClient == null) {
+ throw new IggyNotConnectedException();
+ }
return consumerGroupsClient;
}
@Override
public ConsumerOffsetsClient consumerOffsets() {
+ if (consumerOffsetsClient == null) {
+ throw new IggyNotConnectedException();
+ }
return consumerOffsetsClient;
}
@Override
public MessagesClient messages() {
+ if (messagesClient == null) {
+ throw new IggyNotConnectedException();
+ }
return messagesClient;
}
@Override
public PersonalAccessTokensClient personalAccessTokens() {
- return personalAccessTokensClient;
- }
-
- /**
- * Builder for creating configured IggyTcpClient instances.
- */
- public static final class Builder {
- private String host = "localhost";
- private Integer port = 8090;
- private String username;
- private String password;
- private Duration connectionTimeout;
- private Duration requestTimeout;
- private Integer connectionPoolSize;
- private RetryPolicy retryPolicy;
- private boolean enableTls = false;
- private File tlsCertificate;
-
- private Builder() {}
-
- /**
- * Sets the host address for the Iggy server.
- *
- * @param host the host address
- * @return this builder
- */
- public Builder host(String host) {
- this.host = host;
- return this;
- }
-
- /**
- * Sets the port for the Iggy server.
- *
- * @param port the port number
- * @return this builder
- */
- public Builder port(Integer port) {
- this.port = port;
- return this;
- }
-
- /**
- * Sets the credentials for authentication.
- *
- * @param username the username
- * @param password the password
- * @return this builder
- */
- public Builder credentials(String username, String password) {
- this.username = username;
- this.password = password;
- return this;
- }
-
- /**
- * Sets the connection timeout.
- *
- * @param connectionTimeout the connection timeout duration
- * @return this builder
- */
- public Builder connectionTimeout(Duration connectionTimeout) {
- this.connectionTimeout = connectionTimeout;
- return this;
- }
-
- /**
- * Sets the request timeout.
- *
- * @param requestTimeout the request timeout duration
- * @return this builder
- */
- public Builder requestTimeout(Duration requestTimeout) {
- this.requestTimeout = requestTimeout;
- return this;
- }
-
- /**
- * Sets the connection pool size.
- *
- * @param connectionPoolSize the size of the connection pool
- * @return this builder
- */
- public Builder connectionPoolSize(Integer connectionPoolSize) {
- this.connectionPoolSize = connectionPoolSize;
- return this;
- }
-
- /**
- * Sets the retry policy.
- *
- * @param retryPolicy the retry policy to use
- * @return this builder
- */
- public Builder retryPolicy(RetryPolicy retryPolicy) {
- this.retryPolicy = retryPolicy;
- return this;
- }
-
- /**
- * Enables or disables TLS for the TCP connection.
- *
- * @param enableTls whether to enable TLS
- * @return this builder
- */
- public Builder tls(boolean enableTls) {
- this.enableTls = enableTls;
- return this;
- }
-
- /**
- * Enables TLS for the TCP connection.
- *
- * @return this builder
- */
- public Builder enableTls() {
- this.enableTls = true;
- return this;
- }
-
- /**
- * Sets a custom trusted certificate (PEM file) to validate the server certificate.
- *
- * @param tlsCertificate the PEM file containing the certificate or CA chain
- * @return this builder
- */
- public Builder tlsCertificate(File tlsCertificate) {
- this.tlsCertificate = tlsCertificate;
- return this;
- }
-
- /**
- * Sets a custom trusted certificate (PEM file path) to validate the server certificate.
- *
- * @param tlsCertificatePath the PEM file path containing the certificate or CA chain
- * @return this builder
- */
- public Builder tlsCertificate(String tlsCertificatePath) {
- this.tlsCertificate = StringUtils.isBlank(tlsCertificatePath) ? null : new File(tlsCertificatePath);
- return this;
- }
-
- /**
- * Builds and returns a configured IggyTcpClient instance.
- *
- * @return a new IggyTcpClient instance
- */
- public IggyTcpClient build() {
- if (host == null || host.isEmpty()) {
- throw new IllegalArgumentException("Host cannot be null or empty");
- }
- if (port == null || port <= 0) {
- throw new IllegalArgumentException("Port must be a positive integer");
- }
- return new IggyTcpClient(
- host,
- port,
- username,
- password,
- connectionTimeout,
- requestTimeout,
- connectionPoolSize,
- retryPolicy,
- this.enableTls,
- Optional.ofNullable(tlsCertificate));
- }
- }
-
- /**
- * Retry policy for client operations.
- */
- public static final class RetryPolicy {
- private final int maxRetries;
- private final Duration initialDelay;
- private final Duration maxDelay;
- private final double multiplier;
-
- private RetryPolicy(int maxRetries, Duration initialDelay, Duration maxDelay, double multiplier) {
- this.maxRetries = maxRetries;
- this.initialDelay = initialDelay;
- this.maxDelay = maxDelay;
- this.multiplier = multiplier;
- }
-
- /**
- * Creates a retry policy with exponential backoff.
- *
- * @return a RetryPolicy with exponential backoff configuration
- */
- public static RetryPolicy exponentialBackoff() {
- return new RetryPolicy(3, Duration.ofMillis(100), Duration.ofSeconds(5), 2.0);
- }
-
- /**
- * Creates a retry policy with exponential backoff and custom parameters.
- *
- * @param maxRetries the maximum number of retries
- * @param initialDelay the initial delay before the first retry
- * @param maxDelay the maximum delay between retries
- * @param multiplier the multiplier for exponential backoff
- * @return a RetryPolicy with custom exponential backoff configuration
- */
- public static RetryPolicy exponentialBackoff(
- int maxRetries, Duration initialDelay, Duration maxDelay, double multiplier) {
- return new RetryPolicy(maxRetries, initialDelay, maxDelay, multiplier);
- }
-
- /**
- * Creates a retry policy with fixed delay.
- *
- * @param maxRetries the maximum number of retries
- * @param delay the fixed delay between retries
- * @return a RetryPolicy with fixed delay configuration
- */
- public static RetryPolicy fixedDelay(int maxRetries, Duration delay) {
- return new RetryPolicy(maxRetries, delay, delay, 1.0);
- }
-
- /**
- * Creates a no-retry policy.
- *
- * @return a RetryPolicy that does not retry
- */
- public static RetryPolicy noRetry() {
- return new RetryPolicy(0, Duration.ZERO, Duration.ZERO, 1.0);
- }
-
- public int getMaxRetries() {
- return maxRetries;
- }
-
- public Duration getInitialDelay() {
- return initialDelay;
- }
-
- public Duration getMaxDelay() {
- return maxDelay;
- }
-
- public double getMultiplier() {
- return multiplier;
+ if (personalAccessTokensClient == null) {
+ throw new IggyNotConnectedException();
}
+ return personalAccessTokensClient;
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java
new file mode 100644
index 000000000..c5b9c39ac
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilder.java
@@ -0,0 +1,245 @@
+/*
+ * 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.client.blocking.tcp;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.iggy.config.RetryPolicy;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+import org.apache.iggy.exception.IggyMissingCredentialsException;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.Optional;
+
+/**
+ * Builder for creating configured IggyTcpClient instances.
+ *
+ * <p>Example usage:
+ * <pre>{@code
+ * // Basic usage with explicit connect and login
+ * var client = IggyTcpClient.builder()
+ * .host("localhost")
+ * .port(8090)
+ * .build();
+ * client.connect();
+ * client.users().login("iggy", "iggy");
+ *
+ * // Convenience method with auto-login
+ * var client = IggyTcpClient.builder()
+ * .host("localhost")
+ * .port(8090)
+ * .credentials("iggy", "iggy")
+ * .buildAndLogin();
+ *
+ * // With TLS enabled
+ * var client = IggyTcpClient.builder()
+ * .host("iggy-server.example.com")
+ * .port(8090)
+ * .enableTls()
+ * .credentials("admin", "secret")
+ * .buildAndLogin();
+ * }</pre>
+ *
+ * @see IggyTcpClient#builder()
+ */
+public final class IggyTcpClientBuilder {
+ private String host = "localhost";
+ private Integer port = 8090;
+ private String username;
+ private String password;
+ private Duration connectionTimeout;
+ private Duration requestTimeout;
+ private Integer connectionPoolSize;
+ private RetryPolicy retryPolicy;
+ private boolean enableTls = false;
+ private File tlsCertificate;
+
+ IggyTcpClientBuilder() {}
+
+ /**
+ * Sets the host address for the Iggy server.
+ *
+ * @param host the host address
+ * @return this builder
+ */
+ public IggyTcpClientBuilder host(String host) {
+ this.host = host;
+ return this;
+ }
+
+ /**
+ * Sets the port for the Iggy server.
+ *
+ * @param port the port number
+ * @return this builder
+ */
+ public IggyTcpClientBuilder port(Integer port) {
+ this.port = port;
+ return this;
+ }
+
+ /**
+ * Sets the credentials for authentication.
+ * These credentials are stored and can be used with {@link IggyTcpClient#login()}.
+ *
+ * @param username the username
+ * @param password the password
+ * @return this builder
+ */
+ public IggyTcpClientBuilder credentials(String username, String password) {
+ this.username = username;
+ this.password = password;
+ return this;
+ }
+
+ /**
+ * Sets the connection timeout.
+ *
+ * @param connectionTimeout the connection timeout duration
+ * @return this builder
+ */
+ public IggyTcpClientBuilder connectionTimeout(Duration connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ return this;
+ }
+
+ /**
+ * Sets the request timeout.
+ *
+ * @param requestTimeout the request timeout duration
+ * @return this builder
+ */
+ public IggyTcpClientBuilder requestTimeout(Duration requestTimeout) {
+ this.requestTimeout = requestTimeout;
+ return this;
+ }
+
+ /**
+ * Sets the connection pool size.
+ *
+ * @param connectionPoolSize the size of the connection pool
+ * @return this builder
+ */
+ public IggyTcpClientBuilder connectionPoolSize(Integer connectionPoolSize) {
+ this.connectionPoolSize = connectionPoolSize;
+ return this;
+ }
+
+ /**
+ * Sets the retry policy.
+ *
+ * @param retryPolicy the retry policy to use
+ * @return this builder
+ */
+ public IggyTcpClientBuilder retryPolicy(RetryPolicy retryPolicy) {
+ this.retryPolicy = retryPolicy;
+ return this;
+ }
+
+ /**
+ * Enables or disables TLS for the TCP connection.
+ *
+ * @param enableTls whether to enable TLS
+ * @return this builder
+ */
+ public IggyTcpClientBuilder tls(boolean enableTls) {
+ this.enableTls = enableTls;
+ return this;
+ }
+
+ /**
+ * Enables TLS for the TCP connection.
+ *
+ * @return this builder
+ */
+ public IggyTcpClientBuilder enableTls() {
+ this.enableTls = true;
+ return this;
+ }
+
+ /**
+ * Sets a custom trusted certificate (PEM file) to validate the server certificate.
+ *
+ * @param tlsCertificate the PEM file containing the certificate or CA chain
+ * @return this builder
+ */
+ public IggyTcpClientBuilder tlsCertificate(File tlsCertificate) {
+ this.tlsCertificate = tlsCertificate;
+ return this;
+ }
+
+ /**
+ * Sets a custom trusted certificate (PEM file path) to validate the server certificate.
+ *
+ * @param tlsCertificatePath the PEM file path containing the certificate or CA chain
+ * @return this builder
+ */
+ public IggyTcpClientBuilder tlsCertificate(String tlsCertificatePath) {
+ this.tlsCertificate = StringUtils.isBlank(tlsCertificatePath) ? null : new File(tlsCertificatePath);
+ return this;
+ }
+
+ /**
+ * Builds and returns a configured IggyTcpClient instance.
+ * Note: You still need to call {@link IggyTcpClient#connect()} on the returned client.
+ *
+ * @return a new IggyTcpClient instance
+ * @throws IggyInvalidArgumentException if the host is null or empty, or if the port is not positive
+ */
+ public IggyTcpClient build() {
+ if (host == null || host.isEmpty()) {
+ throw new IggyInvalidArgumentException("Host cannot be null or empty");
+ }
+ if (port == null || port <= 0) {
+ throw new IggyInvalidArgumentException("Port must be a positive integer");
+ }
+ return new IggyTcpClient(
+ host,
+ port,
+ username,
+ password,
+ connectionTimeout,
+ requestTimeout,
+ connectionPoolSize,
+ retryPolicy,
+ this.enableTls,
+ Optional.ofNullable(tlsCertificate));
+ }
+
+ /**
+ * Builds, connects, and logs in using the provided credentials.
+ * This is a convenience method equivalent to calling {@code build()}, {@code connect()},
+ * and {@code login()}.
+ *
+ * @return a new IggyTcpClient instance that is connected and logged in
+ * @throws IggyMissingCredentialsException if no credentials were provided
+ * @throws IggyInvalidArgumentException if the host is null or empty, or if the port is not positive
+ */
+ public IggyTcpClient buildAndLogin() {
+ if (username == null || password == null) {
+ throw new IggyMissingCredentialsException(
+ "Credentials must be provided to use buildAndLogin(). Use credentials(username, password).");
+ }
+ IggyTcpClient client = build();
+ client.connect();
+ client.login();
+ return client;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/InternalTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/InternalTcpClient.java
index 234a29455..944309937 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/InternalTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/InternalTcpClient.java
@@ -25,6 +25,10 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
+import org.apache.iggy.exception.IggyConnectionException;
+import org.apache.iggy.exception.IggyEmptyResponseException;
+import org.apache.iggy.exception.IggyServerException;
+import org.apache.iggy.exception.IggyTlsException;
import org.apache.iggy.serde.CommandCode;
import reactor.core.publisher.Mono;
import reactor.netty.Connection;
@@ -66,7 +70,7 @@ final class InternalTcpClient {
SslContext sslContext = builder.build();
tcpClient = tcpClient.secure(sslSpec -> sslSpec.sslContext(sslContext));
} catch (SSLException e) {
- throw new RuntimeException("Failed to configure TLS for TcpClient", e);
+ throw new IggyTlsException("Failed to configure TLS for TcpClient", e);
}
}
@@ -106,7 +110,7 @@ final class InternalTcpClient {
<T> T exchangeForEntity(CommandCode command, ByteBuf payload, Function<ByteBuf, T> responseMapper) {
return exchange(command, payload, response -> {
if (!response.payload().isReadable()) {
- throw new RuntimeException("Received an empty response for command " + command);
+ throw new IggyEmptyResponseException(command.toString());
}
return responseMapper.apply(response.payload());
});
@@ -131,7 +135,10 @@ final class InternalTcpClient {
try {
IggyResponse response = responses.take();
if (response.status() != 0) {
- throw new RuntimeException("Received an invalid response with status " + response.status());
+ byte[] errorPayload = new byte[response.payload().readableBytes()];
+ response.payload().readBytes(errorPayload);
+ response.payload().release();
+ throw IggyServerException.fromTcpResponse(response.status(), errorPayload);
}
try {
return responseMapper.apply(response);
@@ -139,7 +146,7 @@ final class InternalTcpClient {
response.payload().release();
}
} catch (InterruptedException e) {
- throw new RuntimeException(e);
+ throw new IggyConnectionException("Interrupted while waiting for response", e);
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/config/RetryPolicy.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/config/RetryPolicy.java
new file mode 100644
index 000000000..1bf85a366
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/config/RetryPolicy.java
@@ -0,0 +1,130 @@
+/*
+ * 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.config;
+
+import java.time.Duration;
+
+/**
+ * Retry policy for client operations.
+ *
+ * <p>This class provides configuration for retry behavior including exponential backoff,
+ * fixed delay, and no-retry strategies.
+ */
+public final class RetryPolicy {
+
+ private final int maxRetries;
+ private final Duration initialDelay;
+ private final Duration maxDelay;
+ private final double multiplier;
+
+ private RetryPolicy(int maxRetries, Duration initialDelay, Duration maxDelay, double multiplier) {
+ this.maxRetries = maxRetries;
+ this.initialDelay = initialDelay;
+ this.maxDelay = maxDelay;
+ this.multiplier = multiplier;
+ }
+
+ /**
+ * Creates a retry policy with exponential backoff using default parameters.
+ *
+ * <p>Default configuration:
+ * <ul>
+ * <li>Max retries: 3</li>
+ * <li>Initial delay: 100ms</li>
+ * <li>Max delay: 5s</li>
+ * <li>Multiplier: 2.0</li>
+ * </ul>
+ *
+ * @return a RetryPolicy with exponential backoff configuration
+ */
+ public static RetryPolicy exponentialBackoff() {
+ return new RetryPolicy(3, Duration.ofMillis(100), Duration.ofSeconds(5), 2.0);
+ }
+
+ /**
+ * Creates a retry policy with exponential backoff and custom parameters.
+ *
+ * @param maxRetries the maximum number of retries
+ * @param initialDelay the initial delay before the first retry
+ * @param maxDelay the maximum delay between retries
+ * @param multiplier the multiplier for exponential backoff
+ * @return a RetryPolicy with custom exponential backoff configuration
+ */
+ public static RetryPolicy exponentialBackoff(
+ int maxRetries, Duration initialDelay, Duration maxDelay, double multiplier) {
+ return new RetryPolicy(maxRetries, initialDelay, maxDelay, multiplier);
+ }
+
+ /**
+ * Creates a retry policy with fixed delay between retries.
+ *
+ * @param maxRetries the maximum number of retries
+ * @param delay the fixed delay between retries
+ * @return a RetryPolicy with fixed delay configuration
+ */
+ public static RetryPolicy fixedDelay(int maxRetries, Duration delay) {
+ return new RetryPolicy(maxRetries, delay, delay, 1.0);
+ }
+
+ /**
+ * Creates a no-retry policy.
+ *
+ * @return a RetryPolicy that does not retry
+ */
+ public static RetryPolicy noRetry() {
+ return new RetryPolicy(0, Duration.ZERO, Duration.ZERO, 1.0);
+ }
+
+ /**
+ * Gets the maximum number of retry attempts.
+ *
+ * @return the maximum number of retries
+ */
+ public int getMaxRetries() {
+ return maxRetries;
+ }
+
+ /**
+ * Gets the initial delay before the first retry.
+ *
+ * @return the initial delay duration
+ */
+ public Duration getInitialDelay() {
+ return initialDelay;
+ }
+
+ /**
+ * Gets the maximum delay between retries.
+ *
+ * @return the maximum delay duration
+ */
+ public Duration getMaxDelay() {
+ return maxDelay;
+ }
+
+ /**
+ * Gets the multiplier for exponential backoff.
+ *
+ * @return the backoff multiplier
+ */
+ public double getMultiplier() {
+ return multiplier;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/config/package-info.java
similarity index 72%
rename from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyClient.java
rename to foreign/java/java-sdk/src/main/java/org/apache/iggy/config/package-info.java
index af62082f8..7fdf84abe 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/config/package-info.java
@@ -17,17 +17,15 @@
* under the License.
*/
-package org.apache.iggy.client.blocking;
-
-public class IggyClient {
-
- private final IggyBaseClient baseClient;
-
- public IggyClient(IggyBaseClient baseClient) {
- this.baseClient = baseClient;
- }
+/**
+ * Configuration classes for the Iggy Java SDK.
+ *
+ * <p>This package provides configuration options for client behavior:
+ * <ul>
+ * <li>{@link org.apache.iggy.config.RetryPolicy} - Retry behavior configuration</li>
+ * </ul>
+ */
+@NonNullApi
+package org.apache.iggy.config;
- public IggyBaseClient getBaseClient() {
- return baseClient;
- }
-}
+import org.apache.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyAuthenticationException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyAuthenticationException.java
new file mode 100644
index 000000000..0c4c8b51b
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyAuthenticationException.java
@@ -0,0 +1,49 @@
+/*
+ * 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.exception;
+
+import java.util.Optional;
+
+/**
+ * Exception thrown when authentication fails.
+ *
+ * <p>This corresponds to 401-type errors, such as invalid credentials, unauthenticated
+ * requests, invalid username/password, etc.
+ */
+public class IggyAuthenticationException extends IggyServerException {
+
+ /**
+ * Constructs a new IggyAuthenticationException.
+ *
+ * @param errorCode the error code enum
+ * @param rawErrorCode the raw numeric error code from the server
+ * @param reason the error reason/message
+ * @param field the optional field related to the error
+ * @param errorId the optional error ID for correlation with server logs
+ */
+ public IggyAuthenticationException(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ super(errorCode, rawErrorCode, reason, field, errorId);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyAuthorizationException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyAuthorizationException.java
new file mode 100644
index 000000000..b16cd7326
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyAuthorizationException.java
@@ -0,0 +1,49 @@
+/*
+ * 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.exception;
+
+import java.util.Optional;
+
+/**
+ * Exception thrown when the user is not authorized to perform an operation.
+ *
+ * <p>This corresponds to 403-type errors, where the user is authenticated but
+ * lacks the necessary permissions.
+ */
+public class IggyAuthorizationException extends IggyServerException {
+
+ /**
+ * Constructs a new IggyAuthorizationException.
+ *
+ * @param errorCode the error code enum
+ * @param rawErrorCode the raw numeric error code from the server
+ * @param reason the error reason/message
+ * @param field the optional field related to the error
+ * @param errorId the optional error ID for correlation with server logs
+ */
+ public IggyAuthorizationException(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ super(errorCode, rawErrorCode, reason, field, errorId);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyClientException.java
similarity index 54%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyClientException.java
index 3dfc40792..4f80685dd 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyClientException.java
@@ -17,30 +17,32 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown for client-side errors.
+ *
+ * <p>This is the base class for client-side exceptions that are not related
+ * to server responses or connection issues.
*/
-public interface UsersClient {
+public class IggyClientException extends IggyException {
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyClientException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyClientException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyClientException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyClientException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConflictException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConflictException.java
new file mode 100644
index 000000000..82cd53c1c
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConflictException.java
@@ -0,0 +1,49 @@
+/*
+ * 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.exception;
+
+import java.util.Optional;
+
+/**
+ * Exception thrown when a resource conflict occurs.
+ *
+ * <p>This corresponds to 409-type errors, such as trying to create a resource
+ * that already exists (stream, topic, user, consumer group, etc.).
+ */
+public class IggyConflictException extends IggyServerException {
+
+ /**
+ * Constructs a new IggyConflictException.
+ *
+ * @param errorCode the error code enum
+ * @param rawErrorCode the raw numeric error code from the server
+ * @param reason the error reason/message
+ * @param field the optional field related to the error
+ * @param errorId the optional error ID for correlation with server logs
+ */
+ public IggyConflictException(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ super(errorCode, rawErrorCode, reason, field, errorId);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConnectionClosedException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConnectionClosedException.java
new file mode 100644
index 000000000..4e028646a
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConnectionClosedException.java
@@ -0,0 +1,52 @@
+/*
+ * 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.exception;
+
+/**
+ * Exception thrown when the connection has been closed unexpectedly.
+ */
+public class IggyConnectionClosedException extends IggyConnectionException {
+
+ /**
+ * Constructs a new IggyConnectionClosedException with a default message.
+ */
+ public IggyConnectionClosedException() {
+ super("Connection has been closed");
+ }
+
+ /**
+ * Constructs a new IggyConnectionClosedException with the specified message.
+ *
+ * @param message the detail message
+ */
+ public IggyConnectionClosedException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new IggyConnectionClosedException with the specified message and cause.
+ *
+ * @param message the detail message
+ * @param cause the cause of the exception
+ */
+ public IggyConnectionClosedException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConnectionException.java
similarity index 52%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConnectionException.java
index 3dfc40792..90688cc79 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyConnectionException.java
@@ -17,30 +17,32 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when a connection-related error occurs.
+ *
+ * <p>This is the base class for all connection-related exceptions, including
+ * TLS errors, connection closed errors, and general connection failures.
*/
-public interface UsersClient {
+public class IggyConnectionException extends IggyException {
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyConnectionException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyConnectionException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyConnectionException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyConnectionException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyEmptyResponseException.java
similarity index 55%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyEmptyResponseException.java
index 3dfc40792..ce531f5e3 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyEmptyResponseException.java
@@ -17,30 +17,31 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when an empty response is received for a command that expects data.
*/
-public interface UsersClient {
+public class IggyEmptyResponseException extends IggyProtocolException {
+
+ private final String command;
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyEmptyResponseException for the specified command.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param command the command that received an empty response
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyEmptyResponseException(String command) {
+ super("Received an empty response for command: " + command);
+ this.command = command;
+ }
/**
- * Logs out from the server.
+ * Returns the command that received an empty response.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @return the command name
*/
- CompletableFuture<Void> logoutAsync();
+ public String getCommand() {
+ return command;
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.java
new file mode 100644
index 000000000..fe001bae1
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyErrorCode.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.exception;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Error codes returned by the Iggy server.
+ *
+ * <p>These codes correspond to the error codes defined in the Iggy server's iggy_error.rs.
+ */
+public enum IggyErrorCode {
+ // General errors
+ ERROR(1),
+ INVALID_COMMAND(3),
+ INVALID_FORMAT(4),
+ FEATURE_UNAVAILABLE(6),
+ CANNOT_PARSE_INT(7),
+ CANNOT_PARSE_SLICE(8),
+ CANNOT_PARSE_UTF8(9),
+
+ // Resource errors
+ RESOURCE_NOT_FOUND(20),
+ CANNOT_LOAD_RESOURCE(100),
+
+ // Authentication/Authorization errors
+ UNAUTHENTICATED(40),
+ UNAUTHORIZED(41),
+ INVALID_CREDENTIALS(42),
+ INVALID_USERNAME(43),
+ INVALID_PASSWORD(44),
+ CLEAR_TEXT_PASSWORD_REQUIRED(45),
+ USER_ALREADY_EXISTS(46),
+ USER_INACTIVE(47),
+ CANNOT_DELETE_USER_WITH_ACTIVE_PAT(48),
+ CANNOT_UPDATE_OWN_PERMISSIONS(49),
+ CANNOT_DELETE_YOURSELF(50),
+ CLIENT_ALREADY_EXISTS(51),
+ CLIENT_NOT_FOUND(52),
+ INVALID_PAT_TOKEN(53),
+ PAT_NAME_ALREADY_EXISTS(54),
+ PASSWORD_DOES_NOT_MATCH(77),
+ PASSWORD_HASH_INTERNAL_ERROR(78),
+
+ // Stream errors
+ STREAM_ID_NOT_FOUND(1009),
+ STREAM_NAME_NOT_FOUND(1010),
+ STREAM_ALREADY_EXISTS(1012),
+ INVALID_STREAM_NAME(1013),
+ CANNOT_CREATE_STREAM_DIRECTORY(1014),
+
+ // Topic errors
+ TOPIC_ID_NOT_FOUND(2010),
+ TOPIC_NAME_NOT_FOUND(2011),
+ TOPICS_COUNT_EXCEEDED(2012),
+ TOPIC_ALREADY_EXISTS(2013),
+ INVALID_TOPIC_NAME(2014),
+ INVALID_REPLICATION_FACTOR(2015),
+ CANNOT_CREATE_TOPIC_DIRECTORY(2016),
+
+ // Partition errors
+ PARTITION_NOT_FOUND(3007),
+
+ // Consumer group errors
+ CONSUMER_GROUP_ID_NOT_FOUND(5000),
+ CONSUMER_GROUP_MEMBER_NOT_FOUND(5002),
+ CONSUMER_GROUP_NAME_NOT_FOUND(5003),
+ CONSUMER_GROUP_ALREADY_EXISTS(5004),
+ INVALID_CONSUMER_GROUP_NAME(5005),
+ CONSUMER_GROUP_NOT_JOINED(5006),
+
+ // Segment errors
+ SEGMENT_NOT_FOUND(4000),
+ SEGMENT_CLOSED(4001),
+ CANNOT_READ_SEGMENT(4002),
+ CANNOT_SAVE_SEGMENT(4003),
+
+ // Message errors
+ TOO_MANY_MESSAGES(7000),
+ EMPTY_MESSAGES(7001),
+ TOO_BIG_MESSAGE(7002),
+ INVALID_MESSAGE_CHECKSUM(7003),
+ MESSAGE_NOT_FOUND(7004),
+
+ // Unknown error code
+ UNKNOWN(-1);
+
+ private static final Map<Integer, IggyErrorCode> CODE_MAP = new HashMap<>();
+
+ static {
+ for (IggyErrorCode errorCode : values()) {
+ CODE_MAP.put(errorCode.code, errorCode);
+ }
+ }
+
+ private final int code;
+
+ IggyErrorCode(int code) {
+ this.code = code;
+ }
+
+ /**
+ * Returns the numeric error code.
+ *
+ * @return the error code
+ */
+ public int getCode() {
+ return code;
+ }
+
+ /**
+ * Returns the IggyErrorCode for the given numeric code.
+ *
+ * @param code the numeric error code
+ * @return the corresponding IggyErrorCode, or UNKNOWN if not found
+ */
+ public static IggyErrorCode fromCode(int code) {
+ return CODE_MAP.getOrDefault(code, UNKNOWN);
+ }
+
+ /**
+ * Returns the IggyErrorCode for the given string code.
+ *
+ * @param code the string error code (can be numeric or enum name)
+ * @return the corresponding IggyErrorCode, or UNKNOWN if not found
+ */
+ public static IggyErrorCode fromString(String code) {
+ if (StringUtils.isBlank(code)) {
+ return UNKNOWN;
+ }
+ try {
+ int numericCode = Integer.parseInt(code);
+ return fromCode(numericCode);
+ } catch (NumberFormatException e) {
+ // Try to match by name
+ try {
+ return valueOf(code.toUpperCase().replace(".", "_").replace(" ", "_"));
+ } catch (IllegalArgumentException ex) {
+ return UNKNOWN;
+ }
+ }
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyException.java
new file mode 100644
index 000000000..e997cf57c
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyException.java
@@ -0,0 +1,58 @@
+/*
+ * 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.exception;
+
+/**
+ * Base exception class for all Iggy SDK exceptions.
+ *
+ * <p>This is the root of the Iggy exception hierarchy. All exceptions thrown by the SDK
+ * extend this class, allowing callers to catch all Iggy-related exceptions with a single
+ * catch block if desired.
+ */
+public abstract class IggyException extends RuntimeException {
+
+ /**
+ * Constructs a new IggyException with the specified message.
+ *
+ * @param message the detail message
+ */
+ protected IggyException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new IggyException with the specified message and cause.
+ *
+ * @param message the detail message
+ * @param cause the cause of the exception
+ */
+ protected IggyException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ /**
+ * Constructs a new IggyException with the specified cause.
+ *
+ * @param cause the cause of the exception
+ */
+ protected IggyException(Throwable cause) {
+ super(cause);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyInvalidArgumentException.java
similarity index 51%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyInvalidArgumentException.java
index 3dfc40792..94ad4dbe0 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyInvalidArgumentException.java
@@ -17,30 +17,32 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when an invalid argument is passed to the SDK.
+ *
+ * <p>This is used for client-side argument validation, such as invalid identifiers,
+ * out-of-range values, or improperly formatted inputs.
*/
-public interface UsersClient {
+public class IggyInvalidArgumentException extends IggyClientException {
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyInvalidArgumentException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyInvalidArgumentException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyInvalidArgumentException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyInvalidArgumentException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyMalformedResponseException.java
similarity index 55%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyMalformedResponseException.java
index 3dfc40792..de0e171f6 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyMalformedResponseException.java
@@ -17,30 +17,29 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when a response cannot be parsed or is in an invalid format.
*/
-public interface UsersClient {
+public class IggyMalformedResponseException extends IggyProtocolException {
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyMalformedResponseException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyMalformedResponseException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyMalformedResponseException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyMalformedResponseException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyMissingCredentialsException.java
similarity index 50%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyMissingCredentialsException.java
index 3dfc40792..445295441 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyMissingCredentialsException.java
@@ -17,30 +17,32 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when credentials are required but not provided.
+ *
+ * <p>This is thrown when attempting to login without providing credentials,
+ * either through the builder or the login method.
*/
-public interface UsersClient {
+public class IggyMissingCredentialsException extends IggyClientException {
+
+ private static final String DEFAULT_MESSAGE =
+ "No credentials provided. Use login(username, password) or provide credentials when building the client.";
/**
- * Logs in to the server with the specified username and password.
- *
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * Constructs a new IggyMissingCredentialsException with a default message.
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyMissingCredentialsException() {
+ super(DEFAULT_MESSAGE);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyMissingCredentialsException with the specified message.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyMissingCredentialsException(String message) {
+ super(message);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/error/IggyHttpException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyNotConnectedException.java
similarity index 57%
rename from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/error/IggyHttpException.java
rename to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyNotConnectedException.java
index 4d6230863..99037f53d 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/error/IggyHttpException.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyNotConnectedException.java
@@ -17,17 +17,26 @@
* under the License.
*/
-package org.apache.iggy.client.blocking.http.error;
+package org.apache.iggy.exception;
-public class IggyHttpException extends RuntimeException {
- private final IggyHttpError error;
+/**
+ * Exception thrown when an operation is attempted on a client that is not connected.
+ */
+public class IggyNotConnectedException extends IggyClientException {
- public IggyHttpException(IggyHttpError error) {
- super(error.reason());
- this.error = error;
+ /**
+ * Constructs a new IggyNotConnectedException with a default message.
+ */
+ public IggyNotConnectedException() {
+ super("Client not connected. Call connect() first.");
}
- public IggyHttpError getError() {
- return error;
+ /**
+ * Constructs a new IggyNotConnectedException with the specified message.
+ *
+ * @param message the detail message
+ */
+ public IggyNotConnectedException(String message) {
+ super(message);
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyOperationNotSupportedException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyOperationNotSupportedException.java
new file mode 100644
index 000000000..4ef053c27
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyOperationNotSupportedException.java
@@ -0,0 +1,72 @@
+/*
+ * 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.exception;
+
+/**
+ * Exception thrown when an operation is not supported by the current client implementation.
+ *
+ * <p>For example, certain operations may only be available via TCP and not HTTP.
+ */
+public class IggyOperationNotSupportedException extends IggyException {
+
+ private final String operation;
+ private final String clientType;
+
+ /**
+ * Constructs a new IggyOperationNotSupportedException.
+ *
+ * @param operation the operation that is not supported
+ * @param clientType the type of client (e.g., "HTTP", "TCP")
+ */
+ public IggyOperationNotSupportedException(String operation, String clientType) {
+ super("Operation '" + operation + "' is not supported by the " + clientType + " client");
+ this.operation = operation;
+ this.clientType = clientType;
+ }
+
+ /**
+ * Constructs a new IggyOperationNotSupportedException with a custom message.
+ *
+ * @param message the detail message
+ */
+ public IggyOperationNotSupportedException(String message) {
+ super(message);
+ this.operation = null;
+ this.clientType = null;
+ }
+
+ /**
+ * Returns the operation that is not supported.
+ *
+ * @return the operation name, or null if not specified
+ */
+ public String getOperation() {
+ return operation;
+ }
+
+ /**
+ * Returns the client type that doesn't support the operation.
+ *
+ * @return the client type, or null if not specified
+ */
+ public String getClientType() {
+ return clientType;
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyProtocolException.java
similarity index 53%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyProtocolException.java
index 3dfc40792..8fa11e3e5 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyProtocolException.java
@@ -17,30 +17,32 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when a protocol-level error occurs.
+ *
+ * <p>This is the base class for protocol-related exceptions, including
+ * empty responses and malformed responses.
*/
-public interface UsersClient {
+public class IggyProtocolException extends IggyException {
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyProtocolException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyProtocolException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyProtocolException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyProtocolException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyResourceNotFoundException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyResourceNotFoundException.java
new file mode 100644
index 000000000..00058d5db
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyResourceNotFoundException.java
@@ -0,0 +1,49 @@
+/*
+ * 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.exception;
+
+import java.util.Optional;
+
+/**
+ * Exception thrown when a requested resource is not found on the server.
+ *
+ * <p>This corresponds to 404-type errors, such as stream not found, topic not found,
+ * partition not found, consumer group not found, etc.
+ */
+public class IggyResourceNotFoundException extends IggyServerException {
+
+ /**
+ * Constructs a new IggyResourceNotFoundException.
+ *
+ * @param errorCode the error code enum
+ * @param rawErrorCode the raw numeric error code from the server
+ * @param reason the error reason/message
+ * @param field the optional field related to the error
+ * @param errorId the optional error ID for correlation with server logs
+ */
+ public IggyResourceNotFoundException(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ super(errorCode, rawErrorCode, reason, field, errorId);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyServerException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyServerException.java
new file mode 100644
index 000000000..ebfec159f
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyServerException.java
@@ -0,0 +1,255 @@
+/*
+ * 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.exception;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.EnumSet;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * Exception thrown when the server returns an error response.
+ *
+ * <p>This exception carries the error code, reason, and optional field information
+ * from the server response. The factory methods automatically map error codes to
+ * more specific exception subclasses where appropriate.
+ */
+public class IggyServerException extends IggyException {
+
+ private static final Set<IggyErrorCode> NOT_FOUND_CODES = EnumSet.of(
+ IggyErrorCode.RESOURCE_NOT_FOUND,
+ IggyErrorCode.CANNOT_LOAD_RESOURCE,
+ IggyErrorCode.STREAM_ID_NOT_FOUND,
+ IggyErrorCode.STREAM_NAME_NOT_FOUND,
+ IggyErrorCode.TOPIC_ID_NOT_FOUND,
+ IggyErrorCode.TOPIC_NAME_NOT_FOUND,
+ IggyErrorCode.PARTITION_NOT_FOUND,
+ IggyErrorCode.SEGMENT_NOT_FOUND,
+ IggyErrorCode.CLIENT_NOT_FOUND,
+ IggyErrorCode.CONSUMER_GROUP_ID_NOT_FOUND,
+ IggyErrorCode.CONSUMER_GROUP_NAME_NOT_FOUND,
+ IggyErrorCode.CONSUMER_GROUP_NOT_JOINED,
+ IggyErrorCode.MESSAGE_NOT_FOUND);
+
+ private static final Set<IggyErrorCode> AUTHENTICATION_CODES = EnumSet.of(
+ IggyErrorCode.UNAUTHENTICATED,
+ IggyErrorCode.INVALID_CREDENTIALS,
+ IggyErrorCode.INVALID_USERNAME,
+ IggyErrorCode.INVALID_PASSWORD,
+ IggyErrorCode.INVALID_PAT_TOKEN,
+ IggyErrorCode.PASSWORD_DOES_NOT_MATCH,
+ IggyErrorCode.PASSWORD_HASH_INTERNAL_ERROR);
+
+ private static final Set<IggyErrorCode> AUTHORIZATION_CODES = EnumSet.of(IggyErrorCode.UNAUTHORIZED);
+
+ private static final Set<IggyErrorCode> CONFLICT_CODES = EnumSet.of(
+ IggyErrorCode.USER_ALREADY_EXISTS,
+ IggyErrorCode.CLIENT_ALREADY_EXISTS,
+ IggyErrorCode.STREAM_ALREADY_EXISTS,
+ IggyErrorCode.TOPIC_ALREADY_EXISTS,
+ IggyErrorCode.CONSUMER_GROUP_ALREADY_EXISTS,
+ IggyErrorCode.PAT_NAME_ALREADY_EXISTS);
+
+ private static final Set<IggyErrorCode> VALIDATION_CODES = EnumSet.of(
+ IggyErrorCode.INVALID_COMMAND,
+ IggyErrorCode.INVALID_FORMAT,
+ IggyErrorCode.FEATURE_UNAVAILABLE,
+ IggyErrorCode.CANNOT_PARSE_INT,
+ IggyErrorCode.CANNOT_PARSE_SLICE,
+ IggyErrorCode.CANNOT_PARSE_UTF8,
+ IggyErrorCode.INVALID_STREAM_NAME,
+ IggyErrorCode.CANNOT_CREATE_STREAM_DIRECTORY,
+ IggyErrorCode.INVALID_TOPIC_NAME,
+ IggyErrorCode.INVALID_REPLICATION_FACTOR,
+ IggyErrorCode.CANNOT_CREATE_TOPIC_DIRECTORY,
+ IggyErrorCode.CONSUMER_GROUP_MEMBER_NOT_FOUND,
+ IggyErrorCode.INVALID_CONSUMER_GROUP_NAME,
+ IggyErrorCode.TOO_MANY_MESSAGES,
+ IggyErrorCode.EMPTY_MESSAGES,
+ IggyErrorCode.TOO_BIG_MESSAGE,
+ IggyErrorCode.INVALID_MESSAGE_CHECKSUM);
+
+ private final IggyErrorCode errorCode;
+ private final int rawErrorCode;
+ private final String reason;
+ private final Optional<String> field;
+ private final Optional<String> errorId;
+
+ /**
+ * Constructs a new IggyServerException.
+ *
+ * @param errorCode the error code enum
+ * @param rawErrorCode the raw numeric error code from the server
+ * @param reason the error reason/message
+ * @param field the optional field related to the error
+ * @param errorId the optional error ID for correlation with server logs
+ */
+ public IggyServerException(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ super(buildMessage(errorCode, rawErrorCode, reason, field, errorId));
+ this.errorCode = errorCode;
+ this.rawErrorCode = rawErrorCode;
+ this.reason = reason;
+ this.field = field;
+ this.errorId = errorId;
+ }
+
+ /**
+ * Constructs a new IggyServerException with just a status code.
+ *
+ * @param rawErrorCode the raw numeric error code from the server
+ */
+ public IggyServerException(int rawErrorCode) {
+ this(IggyErrorCode.fromCode(rawErrorCode), rawErrorCode, "Server error", Optional.empty(), Optional.empty());
+ }
+
+ /**
+ * Returns the error code enum.
+ *
+ * @return the error code
+ */
+ public IggyErrorCode getErrorCode() {
+ return errorCode;
+ }
+
+ /**
+ * Returns the raw numeric error code from the server.
+ *
+ * @return the raw error code
+ */
+ public int getRawErrorCode() {
+ return rawErrorCode;
+ }
+
+ /**
+ * Returns the error reason/message.
+ *
+ * @return the reason
+ */
+ public String getReason() {
+ return reason;
+ }
+
+ /**
+ * Returns the optional field related to the error.
+ *
+ * @return the field, if present
+ */
+ public Optional<String> getField() {
+ return field;
+ }
+
+ /**
+ * Returns the optional error ID for correlation with server logs.
+ *
+ * <p>This ID is only available for HTTP responses and can be used to find
+ * the corresponding error in server logs.
+ *
+ * @return the error ID, if present
+ */
+ public Optional<String> getErrorId() {
+ return errorId;
+ }
+
+ /**
+ * Creates an appropriate exception from a TCP response.
+ *
+ * @param status the status code from the TCP response
+ * @param payload the error payload bytes (may contain error message)
+ * @return an appropriate IggyServerException subclass
+ */
+ public static IggyServerException fromTcpResponse(long status, byte[] payload) {
+ int errorCode = (int) status;
+ String reason =
+ payload != null && payload.length > 0 ? new String(payload, StandardCharsets.UTF_8) : "Server error";
+ return createFromCode(errorCode, reason, Optional.empty(), Optional.empty());
+ }
+
+ /**
+ * Creates an appropriate exception from an HTTP response.
+ *
+ * @param id the error ID for correlation with server logs
+ * @param code the error code string
+ * @param reason the error reason
+ * @param field the optional field related to the error
+ * @return an appropriate IggyServerException subclass
+ */
+ public static IggyServerException fromHttpResponse(String id, String code, String reason, String field) {
+ IggyErrorCode errorCode = IggyErrorCode.fromString(code);
+ int rawCode = errorCode.getCode();
+ if (rawCode == -1) {
+ // Try parsing code as integer
+ try {
+ rawCode = Integer.parseInt(code);
+ } catch (NumberFormatException e) {
+ rawCode = -1;
+ }
+ }
+ Optional<String> fieldOpt = Optional.ofNullable(StringUtils.stripToNull(field));
+ Optional<String> errorIdOpt = Optional.ofNullable(StringUtils.stripToNull(id));
+ return createFromCode(rawCode, reason != null ? reason : "Server error", fieldOpt, errorIdOpt);
+ }
+
+ private static IggyServerException createFromCode(
+ int code, String reason, Optional<String> field, Optional<String> errorId) {
+ IggyErrorCode errorCode = IggyErrorCode.fromCode(code);
+
+ if (NOT_FOUND_CODES.contains(errorCode)) {
+ return new IggyResourceNotFoundException(errorCode, code, reason, field, errorId);
+ }
+ if (AUTHENTICATION_CODES.contains(errorCode)) {
+ return new IggyAuthenticationException(errorCode, code, reason, field, errorId);
+ }
+ if (AUTHORIZATION_CODES.contains(errorCode)) {
+ return new IggyAuthorizationException(errorCode, code, reason, field, errorId);
+ }
+ if (CONFLICT_CODES.contains(errorCode)) {
+ return new IggyConflictException(errorCode, code, reason, field, errorId);
+ }
+ if (VALIDATION_CODES.contains(errorCode)) {
+ return new IggyValidationException(errorCode, code, reason, field, errorId);
+ }
+
+ return new IggyServerException(errorCode, code, reason, field, errorId);
+ }
+
+ private static String buildMessage(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ StringBuilder sb = new StringBuilder();
+ sb.append("Server error [code=").append(rawErrorCode);
+ if (errorCode != IggyErrorCode.UNKNOWN) {
+ sb.append(" (").append(errorCode.name()).append(")");
+ }
+ sb.append("]: ").append(reason);
+ field.ifPresent(f -> sb.append(" (field: ").append(f).append(")"));
+ errorId.ifPresent(id -> sb.append(" [errorId: ").append(id).append("]"));
+ return sb.toString();
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyTimeoutException.java
similarity index 52%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyTimeoutException.java
index 3dfc40792..fc9eb41a4 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyTimeoutException.java
@@ -17,30 +17,36 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when an operation times out.
*/
-public interface UsersClient {
+public class IggyTimeoutException extends IggyException {
+
+ /**
+ * Constructs a new IggyTimeoutException with a default message.
+ */
+ public IggyTimeoutException() {
+ super("Operation timed out");
+ }
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyTimeoutException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyTimeoutException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyTimeoutException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyTimeoutException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyTlsException.java
similarity index 54%
copy from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
copy to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyTlsException.java
index 3dfc40792..8a703caf0 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/UsersClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyTlsException.java
@@ -17,30 +17,32 @@
* under the License.
*/
-package org.apache.iggy.client.async;
-
-import org.apache.iggy.user.IdentityInfo;
-
-import java.util.concurrent.CompletableFuture;
+package org.apache.iggy.exception;
/**
- * Async client for user management operations.
+ * Exception thrown when a TLS/SSL-related error occurs.
+ *
+ * <p>This includes errors during TLS context setup, certificate validation,
+ * and SSL handshake failures.
*/
-public interface UsersClient {
+public class IggyTlsException extends IggyConnectionException {
/**
- * Logs in to the server with the specified username and password.
+ * Constructs a new IggyTlsException with the specified message.
*
- * @param username The username to login with
- * @param password The password to login with
- * @return A CompletableFuture that completes with the user's identity information
+ * @param message the detail message
*/
- CompletableFuture<IdentityInfo> loginAsync(String username, String password);
+ public IggyTlsException(String message) {
+ super(message);
+ }
/**
- * Logs out from the server.
+ * Constructs a new IggyTlsException with the specified message and cause.
*
- * @return A CompletableFuture that completes when logout is successful
+ * @param message the detail message
+ * @param cause the cause of the exception
*/
- CompletableFuture<Void> logoutAsync();
+ public IggyTlsException(String message, Throwable cause) {
+ super(message, cause);
+ }
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyValidationException.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyValidationException.java
new file mode 100644
index 000000000..bdd270d25
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/IggyValidationException.java
@@ -0,0 +1,49 @@
+/*
+ * 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.exception;
+
+import java.util.Optional;
+
+/**
+ * Exception thrown when input validation fails.
+ *
+ * <p>This includes errors such as invalid command format, invalid names,
+ * message validation failures, etc.
+ */
+public class IggyValidationException extends IggyServerException {
+
+ /**
+ * Constructs a new IggyValidationException.
+ *
+ * @param errorCode the error code enum
+ * @param rawErrorCode the raw numeric error code from the server
+ * @param reason the error reason/message
+ * @param field the optional field related to the error
+ * @param errorId the optional error ID for correlation with server logs
+ */
+ public IggyValidationException(
+ IggyErrorCode errorCode,
+ int rawErrorCode,
+ String reason,
+ Optional<String> field,
+ Optional<String> errorId) {
+ super(errorCode, rawErrorCode, reason, field, errorId);
+ }
+}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/error/IggyHttpError.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/package-info.java
similarity index 52%
rename from foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/error/IggyHttpError.java
rename to foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/package-info.java
index f84e1d8f8..f07549365 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/error/IggyHttpError.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/exception/package-info.java
@@ -17,6 +17,19 @@
* under the License.
*/
-package org.apache.iggy.client.blocking.http.error;
+/**
+ * Exception hierarchy for the Iggy Java SDK.
+ *
+ * <p>This package provides a unified exception hierarchy for all SDK operations:
+ * <ul>
+ * <li>{@link org.apache.iggy.exception.IggyException} - Base class for all Iggy exceptions</li>
+ * <li>{@link org.apache.iggy.exception.IggyConnectionException} - Connection-related failures</li>
+ * <li>{@link org.apache.iggy.exception.IggyProtocolException} - Protocol-level errors</li>
+ * <li>{@link org.apache.iggy.exception.IggyServerException} - Server-side errors with error codes</li>
+ * <li>{@link org.apache.iggy.exception.IggyClientException} - Client-side errors</li>
+ * </ul>
+ */
+@NonNullApi
+package org.apache.iggy.exception;
-public record IggyHttpError(String id, String code, String reason, String field) {}
+import org.apache.iggy.NonNullApi;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java
index ac5c31b14..945d35696 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java
@@ -20,6 +20,7 @@
package org.apache.iggy.identifier;
import org.apache.commons.lang3.StringUtils;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
import javax.annotation.Nullable;
@@ -30,10 +31,10 @@ public abstract class Identifier {
protected Identifier(@Nullable String name, @Nullable Long id) {
if (StringUtils.isBlank(name) && id == null) {
- throw new IllegalArgumentException("Name and id cannot be blank");
+ throw new IggyInvalidArgumentException("Name and id cannot be blank");
}
if (StringUtils.isNotBlank(name) && id != null) {
- throw new IllegalArgumentException("Name and id cannot be both present");
+ throw new IggyInvalidArgumentException("Name and id cannot be both present");
}
if (StringUtils.isNotBlank(name)) {
this.name = name;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/BytesMessageId.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/BytesMessageId.java
index f1e737c46..48cc38136 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/BytesMessageId.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/BytesMessageId.java
@@ -21,6 +21,7 @@ package org.apache.iggy.message;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
import java.math.BigInteger;
import java.util.Arrays;
@@ -31,7 +32,7 @@ public class BytesMessageId implements MessageId {
public BytesMessageId(byte[] value) {
if (value.length != 16) {
- throw new IllegalArgumentException("Message id must have 16 bytes");
+ throw new IggyInvalidArgumentException("Message id must have 16 bytes");
}
this.value = value;
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderKind.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderKind.java
index 23ef5d114..00f024c25 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderKind.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderKind.java
@@ -19,6 +19,8 @@
package org.apache.iggy.message;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+
public enum HeaderKind {
Raw(1),
String(2),
@@ -48,7 +50,7 @@ public enum HeaderKind {
return kind;
}
}
- throw new IllegalArgumentException("Unknown header kind: " + code);
+ throw new IggyInvalidArgumentException("Unknown header kind: " + code);
}
public int asCode() {
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/MessageState.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/MessageState.java
index b694f7c82..32212f0dd 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/MessageState.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/MessageState.java
@@ -19,6 +19,8 @@
package org.apache.iggy.message;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+
public enum MessageState {
Available(1),
Unavailable(10),
@@ -37,6 +39,6 @@ public enum MessageState {
return state;
}
}
- throw new IllegalArgumentException("Unknown message state code: " + code);
+ throw new IggyInvalidArgumentException("Unknown message state code: " + code);
}
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java
index 424fc78c3..c725eb4ee 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java
@@ -20,6 +20,7 @@
package org.apache.iggy.message;
import org.apache.commons.lang3.ArrayUtils;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
import java.nio.ByteBuffer;
@@ -39,7 +40,7 @@ public record Partitioning(PartitioningKind kind, byte[] value) {
public static Partitioning messagesKey(String key) {
if (key == null || key.isBlank() || key.length() > 255) {
- throw new IllegalArgumentException("Key must be non-empty and less than 255 characters long");
+ throw new IggyInvalidArgumentException("Key must be non-empty and less than 255 characters long");
}
return new Partitioning(PartitioningKind.MessagesKey, key.getBytes());
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
index 4014f3914..9971575ee 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java
@@ -23,6 +23,7 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
import org.apache.iggy.identifier.Identifier;
import org.apache.iggy.message.HeaderValue;
import org.apache.iggy.message.Message;
@@ -68,7 +69,7 @@ public final class BytesSerializer {
buffer.writeBytes(identifier.getName().getBytes());
return buffer;
} else {
- throw new IllegalArgumentException("Unknown identifier kind: " + identifier.getKind());
+ throw new IggyInvalidArgumentException("Unknown identifier kind: " + identifier.getKind());
}
}
@@ -212,12 +213,12 @@ public final class BytesSerializer {
public static ByteBuf toBytesAsU64(BigInteger value) {
if (value.signum() == -1) {
- throw new IllegalArgumentException("Negative value cannot be serialized to unsigned 64: " + value);
+ throw new IggyInvalidArgumentException("Negative value cannot be serialized to unsigned 64: " + value);
}
ByteBuf buffer = Unpooled.buffer(8, 8);
byte[] valueAsBytes = value.toByteArray();
if (valueAsBytes.length > 9 || valueAsBytes.length == 9 && valueAsBytes[0] != 0) {
- throw new IllegalArgumentException("Value too large for U64: " + value);
+ throw new IggyInvalidArgumentException("Value too large for U64: " + value);
}
ArrayUtils.reverse(valueAsBytes);
buffer.writeBytes(valueAsBytes, 0, Math.min(8, valueAsBytes.length));
@@ -229,12 +230,12 @@ public final class BytesSerializer {
public static ByteBuf toBytesAsU128(BigInteger value) {
if (value.signum() == -1) {
- throw new IllegalArgumentException("Negative value cannot be serialized to unsigned 128: " + value);
+ throw new IggyInvalidArgumentException("Negative value cannot be serialized to unsigned 128: " + value);
}
ByteBuf buffer = Unpooled.buffer(16, 16);
byte[] valueAsBytes = value.toByteArray();
if (valueAsBytes.length > 17 || valueAsBytes.length == 17 && valueAsBytes[0] != 0) {
- throw new IllegalArgumentException("Value too large for U128: " + value);
+ throw new IggyInvalidArgumentException("Value too large for U128: " + value);
}
ArrayUtils.reverse(valueAsBytes);
buffer.writeBytes(valueAsBytes, 0, Math.min(16, valueAsBytes.length));
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/topic/CompressionAlgorithm.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/topic/CompressionAlgorithm.java
index 5f4684023..7335bfe78 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/topic/CompressionAlgorithm.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/topic/CompressionAlgorithm.java
@@ -19,6 +19,8 @@
package org.apache.iggy.topic;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+
public enum CompressionAlgorithm {
None(1),
Gzip(2);
@@ -35,7 +37,7 @@ public enum CompressionAlgorithm {
return algorithm;
}
}
- throw new IllegalArgumentException();
+ throw new IggyInvalidArgumentException("Unknown compression algorithm code: " + code);
}
public Integer asCode() {
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/user/UserStatus.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/user/UserStatus.java
index eb6660f2e..76b215043 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/user/UserStatus.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/user/UserStatus.java
@@ -20,6 +20,7 @@
package org.apache.iggy.user;
import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
public enum UserStatus {
@JsonProperty("active")
@@ -39,7 +40,7 @@ public enum UserStatus {
return userStatus;
}
}
- throw new IllegalArgumentException("Invalid user status: " + code);
+ throw new IggyInvalidArgumentException("Invalid user status: " + code);
}
public int asCode() {
diff --git a/foreign/java/java-sdk/src/main/resources/iggy-version.properties b/foreign/java/java-sdk/src/main/resources/iggy-version.properties
new file mode 100644
index 000000000..64c4f17a2
--- /dev/null
+++ b/foreign/java/java-sdk/src/main/resources/iggy-version.properties
@@ -0,0 +1,3 @@
+version=${version}
+buildTime=${buildTime}
+gitCommit=${gitCommit}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/IggyTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/IggyTest.java
index f1b3fb197..dc5d2eecb 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/IggyTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/IggyTest.java
@@ -19,7 +19,11 @@
package org.apache.iggy;
-import org.apache.iggy.client.blocking.http.IggyHttpClient;
+import org.apache.iggy.builder.HttpClientBuilder;
+import org.apache.iggy.builder.TcpClientBuilder;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClientBuilder;
+import org.apache.iggy.client.blocking.http.IggyHttpClientBuilder;
+import org.apache.iggy.client.blocking.tcp.IggyTcpClientBuilder;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -27,11 +31,71 @@ import static org.assertj.core.api.Assertions.assertThat;
class IggyTest {
@Test
- void test() {
- var baseClient = new IggyHttpClient("http://localhost:8080");
+ void tcpBuilderReturnsCorrectTypes() {
+ TcpClientBuilder tcpBuilder = Iggy.tcp();
+ assertThat(tcpBuilder).isNotNull();
- var client = Iggy.clientBuilder().withBaseClient(baseClient).build();
+ IggyTcpClientBuilder blockingBuilder = tcpBuilder.blocking();
+ assertThat(blockingBuilder).isNotNull();
- assertThat(client).isNotNull();
+ AsyncIggyTcpClientBuilder asyncBuilder = tcpBuilder.async();
+ assertThat(asyncBuilder).isNotNull();
+ }
+
+ @Test
+ void httpBuilderReturnsCorrectTypes() {
+ HttpClientBuilder httpBuilder = Iggy.http();
+ assertThat(httpBuilder).isNotNull();
+
+ IggyHttpClientBuilder blockingBuilder = httpBuilder.blocking();
+ assertThat(blockingBuilder).isNotNull();
+ }
+
+ @Test
+ void versionMethodsReturnValues() {
+ String version = Iggy.version();
+ assertThat(version).isNotNull();
+ assertThat(version).isNotEqualTo("unknown");
+
+ IggyVersion versionInfo = Iggy.versionInfo();
+ assertThat(versionInfo).isNotNull();
+ assertThat(versionInfo.getVersion()).isNotNull().isNotEqualTo("unknown");
+ assertThat(versionInfo.getBuildTime()).isNotNull().isNotEqualTo("unknown");
+ assertThat(versionInfo.getGitCommit()).isNotNull().isNotEqualTo("unknown");
+ assertThat(versionInfo.getUserAgent()).startsWith("iggy-java-sdk/");
+ }
+
+ @Test
+ void tcpBlockingBuilderHasFluentApi() {
+ IggyTcpClientBuilder builder = Iggy.tcp().blocking();
+
+ // Verify fluent API returns same builder
+ assertThat(builder.host("localhost")).isSameAs(builder);
+ assertThat(builder.port(8090)).isSameAs(builder);
+ assertThat(builder.enableTls()).isSameAs(builder);
+ assertThat(builder.tls(false)).isSameAs(builder);
+ }
+
+ @Test
+ void tcpAsyncBuilderHasFluentApi() {
+ AsyncIggyTcpClientBuilder builder = Iggy.tcp().async();
+
+ // Verify fluent API returns same builder
+ assertThat(builder.host("localhost")).isSameAs(builder);
+ assertThat(builder.port(8090)).isSameAs(builder);
+ assertThat(builder.enableTls()).isSameAs(builder);
+ assertThat(builder.tls(false)).isSameAs(builder);
+ }
+
+ @Test
+ void httpBlockingBuilderHasFluentApi() {
+ IggyHttpClientBuilder builder = Iggy.http().blocking();
+
+ // Verify fluent API returns same builder
+ assertThat(builder.host("localhost")).isSameAs(builder);
+ assertThat(builder.port(3000)).isSameAs(builder);
+ assertThat(builder.url("http://localhost:3000")).isSameAs(builder);
+ assertThat(builder.enableTls()).isSameAs(builder);
+ assertThat(builder.tls(false)).isSameAs(builder);
}
}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncClientIntegrationTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncClientIntegrationTest.java
index 0e8038d4a..d90b010d0 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncClientIntegrationTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncClientIntegrationTest.java
@@ -75,7 +75,7 @@ class AsyncClientIntegrationTest {
client.connect()
.thenCompose(v -> {
log.info("Connected to Iggy server");
- return client.users().loginAsync(USERNAME, PASSWORD);
+ return client.users().login(USERNAME, PASSWORD);
})
.get(5, TimeUnit.SECONDS);
@@ -88,7 +88,7 @@ class AsyncClientIntegrationTest {
try {
// Clean up test stream if it exists
- client.streams().deleteStreamAsync(StreamId.of(TEST_STREAM)).get(5, TimeUnit.SECONDS);
+ client.streams().deleteStream(StreamId.of(TEST_STREAM)).get(5, TimeUnit.SECONDS);
log.info("Deleted test stream: {}", TEST_STREAM);
} catch (RuntimeException e) {
// Stream may not exist, which is fine
@@ -107,7 +107,7 @@ class AsyncClientIntegrationTest {
public void testCreateStream() throws Exception {
log.info("Testing stream creation");
- var streamDetails = client.streams().createStreamAsync(TEST_STREAM).get(5, TimeUnit.SECONDS);
+ var streamDetails = client.streams().createStream(TEST_STREAM).get(5, TimeUnit.SECONDS);
assertThat(streamDetails).isNotNull();
assertThat(streamDetails.name()).isEqualTo(TEST_STREAM);
@@ -119,8 +119,7 @@ class AsyncClientIntegrationTest {
public void testGetStream() throws Exception {
log.info("Testing stream retrieval");
- var streamOpt =
- client.streams().getStreamAsync(StreamId.of(TEST_STREAM)).get(5, TimeUnit.SECONDS);
+ var streamOpt = client.streams().getStream(StreamId.of(TEST_STREAM)).get(5, TimeUnit.SECONDS);
assertThat(streamOpt).isPresent();
assertThat(streamOpt.get().name()).isEqualTo(TEST_STREAM);
@@ -133,7 +132,7 @@ class AsyncClientIntegrationTest {
log.info("Testing topic creation");
var topicDetails = client.topics()
- .createTopicAsync(
+ .createTopic(
StreamId.of(TEST_STREAM),
2L, // 2 partitions
CompressionAlgorithm.None,
@@ -158,7 +157,7 @@ class AsyncClientIntegrationTest {
log.info("Testing topic retrieval");
var topicOpt = client.topics()
- .getTopicAsync(StreamId.of(TEST_STREAM), TopicId.of(TEST_TOPIC))
+ .getTopic(StreamId.of(TEST_STREAM), TopicId.of(TEST_TOPIC))
.get(5, TimeUnit.SECONDS);
assertThat(topicOpt).isPresent();
@@ -179,7 +178,7 @@ class AsyncClientIntegrationTest {
// Send messages to partition 1
client.messages()
- .sendMessagesAsync(
+ .sendMessages(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
Partitioning.partitionId(PARTITION_ID),
@@ -197,7 +196,7 @@ class AsyncClientIntegrationTest {
// Poll messages from partition 1 - Use valid consumer instead of null
var consumer = Consumer.of(12345L); // Create consumer with ID
var polledMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
Optional.of(PARTITION_ID),
@@ -244,7 +243,7 @@ class AsyncClientIntegrationTest {
}
var future = client.messages()
- .sendMessagesAsync(
+ .sendMessages(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
Partitioning.partitionId(PARTITION_ID),
@@ -259,7 +258,7 @@ class AsyncClientIntegrationTest {
// Poll all messages
var polledMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
Optional.of(PARTITION_ID),
@@ -284,7 +283,7 @@ class AsyncClientIntegrationTest {
// Update topic with new compression algorithm
client.topics()
- .updateTopicAsync(
+ .updateTopic(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
CompressionAlgorithm.Gzip,
@@ -296,7 +295,7 @@ class AsyncClientIntegrationTest {
// Verify the update
var updatedTopic = client.topics()
- .getTopicAsync(StreamId.of(TEST_STREAM), TopicId.of(TEST_TOPIC))
+ .getTopic(StreamId.of(TEST_STREAM), TopicId.of(TEST_TOPIC))
.get(5, TimeUnit.SECONDS);
assertThat(updatedTopic).isPresent();
@@ -317,7 +316,7 @@ class AsyncClientIntegrationTest {
// Create a temporary topic to delete
String tempTopic = "temp-topic-" + UUID.randomUUID();
client.topics()
- .createTopicAsync(
+ .createTopic(
StreamId.of(TEST_STREAM),
1L,
CompressionAlgorithm.None,
@@ -329,12 +328,12 @@ class AsyncClientIntegrationTest {
// Delete the topic
client.topics()
- .deleteTopicAsync(StreamId.of(TEST_STREAM), TopicId.of(tempTopic))
+ .deleteTopic(StreamId.of(TEST_STREAM), TopicId.of(tempTopic))
.get(5, TimeUnit.SECONDS);
// Verify deletion
var deletedTopic = client.topics()
- .getTopicAsync(StreamId.of(TEST_STREAM), TopicId.of(tempTopic))
+ .getTopic(StreamId.of(TEST_STREAM), TopicId.of(tempTopic))
.get(5, TimeUnit.SECONDS);
assertThat(deletedTopic).isNotPresent();
@@ -361,7 +360,7 @@ class AsyncClientIntegrationTest {
return messages;
})
.thenCompose(messages -> client.messages()
- .sendMessagesAsync(
+ .sendMessages(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
Partitioning.partitionId(PARTITION_ID),
@@ -373,7 +372,7 @@ class AsyncClientIntegrationTest {
for (int i = 0; i < 3; i++) {
final long consumerId = 10000L + i;
var future = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(TEST_STREAM),
TopicId.of(TEST_TOPIC),
Optional.of(PARTITION_ID),
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncPollMessageTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncPollMessageTest.java
index e2e6b979f..f66aa5686 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncPollMessageTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncPollMessageTest.java
@@ -81,7 +81,7 @@ public class AsyncPollMessageTest {
}
client = new AsyncIggyTcpClient("127.0.0.1", 8090);
client.connect().get(5, TimeUnit.SECONDS);
- client.users().loginAsync("iggy", "iggy").get(5, TimeUnit.SECONDS);
+ client.users().login("iggy", "iggy").get(5, TimeUnit.SECONDS);
log.info("Client reconnected successfully");
}
}
@@ -89,7 +89,7 @@ public class AsyncPollMessageTest {
private boolean isConnected(AsyncIggyTcpClient client) {
// Check if client is connected by attempting a simple operation
try {
- client.streams().getStreamsAsync().get(1, TimeUnit.SECONDS);
+ client.streams().getStreams().get(1, TimeUnit.SECONDS);
return true;
} catch (InterruptedException | ExecutionException | TimeoutException e) {
return false;
@@ -103,17 +103,17 @@ public class AsyncPollMessageTest {
// Initialize client
client = new AsyncIggyTcpClient("127.0.0.1", 8090);
client.connect().get(5, TimeUnit.SECONDS);
- client.users().loginAsync("iggy", "iggy").get(5, TimeUnit.SECONDS);
+ client.users().login("iggy", "iggy").get(5, TimeUnit.SECONDS);
log.info("Successfully connected and logged in");
// Create unique stream for this test run
testStream = "poll-test-stream-" + UUID.randomUUID();
- var stream = client.streams().createStreamAsync(testStream).get(5, TimeUnit.SECONDS);
+ var stream = client.streams().createStream(testStream).get(5, TimeUnit.SECONDS);
log.info("Created test stream: {}", stream.name());
// Create topic with 2 partitions
var topic = client.topics()
- .createTopicAsync(
+ .createTopic(
StreamId.of(testStream),
2L,
CompressionAlgorithm.None,
@@ -132,7 +132,7 @@ public class AsyncPollMessageTest {
.toList();
client.messages()
- .sendMessagesAsync(
+ .sendMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Partitioning.partitionId((long) partition),
@@ -149,7 +149,7 @@ public class AsyncPollMessageTest {
// Delete stream (cascades to topics and consumer groups)
try {
- client.streams().deleteStreamAsync(StreamId.of(testStream)).get(5, TimeUnit.SECONDS);
+ client.streams().deleteStream(StreamId.of(testStream)).get(5, TimeUnit.SECONDS);
log.info("Deleted test stream: {}", testStream);
} catch (RuntimeException e) {
log.warn("Failed to delete test stream: {}", e.getMessage());
@@ -173,7 +173,7 @@ public class AsyncPollMessageTest {
try {
var polledMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Optional.of(PARTITION_ID),
@@ -204,7 +204,7 @@ public class AsyncPollMessageTest {
try {
var polledMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Optional.of(PARTITION_ID),
@@ -240,7 +240,7 @@ public class AsyncPollMessageTest {
var sessionConsumer = Consumer.of(0L);
var polledMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Optional.of(PARTITION_ID),
@@ -275,7 +275,7 @@ public class AsyncPollMessageTest {
log.info(" Testing FIRST strategy");
try {
var firstMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Optional.of(1L),
@@ -295,7 +295,7 @@ public class AsyncPollMessageTest {
log.info(" Testing OFFSET strategy");
try {
var offsetMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Optional.of(1L),
@@ -315,7 +315,7 @@ public class AsyncPollMessageTest {
log.info(" Testing LAST strategy");
try {
var lastMessages = client.messages()
- .pollMessagesAsync(
+ .pollMessages(
StreamId.of(testStream),
TopicId.of(TEST_TOPIC),
Optional.of(1L),
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilderTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilderTest.java
index ea116bdf4..830b9d0fe 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilderTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientBuilderTest.java
@@ -22,7 +22,6 @@ package org.apache.iggy.client.async.tcp;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
-import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -65,136 +64,10 @@ class AsyncIggyTcpClientBuilderTest {
assertNotNull(client.consumerGroups());
}
- @Test
- void shouldCreateClientWithCredentials() throws Exception {
- // Given: Builder with credentials configured
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .credentials("iggy", "iggy")
- .build();
-
- // When: Connect (auto-login should happen)
- CompletableFuture<Void> connectFuture = client.connect();
- connectFuture.get(5, TimeUnit.SECONDS);
-
- // Then: Should be connected and logged in
- assertNotNull(client.users());
- }
-
- @Test
- void shouldCreateClientWithTimeoutConfiguration() throws Exception {
- // Given: Builder with timeout configuration
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .connectionTimeout(Duration.ofSeconds(30))
- .requestTimeout(Duration.ofSeconds(10))
- .credentials("iggy", "iggy")
- .build();
-
- // When: Connect to server
- client.connect().get(5, TimeUnit.SECONDS);
-
- // Then: Should succeed
- assertNotNull(client.users());
- }
-
- @Test
- void shouldCreateClientWithConnectionPoolSize() throws Exception {
- // Given: Builder with connection pool size
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .connectionPoolSize(10)
- .credentials("iggy", "iggy")
- .build();
-
- // When: Connect to server
- client.connect().get(5, TimeUnit.SECONDS);
-
- // Then: Should succeed
- assertNotNull(client.users());
- }
-
- @Test
- void shouldCreateClientWithRetryPolicy() throws Exception {
- // Given: Builder with exponential backoff retry policy
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .retryPolicy(AsyncIggyTcpClient.RetryPolicy.exponentialBackoff())
- .credentials("iggy", "iggy")
- .build();
-
- // When: Connect to server
- client.connect().get(5, TimeUnit.SECONDS);
-
- // Then: Should succeed
- assertNotNull(client.users());
- }
-
- @Test
- void shouldCreateClientWithCustomRetryPolicy() throws Exception {
- // Given: Builder with custom retry policy
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .retryPolicy(AsyncIggyTcpClient.RetryPolicy.fixedDelay(5, Duration.ofMillis(500)))
- .credentials("iggy", "iggy")
- .build();
-
- // When: Connect to server
- client.connect().get(5, TimeUnit.SECONDS);
-
- // Then: Should succeed
- assertNotNull(client.users());
- }
-
- @Test
- void shouldCreateClientWithNoRetryPolicy() throws Exception {
- // Given: Builder with no retry policy
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .retryPolicy(AsyncIggyTcpClient.RetryPolicy.noRetry())
- .credentials("iggy", "iggy")
- .build();
-
- // When: Connect to server
- client.connect().get(5, TimeUnit.SECONDS);
-
- // Then: Should succeed
- assertNotNull(client.users());
- }
-
- @Test
- void shouldCreateClientWithAllOptions() throws Exception {
- // Given: Builder with all configuration options
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .credentials("iggy", "iggy")
- .connectionTimeout(Duration.ofSeconds(30))
- .requestTimeout(Duration.ofSeconds(10))
- .connectionPoolSize(10)
- .retryPolicy(AsyncIggyTcpClient.RetryPolicy.exponentialBackoff(
- 3, Duration.ofMillis(100), Duration.ofSeconds(5), 2.0))
- .build();
-
- // When: Connect to server
- client.connect().get(5, TimeUnit.SECONDS);
-
- // Then: Should succeed
- assertNotNull(client.users());
- }
-
@Test
void shouldUseDefaultValues() throws Exception {
- // Given: Builder with only credentials (should use defaults)
- client = AsyncIggyTcpClient.builder()
- .credentials("iggy", "iggy")
- .build(); // Uses default host=localhost, port=8090
+ // Given: Builder with defaults (should use host=localhost, port=8090)
+ client = AsyncIggyTcpClient.builder().build();
// When: Connect to server
client.connect().get(5, TimeUnit.SECONDS);
@@ -206,7 +79,7 @@ class AsyncIggyTcpClientBuilderTest {
@Test
void shouldThrowExceptionForEmptyHost() {
// Given: Builder with empty host
- AsyncIggyTcpClient.Builder builder =
+ AsyncIggyTcpClientBuilder builder =
AsyncIggyTcpClient.builder().host("").port(PORT);
// When/Then: Building should throw IllegalArgumentException
@@ -216,7 +89,7 @@ class AsyncIggyTcpClientBuilderTest {
@Test
void shouldThrowExceptionForNullHost() {
// Given: Builder with null host
- AsyncIggyTcpClient.Builder builder =
+ AsyncIggyTcpClientBuilder builder =
AsyncIggyTcpClient.builder().host(null).port(PORT);
// When/Then: Building should throw IllegalArgumentException
@@ -226,7 +99,7 @@ class AsyncIggyTcpClientBuilderTest {
@Test
void shouldThrowExceptionForInvalidPort() {
// Given: Builder with invalid port
- AsyncIggyTcpClient.Builder builder =
+ AsyncIggyTcpClientBuilder builder =
AsyncIggyTcpClient.builder().host(HOST).port(-1);
// When/Then: Building should throw IllegalArgumentException
@@ -236,7 +109,7 @@ class AsyncIggyTcpClientBuilderTest {
@Test
void shouldThrowExceptionForZeroPort() {
// Given: Builder with zero port
- AsyncIggyTcpClient.Builder builder =
+ AsyncIggyTcpClientBuilder builder =
AsyncIggyTcpClient.builder().host(HOST).port(0);
// When/Then: Building should throw IllegalArgumentException
@@ -257,12 +130,8 @@ class AsyncIggyTcpClientBuilderTest {
@Test
void shouldConnectAndPerformOperations() throws Exception {
- // Given: Client with credentials
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .credentials("iggy", "iggy")
- .build();
+ // Given: Client
+ client = AsyncIggyTcpClient.builder().host(HOST).port(PORT).build();
// When: Connect
client.connect().get(5, TimeUnit.SECONDS);
@@ -278,11 +147,7 @@ class AsyncIggyTcpClientBuilderTest {
@Test
void shouldCloseConnectionGracefully() throws Exception {
// Given: Connected client
- client = AsyncIggyTcpClient.builder()
- .host(HOST)
- .port(PORT)
- .credentials("iggy", "iggy")
- .build();
+ client = AsyncIggyTcpClient.builder().host(HOST).port(PORT).build();
client.connect().get(5, TimeUnit.SECONDS);
// When: Close connection
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilderTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilderTest.java
index 6f7955fd3..a9be80580 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilderTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/IggyTcpClientBuilderTest.java
@@ -21,6 +21,7 @@ package org.apache.iggy.client.blocking.tcp;
import org.apache.iggy.client.blocking.IggyBaseClient;
import org.apache.iggy.client.blocking.IntegrationTest;
+import org.apache.iggy.config.RetryPolicy;
import org.apache.iggy.system.ClientInfo;
import org.junit.jupiter.api.Test;
@@ -47,7 +48,8 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
IggyTcpClient client =
IggyTcpClient.builder().host("127.0.0.1").port(TCP_PORT).build();
- // When: Login to verify connection
+ // When: Connect and login to verify connection
+ client.connect();
client.users().login("iggy", "iggy");
// Then: Client should be able to fetch system info
@@ -55,21 +57,6 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
assertNotNull(clients);
}
- @Test
- void shouldCreateClientWithCredentials() {
- // Given: Builder with credentials configured
- IggyTcpClient client = IggyTcpClient.builder()
- .host("127.0.0.1")
- .port(TCP_PORT)
- .credentials("iggy", "iggy")
- .build();
-
- // When: Try to access system info (auto-login should have happened)
- // Then: Should succeed without explicit login
- List<ClientInfo> clients = client.system().getClients();
- assertNotNull(clients);
- }
-
@Test
void shouldCreateClientWithTimeoutConfiguration() {
// Given: Builder with timeout configuration
@@ -78,10 +65,12 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
.port(TCP_PORT)
.connectionTimeout(Duration.ofSeconds(30))
.requestTimeout(Duration.ofSeconds(10))
- .credentials("iggy", "iggy")
.build();
- // When: Perform an operation
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
+
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -94,10 +83,12 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
.host("127.0.0.1")
.port(TCP_PORT)
.connectionPoolSize(10)
- .credentials("iggy", "iggy")
.build();
- // When: Perform an operation
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
+
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -109,11 +100,13 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
IggyTcpClient client = IggyTcpClient.builder()
.host("127.0.0.1")
.port(TCP_PORT)
- .retryPolicy(IggyTcpClient.RetryPolicy.exponentialBackoff())
- .credentials("iggy", "iggy")
+ .retryPolicy(RetryPolicy.exponentialBackoff())
.build();
- // When: Perform an operation
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
+
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -125,11 +118,13 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
IggyTcpClient client = IggyTcpClient.builder()
.host("127.0.0.1")
.port(TCP_PORT)
- .retryPolicy(IggyTcpClient.RetryPolicy.fixedDelay(5, Duration.ofMillis(500)))
- .credentials("iggy", "iggy")
+ .retryPolicy(RetryPolicy.fixedDelay(5, Duration.ofMillis(500)))
.build();
- // When: Perform an operation
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
+
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -141,11 +136,13 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
IggyTcpClient client = IggyTcpClient.builder()
.host("127.0.0.1")
.port(TCP_PORT)
- .retryPolicy(IggyTcpClient.RetryPolicy.noRetry())
- .credentials("iggy", "iggy")
+ .retryPolicy(RetryPolicy.noRetry())
.build();
- // When: Perform an operation
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
+
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -157,15 +154,16 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
IggyTcpClient client = IggyTcpClient.builder()
.host("127.0.0.1")
.port(TCP_PORT)
- .credentials("iggy", "iggy")
.connectionTimeout(Duration.ofSeconds(30))
.requestTimeout(Duration.ofSeconds(10))
.connectionPoolSize(10)
- .retryPolicy(IggyTcpClient.RetryPolicy.exponentialBackoff(
- 3, Duration.ofMillis(100), Duration.ofSeconds(5), 2.0))
+ .retryPolicy(RetryPolicy.exponentialBackoff(3, Duration.ofMillis(100), Duration.ofSeconds(5), 2.0))
.build();
- // When: Perform an operation
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
+
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -174,10 +172,12 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
@Test
void shouldUseDefaultValues() {
// Given: Builder with only required fields (should use defaults)
- IggyTcpClient client =
- IggyTcpClient.builder().credentials("iggy", "iggy").build(); // Uses default host=localhost, port=8090
+ IggyTcpClient client = IggyTcpClient.builder().build(); // Uses default host=localhost, port=8090
+
+ // When: Connect, login and perform an operation
+ client.connect();
+ client.users().login("iggy", "iggy");
- // When: Perform an operation
// Then: Should succeed
List<ClientInfo> clients = client.system().getClients();
assertNotNull(clients);
@@ -186,7 +186,7 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
@Test
void shouldThrowExceptionForEmptyHost() {
// Given: Builder with empty host
- IggyTcpClient.Builder builder = IggyTcpClient.builder().host("").port(TCP_PORT);
+ IggyTcpClientBuilder builder = IggyTcpClient.builder().host("").port(TCP_PORT);
// When/Then: Building should throw IllegalArgumentException
assertThrows(IllegalArgumentException.class, builder::build);
@@ -195,7 +195,7 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
@Test
void shouldThrowExceptionForNullHost() {
// Given: Builder with null host
- IggyTcpClient.Builder builder = IggyTcpClient.builder().host(null).port(TCP_PORT);
+ IggyTcpClientBuilder builder = IggyTcpClient.builder().host(null).port(TCP_PORT);
// When/Then: Building should throw IllegalArgumentException
assertThrows(IllegalArgumentException.class, builder::build);
@@ -204,8 +204,7 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
@Test
void shouldThrowExceptionForInvalidPort() {
// Given: Builder with invalid port
- IggyTcpClient.Builder builder =
- IggyTcpClient.builder().host("127.0.0.1").port(-1);
+ IggyTcpClientBuilder builder = IggyTcpClient.builder().host("127.0.0.1").port(-1);
// When/Then: Building should throw IllegalArgumentException
assertThrows(IllegalArgumentException.class, builder::build);
@@ -214,23 +213,23 @@ class IggyTcpClientBuilderTest extends IntegrationTest {
@Test
void shouldThrowExceptionForZeroPort() {
// Given: Builder with zero port
- IggyTcpClient.Builder builder =
- IggyTcpClient.builder().host("127.0.0.1").port(0);
+ IggyTcpClientBuilder builder = IggyTcpClient.builder().host("127.0.0.1").port(0);
// When/Then: Building should throw IllegalArgumentException
assertThrows(IllegalArgumentException.class, builder::build);
}
@Test
- void shouldMaintainBackwardCompatibilityWithOldConstructor() {
- // Given: Old constructor approach
+ void shouldWorkWithConstructorAndExplicitConnect() {
+ // Given: Constructor approach with explicit connect
IggyTcpClient client = new IggyTcpClient("127.0.0.1", TCP_PORT);
- // When: Login and perform operation
+ // When: Connect, login and perform operation
+ client.connect();
client.users().login("iggy", "iggy");
List<ClientInfo> clients = client.system().getClients();
- // Then: Should work as before
+ // Then: Should work
assertNotNull(clients);
}
}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/TcpClientFactory.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/TcpClientFactory.java
index e127c2776..897c8ef3d 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/TcpClientFactory.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/TcpClientFactory.java
@@ -28,12 +28,16 @@ final class TcpClientFactory {
private TcpClientFactory() {}
static IggyTcpClient create(GenericContainer<?> iggyServer) {
+ IggyTcpClient client;
if (iggyServer == null) {
// Server is running externally
- return new IggyTcpClient("127.0.0.1", TCP_PORT);
+ client = new IggyTcpClient("127.0.0.1", TCP_PORT);
+ } else {
+ String address = iggyServer.getHost();
+ Integer port = iggyServer.getMappedPort(TCP_PORT);
+ client = new IggyTcpClient(address, port);
}
- String address = iggyServer.getHost();
- Integer port = iggyServer.getMappedPort(TCP_PORT);
- return new IggyTcpClient(address, port);
+ client.connect();
+ return client;
}
}