attr = attributes == null ? Map.of() : Map.copyOf(attributes);
+ return CompletableFuture.supplyAsync(() -> sendWithKey(payloadCopy, attr, routingKey), client.ioExecutor());
+ }
+
+ /**
+ * Sends a message with a routing key for KEY_SHARED subscriptions.
+ *
+ * For partitioned topics, hashes the routing key to a specific partition ensuring
+ * all messages with the same key always go to the same partition. For non-partitioned
+ * topics, simply tags the routing key on the message.
+ *
+ * @param payload message body
+ * @param attributes optional metadata; pass {@code Map.of()} for none
+ * @param routingKey the routing key; must not be null
+ * @return the broker-assigned message sequence ID
+ * @throws com.danubemessaging.client.errors.DanubeClientException on unrecoverable error
+ */
+ public long sendWithKey(byte[] payload, Map attributes, String routingKey) {
+ ensureOpen();
+ TopicProducer topicProducer = selectTopicProducerForKey(routingKey);
+ return sendInternal(payload, attributes, routingKey, topicProducer);
+ }
+
+ private long sendInternal(byte[] payload, Map attributes,
+ String routingKey, TopicProducer topicProducer) {
ensureOpen();
if (lifecycleState.get() != LifecycleState.CREATED) {
create();
}
- TopicProducer topicProducer = selectTopicProducer();
int attempts = 0;
while (true) {
try {
- return topicProducer.send(payload, attributes);
+ return topicProducer.send(payload, attributes, routingKey);
} catch (RuntimeException error) {
boolean unrecoverable = client.retryManager().isUnrecoverable(error);
if (unrecoverable) {
@@ -227,6 +265,32 @@ private TopicProducer selectTopicProducer() {
return topicProducers.get(index);
}
+ private TopicProducer selectTopicProducerForKey(String routingKey) {
+ if (topicProducers.isEmpty()) {
+ throw new DanubeClientException("Producer is not initialized");
+ }
+
+ if (topicProducers.size() == 1) {
+ return topicProducers.get(0);
+ }
+
+ int index = Math.floorMod((int) fnv1aHash(routingKey), topicProducers.size());
+ return topicProducers.get(index);
+ }
+
+ /**
+ * FNV-1a 64-bit hash — must match Rust/Go/Python constants.
+ */
+ static long fnv1aHash(String key) {
+ long hash = 0xcbf29ce484222325L;
+ byte[] bytes = key.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+ for (byte b : bytes) {
+ hash ^= (b & 0xFF);
+ hash *= 0x100000001b3L;
+ }
+ return hash;
+ }
+
private boolean hasCustomRetryOptions() {
return options.maxRetries() > 0 || options.baseBackoffMs() > 0 || options.maxBackoffMs() > 0;
}
diff --git a/danube-client/src/main/java/com/danubemessaging/client/SubType.java b/danube-client/src/main/java/com/danubemessaging/client/SubType.java
index 1be710f..40445df 100644
--- a/danube-client/src/main/java/com/danubemessaging/client/SubType.java
+++ b/danube-client/src/main/java/com/danubemessaging/client/SubType.java
@@ -8,13 +8,15 @@
public enum SubType {
EXCLUSIVE,
SHARED,
- FAILOVER;
+ FAILOVER,
+ KEY_SHARED;
public DanubeApi.ConsumerRequest.SubscriptionType toProto() {
return switch (this) {
case EXCLUSIVE -> DanubeApi.ConsumerRequest.SubscriptionType.Exclusive;
case SHARED -> DanubeApi.ConsumerRequest.SubscriptionType.Shared;
case FAILOVER -> DanubeApi.ConsumerRequest.SubscriptionType.Failover;
+ case KEY_SHARED -> DanubeApi.ConsumerRequest.SubscriptionType.KeyShared;
};
}
}
diff --git a/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java b/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java
index cd05be2..6bfb144 100644
--- a/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java
+++ b/danube-client/src/main/java/com/danubemessaging/client/internal/consumer/TopicConsumer.java
@@ -106,6 +106,7 @@ private long trySubscribe() {
.setConsumerName(consumerName)
.setSubscription(options.subscription())
.setSubscriptionType(options.subType().toProto())
+ .addAllKeyFilters(options.keyFilters())
.build();
try {
diff --git a/danube-client/src/main/java/com/danubemessaging/client/internal/producer/TopicProducer.java b/danube-client/src/main/java/com/danubemessaging/client/internal/producer/TopicProducer.java
index 420769d..350036f 100644
--- a/danube-client/src/main/java/com/danubemessaging/client/internal/producer/TopicProducer.java
+++ b/danube-client/src/main/java/com/danubemessaging/client/internal/producer/TopicProducer.java
@@ -138,7 +138,7 @@ private long tryCreate() {
}
}
- public long send(byte[] payload, Map attributes) {
+ public long send(byte[] payload, Map attributes, String routingKey) {
ensureOpen();
if (lifecycleState.get() != LifecycleState.CREATED || producerId == 0) {
@@ -173,6 +173,9 @@ public long send(byte[] payload, Map attributes) {
if (cachedSchemaVersion != null) {
messageBuilder.setSchemaVersion(cachedSchemaVersion);
}
+ if (routingKey != null) {
+ messageBuilder.setRoutingKey(routingKey);
+ }
DanubeApi.StreamMessage message = messageBuilder.build();
diff --git a/danube-client/src/main/java/com/danubemessaging/client/model/StreamMessage.java b/danube-client/src/main/java/com/danubemessaging/client/model/StreamMessage.java
index 6b6a78e..1e4334b 100644
--- a/danube-client/src/main/java/com/danubemessaging/client/model/StreamMessage.java
+++ b/danube-client/src/main/java/com/danubemessaging/client/model/StreamMessage.java
@@ -15,7 +15,8 @@ public record StreamMessage(
String subscriptionName,
Map attributes,
Long schemaId,
- Integer schemaVersion) {
+ Integer schemaVersion,
+ String routingKey) {
public StreamMessage {
payload = payload == null ? new byte[0] : payload.clone();
@@ -30,6 +31,7 @@ public byte[] payload() {
public static StreamMessage fromProto(DanubeApi.StreamMessage proto) {
Long schemaId = proto.hasSchemaId() ? proto.getSchemaId() : null;
Integer schemaVersion = proto.hasSchemaVersion() ? proto.getSchemaVersion() : null;
+ String routingKey = proto.hasRoutingKey() ? proto.getRoutingKey() : null;
return new StreamMessage(
proto.getRequestId(),
@@ -40,6 +42,7 @@ public static StreamMessage fromProto(DanubeApi.StreamMessage proto) {
proto.getSubscriptionName(),
proto.getAttributesMap(),
schemaId,
- schemaVersion);
+ schemaVersion,
+ routingKey);
}
}
diff --git a/danube-client/src/test/java/com/danubemessaging/client/it/KeySharedIT.java b/danube-client/src/test/java/com/danubemessaging/client/it/KeySharedIT.java
new file mode 100644
index 0000000..c39ce24
--- /dev/null
+++ b/danube-client/src/test/java/com/danubemessaging/client/it/KeySharedIT.java
@@ -0,0 +1,305 @@
+package com.danubemessaging.client.it;
+
+import com.danubemessaging.client.Consumer;
+import com.danubemessaging.client.DanubeClient;
+import com.danubemessaging.client.DispatchStrategy;
+import com.danubemessaging.client.Producer;
+import com.danubemessaging.client.SubType;
+import com.danubemessaging.client.model.StreamMessage;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Flow;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static com.danubemessaging.client.it.TestHelpers.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Key-Shared subscription integration tests.
+ *
+ *
+ * - keySharedBasic: key routing consistency (same key → same consumer)
+ * - keySharedFiltering: deterministic filter routing with explicit filters on both consumers
+ *
+ */
+class KeySharedIT {
+
+ @Test
+ void keySharedBasic() throws Exception {
+ DanubeClient client = newClient();
+ String topic = uniqueTopic("/default/key_shared_basic_java");
+
+ Producer producer = client.newProducer()
+ .withTopic(topic)
+ .withName("producer_ks_basic")
+ .withDispatchStrategy(DispatchStrategy.RELIABLE)
+ .build();
+ producer.create();
+
+ // Two consumers sharing the same Key-Shared subscription
+ Consumer consumer1 = client.newConsumer()
+ .withTopic(topic)
+ .withConsumerName("cons_ks_1")
+ .withSubscription("ks_sub_basic")
+ .withSubscriptionType(SubType.KEY_SHARED)
+ .build();
+ consumer1.subscribe();
+
+ Consumer consumer2 = client.newConsumer()
+ .withTopic(topic)
+ .withConsumerName("cons_ks_2")
+ .withSubscription("ks_sub_basic")
+ .withSubscriptionType(SubType.KEY_SHARED)
+ .build();
+ consumer2.subscribe();
+
+ try {
+ int messageCount = 10;
+
+ // Tagged entries: (routingKey, consumerTag)
+ List entries = new CopyOnWriteArrayList<>();
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicReference error = new AtomicReference<>();
+
+ // Subscribe consumer1 (tag = "1")
+ consumer1.receive().subscribe(new Flow.Subscriber<>() {
+ private Flow.Subscription sub;
+
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ this.sub = subscription;
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(StreamMessage item) {
+ try {
+ String key = item.routingKey() != null ? item.routingKey() : "";
+ consumer1.ack(item);
+ entries.add(new String[]{key, "1"});
+ if (entries.size() >= messageCount) done.countDown();
+ } catch (Throwable t) {
+ error.set(t);
+ done.countDown();
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) { error.set(t); done.countDown(); }
+
+ @Override
+ public void onComplete() {}
+ });
+
+ // Subscribe consumer2 (tag = "2")
+ consumer2.receive().subscribe(new Flow.Subscriber<>() {
+ private Flow.Subscription sub;
+
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ this.sub = subscription;
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(StreamMessage item) {
+ try {
+ String key = item.routingKey() != null ? item.routingKey() : "";
+ consumer2.ack(item);
+ entries.add(new String[]{key, "2"});
+ if (entries.size() >= messageCount) done.countDown();
+ } catch (Throwable t) {
+ error.set(t);
+ done.countDown();
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) { error.set(t); done.countDown(); }
+
+ @Override
+ public void onComplete() {}
+ });
+
+ Thread.sleep(500);
+
+ // Send messages with two different keys
+ for (int i = 0; i < messageCount; i++) {
+ String key = (i % 2 == 0) ? "keyA" : "keyB";
+ String payload = "msg-" + i + "-key-" + key;
+ producer.sendWithKey(payload.getBytes(), Map.of(), key);
+ }
+
+ assertTrue(done.await(15, TimeUnit.SECONDS),
+ "Timeout: received " + entries.size() + "/" + messageCount);
+
+ if (error.get() != null) {
+ fail("Error in subscriber: " + error.get().getMessage());
+ }
+
+ // Verify: all messages with same key went to the same consumer
+ Map keyToConsumer = new HashMap<>();
+ for (String[] entry : entries) {
+ String key = entry[0];
+ String tag = entry[1];
+ if (keyToConsumer.containsKey(key)) {
+ assertEquals(keyToConsumer.get(key), tag,
+ "key '" + key + "' was routed to consumer " + keyToConsumer.get(key) +
+ " and " + tag + " — expected same consumer");
+ } else {
+ keyToConsumer.put(key, tag);
+ }
+ }
+
+ assertEquals(messageCount, entries.size());
+
+ } finally {
+ consumer1.close();
+ consumer2.close();
+ client.close();
+ }
+ }
+
+ @Test
+ void keySharedFiltering() throws Exception {
+ DanubeClient client = newClient();
+ String topic = uniqueTopic("/default/key_shared_filter_java");
+
+ Producer producer = client.newProducer()
+ .withTopic(topic)
+ .withName("producer_ks_filter")
+ .withDispatchStrategy(DispatchStrategy.RELIABLE)
+ .build();
+ producer.create();
+
+ // Consumer A: only receives "payment" keys
+ Consumer consumerA = client.newConsumer()
+ .withTopic(topic)
+ .withConsumerName("cons_ks_payments")
+ .withSubscription("ks_sub_filter")
+ .withSubscriptionType(SubType.KEY_SHARED)
+ .withKeyFilter("payment")
+ .build();
+ consumerA.subscribe();
+
+ // Consumer B: receives "ship*" and "invoice" keys
+ Consumer consumerB = client.newConsumer()
+ .withTopic(topic)
+ .withConsumerName("cons_ks_logistics")
+ .withSubscription("ks_sub_filter")
+ .withSubscriptionType(SubType.KEY_SHARED)
+ .withKeyFilter("ship*")
+ .withKeyFilter("invoice")
+ .build();
+ consumerB.subscribe();
+
+ try {
+ // "payment" → consumer A, "shipping"/"invoice" → consumer B
+ String[] keys = {"payment", "shipping", "payment", "invoice", "payment"};
+ int totalMessages = keys.length;
+
+ List keysOnA = new CopyOnWriteArrayList<>();
+ List keysOnB = new CopyOnWriteArrayList<>();
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicReference error = new AtomicReference<>();
+
+ // Subscribe consumer A (tag = "A")
+ consumerA.receive().subscribe(new Flow.Subscriber<>() {
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(StreamMessage item) {
+ try {
+ String key = item.routingKey() != null ? item.routingKey() : "";
+ consumerA.ack(item);
+ keysOnA.add(key);
+ if (keysOnA.size() + keysOnB.size() >= totalMessages) done.countDown();
+ } catch (Throwable t) {
+ error.set(t);
+ done.countDown();
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) { error.set(t); done.countDown(); }
+
+ @Override
+ public void onComplete() {}
+ });
+
+ // Subscribe consumer B (tag = "B")
+ consumerB.receive().subscribe(new Flow.Subscriber<>() {
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(StreamMessage item) {
+ try {
+ String key = item.routingKey() != null ? item.routingKey() : "";
+ consumerB.ack(item);
+ keysOnB.add(key);
+ if (keysOnA.size() + keysOnB.size() >= totalMessages) done.countDown();
+ } catch (Throwable t) {
+ error.set(t);
+ done.countDown();
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) { error.set(t); done.countDown(); }
+
+ @Override
+ public void onComplete() {}
+ });
+
+ Thread.sleep(500);
+
+ for (int i = 0; i < keys.length; i++) {
+ String payload = "event-" + i + "-" + keys[i];
+ producer.sendWithKey(payload.getBytes(), Map.of(), keys[i]);
+ }
+
+ assertTrue(done.await(15, TimeUnit.SECONDS),
+ "Timeout: received " + (keysOnA.size() + keysOnB.size()) + "/" + totalMessages);
+
+ if (error.get() != null) {
+ fail("Error in subscriber: " + error.get().getMessage());
+ }
+
+ // Verify: consumer A should only have "payment" keys
+ for (String k : keysOnA) {
+ assertEquals("payment", k,
+ "consumer A (filter: payment) received key '" + k + "'");
+ }
+
+ // Verify: consumer B should only have "shipping" or "invoice" keys
+ for (String k : keysOnB) {
+ assertTrue(k.equals("shipping") || k.equals("invoice"),
+ "consumer B (filter: ship*, invoice) received key '" + k + "'");
+ }
+
+ // We sent 3 "payment" + 1 "shipping" + 1 "invoice"
+ assertEquals(3, keysOnA.size(),
+ "expected 3 messages on consumer A, got " + keysOnA.size());
+ assertEquals(2, keysOnB.size(),
+ "expected 2 messages on consumer B, got " + keysOnB.size());
+
+ } finally {
+ consumerA.close();
+ consumerB.close();
+ client.close();
+ }
+ }
+}
diff --git a/examples/KeySharedConsumer.java b/examples/KeySharedConsumer.java
new file mode 100644
index 0000000..ccd3483
--- /dev/null
+++ b/examples/KeySharedConsumer.java
@@ -0,0 +1,83 @@
+import com.danubemessaging.client.Consumer;
+import com.danubemessaging.client.DanubeClient;
+import com.danubemessaging.client.SubType;
+import com.danubemessaging.client.model.StreamMessage;
+
+import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Key-Shared consumer example: subscribes with KEY_SHARED type so the broker
+ * routes messages based on their routing key — all messages with the same key
+ * go to the same consumer.
+ *
+ * Run KeySharedProducer.java in a separate terminal.
+ *
+ * Prerequisites: Danube broker running on localhost:6650
+ * cd docker && docker compose up -d
+ */
+public class KeySharedConsumer {
+
+ private static final String BROKER_URL = System.getenv().getOrDefault("DANUBE_BROKER_URL", "http://127.0.0.1:6650");
+ private static final String TOPIC = "/default/topic_key_shared";
+
+ public static void main(String[] args) throws Exception {
+ DanubeClient client = DanubeClient.builder()
+ .serviceUrl(BROKER_URL)
+ .build();
+
+ Consumer consumer = client.newConsumer()
+ .withTopic(TOPIC)
+ .withConsumerName("consumer_key_shared")
+ .withSubscription("sub_key_shared")
+ .withSubscriptionType(SubType.KEY_SHARED)
+ .build();
+
+ consumer.subscribe();
+ System.out.println("Key-Shared consumer created");
+
+ AtomicBoolean running = new AtomicBoolean(true);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ running.set(false);
+ consumer.close();
+ client.close();
+ System.out.println("Consumer closed");
+ }));
+
+ consumer.receive().subscribe(new Flow.Subscriber<>() {
+ private Flow.Subscription subscription;
+
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ this.subscription = subscription;
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(StreamMessage item) {
+ String routingKey = item.routingKey() != null ? item.routingKey() : "";
+ System.out.printf("Received: key=%s payload=%s topic=%s offset=%d%n",
+ routingKey,
+ new String(item.payload()),
+ item.messageId().topicName(),
+ item.messageId().topicOffset());
+ consumer.ack(item);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ System.err.println("Error: " + throwable.getMessage());
+ }
+
+ @Override
+ public void onComplete() {
+ System.out.println("Stream completed");
+ }
+ });
+
+ // Keep main thread alive
+ while (running.get()) {
+ Thread.sleep(1000);
+ }
+ }
+}
diff --git a/examples/KeySharedFilteredConsumer.java b/examples/KeySharedFilteredConsumer.java
new file mode 100644
index 0000000..e1dfd06
--- /dev/null
+++ b/examples/KeySharedFilteredConsumer.java
@@ -0,0 +1,83 @@
+import com.danubemessaging.client.Consumer;
+import com.danubemessaging.client.DanubeClient;
+import com.danubemessaging.client.SubType;
+import com.danubemessaging.client.model.StreamMessage;
+
+import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Key-Shared filtered consumer example: subscribes with KEY_SHARED type and
+ * key filters so this consumer only receives messages whose routing key
+ * matches "payment" or "invoice".
+ *
+ * Run KeySharedProducer.java in a separate terminal.
+ *
+ * Prerequisites: Danube broker running on localhost:6650
+ * cd docker && docker compose up -d
+ */
+public class KeySharedFilteredConsumer {
+
+ private static final String BROKER_URL = System.getenv().getOrDefault("DANUBE_BROKER_URL", "http://127.0.0.1:6650");
+ private static final String TOPIC = "/default/topic_key_shared";
+
+ public static void main(String[] args) throws Exception {
+ DanubeClient client = DanubeClient.builder()
+ .serviceUrl(BROKER_URL)
+ .build();
+
+ Consumer consumer = client.newConsumer()
+ .withTopic(TOPIC)
+ .withConsumerName("consumer_payments_only")
+ .withSubscription("sub_key_shared")
+ .withSubscriptionType(SubType.KEY_SHARED)
+ .withKeyFilter("payment")
+ .withKeyFilter("invoice")
+ .build();
+
+ consumer.subscribe();
+ System.out.println("Key-Shared filtered consumer created (filters: payment, invoice)");
+
+ AtomicBoolean running = new AtomicBoolean(true);
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ running.set(false);
+ consumer.close();
+ client.close();
+ System.out.println("Consumer closed");
+ }));
+
+ consumer.receive().subscribe(new Flow.Subscriber<>() {
+ private Flow.Subscription subscription;
+
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ this.subscription = subscription;
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(StreamMessage item) {
+ String routingKey = item.routingKey() != null ? item.routingKey() : "";
+ System.out.printf("Received: key=%s payload=%s%n",
+ routingKey,
+ new String(item.payload()));
+ consumer.ack(item);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ System.err.println("Error: " + throwable.getMessage());
+ }
+
+ @Override
+ public void onComplete() {
+ System.out.println("Stream completed");
+ }
+ });
+
+ // Keep main thread alive
+ while (running.get()) {
+ Thread.sleep(1000);
+ }
+ }
+}
diff --git a/examples/KeySharedProducer.java b/examples/KeySharedProducer.java
new file mode 100644
index 0000000..206f916
--- /dev/null
+++ b/examples/KeySharedProducer.java
@@ -0,0 +1,51 @@
+import com.danubemessaging.client.DanubeClient;
+import com.danubemessaging.client.DispatchStrategy;
+import com.danubemessaging.client.Producer;
+
+import java.util.Map;
+
+/**
+ * Key-Shared producer example: sends messages with routing keys so the broker
+ * dispatches them to consumers based on consistent hashing of the key.
+ *
+ * Run KeySharedConsumer.java or KeySharedFilteredConsumer.java in a separate terminal.
+ *
+ * Prerequisites: Danube broker running on localhost:6650
+ * cd docker && docker compose up -d
+ */
+public class KeySharedProducer {
+
+ private static final String BROKER_URL = System.getenv().getOrDefault("DANUBE_BROKER_URL", "http://127.0.0.1:6650");
+ private static final String TOPIC = "/default/topic_key_shared";
+
+ public static void main(String[] args) throws Exception {
+ DanubeClient client = DanubeClient.builder()
+ .serviceUrl(BROKER_URL)
+ .build();
+
+ Producer producer = client.newProducer()
+ .withTopic(TOPIC)
+ .withName("producer_key_shared")
+ .withDispatchStrategy(DispatchStrategy.RELIABLE)
+ .build();
+
+ producer.create();
+ System.out.println("Key-Shared producer created");
+
+ // Send messages with different routing keys.
+ // All messages with the same key are guaranteed to be delivered
+ // to the same consumer, in order.
+ String[] keys = {"payment", "shipping", "invoice", "payment", "shipping"};
+
+ for (int i = 0; i < keys.length; i++) {
+ String key = keys[i];
+ String payload = String.format("Order event #%d for key=%s", i, key);
+ long msgId = producer.sendWithKey(payload.getBytes(), Map.of(), key);
+ System.out.printf("Sent message id=%d key=%s payload=%s%n", msgId, key, payload);
+ Thread.sleep(500);
+ }
+
+ System.out.println("Done");
+ client.close();
+ }
+}
diff --git a/pom.xml b/pom.xml
index 53e4ba5..c7b03f7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
com.danube-messaging
danube-java
- 0.4.0
+ 0.5.0
pom
Danube Java
Danube Java client multi-module build