@@ -144,18 +144,56 @@ public CompletableFuture<Long> sendAsync(byte[] payload, Map<String, String> att
144144 * @throws com.danubemessaging.client.errors.DanubeClientException on unrecoverable error
145145 */
146146 public long send (byte [] payload , Map <String , String > attributes ) {
147+ return sendInternal (payload , attributes , null , selectTopicProducer ());
148+ }
149+
150+ /**
151+ * Sends a message with a routing key asynchronously for KEY_SHARED subscriptions.
152+ *
153+ * @param payload message body
154+ * @param attributes optional metadata
155+ * @param routingKey the routing key; all messages with the same key go to the same consumer
156+ * @return a future resolving to the broker-assigned message sequence ID
157+ */
158+ public CompletableFuture <Long > sendWithKeyAsync (byte [] payload , Map <String , String > attributes ,
159+ String routingKey ) {
160+ byte [] payloadCopy = payload == null ? new byte [0 ] : payload .clone ();
161+ Map <String , String > attr = attributes == null ? Map .of () : Map .copyOf (attributes );
162+ return CompletableFuture .supplyAsync (() -> sendWithKey (payloadCopy , attr , routingKey ), client .ioExecutor ());
163+ }
164+
165+ /**
166+ * Sends a message with a routing key for KEY_SHARED subscriptions.
167+ *
168+ * <p>For partitioned topics, hashes the routing key to a specific partition ensuring
169+ * all messages with the same key always go to the same partition. For non-partitioned
170+ * topics, simply tags the routing key on the message.
171+ *
172+ * @param payload message body
173+ * @param attributes optional metadata; pass {@code Map.of()} for none
174+ * @param routingKey the routing key; must not be null
175+ * @return the broker-assigned message sequence ID
176+ * @throws com.danubemessaging.client.errors.DanubeClientException on unrecoverable error
177+ */
178+ public long sendWithKey (byte [] payload , Map <String , String > attributes , String routingKey ) {
179+ ensureOpen ();
180+ TopicProducer topicProducer = selectTopicProducerForKey (routingKey );
181+ return sendInternal (payload , attributes , routingKey , topicProducer );
182+ }
183+
184+ private long sendInternal (byte [] payload , Map <String , String > attributes ,
185+ String routingKey , TopicProducer topicProducer ) {
147186 ensureOpen ();
148187
149188 if (lifecycleState .get () != LifecycleState .CREATED ) {
150189 create ();
151190 }
152191
153- TopicProducer topicProducer = selectTopicProducer ();
154192 int attempts = 0 ;
155193
156194 while (true ) {
157195 try {
158- return topicProducer .send (payload , attributes );
196+ return topicProducer .send (payload , attributes , routingKey );
159197 } catch (RuntimeException error ) {
160198 boolean unrecoverable = client .retryManager ().isUnrecoverable (error );
161199 if (unrecoverable ) {
@@ -227,6 +265,32 @@ private TopicProducer selectTopicProducer() {
227265 return topicProducers .get (index );
228266 }
229267
268+ private TopicProducer selectTopicProducerForKey (String routingKey ) {
269+ if (topicProducers .isEmpty ()) {
270+ throw new DanubeClientException ("Producer is not initialized" );
271+ }
272+
273+ if (topicProducers .size () == 1 ) {
274+ return topicProducers .get (0 );
275+ }
276+
277+ int index = Math .floorMod ((int ) fnv1aHash (routingKey ), topicProducers .size ());
278+ return topicProducers .get (index );
279+ }
280+
281+ /**
282+ * FNV-1a 64-bit hash — must match Rust/Go/Python constants.
283+ */
284+ static long fnv1aHash (String key ) {
285+ long hash = 0xcbf29ce484222325L ;
286+ byte [] bytes = key .getBytes (java .nio .charset .StandardCharsets .UTF_8 );
287+ for (byte b : bytes ) {
288+ hash ^= (b & 0xFF );
289+ hash *= 0x100000001b3L ;
290+ }
291+ return hash ;
292+ }
293+
230294 private boolean hasCustomRetryOptions () {
231295 return options .maxRetries () > 0 || options .baseBackoffMs () > 0 || options .maxBackoffMs () > 0 ;
232296 }
0 commit comments