diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java index 4cd54420200be..7c03ea66b1aeb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java @@ -286,16 +286,29 @@ public Future sendMessages(final List entries, EntryBatch totalChunkedMessages, redeliveryTracker, DEFAULT_CONSUMER_EPOCH); } + public Future sendMessages(final List entries, EntryBatchSizes batchSizes, + EntryBatchIndexesAcks batchIndexesAcks, + int totalMessages, long totalBytes, long totalChunkedMessages, + RedeliveryTracker redeliveryTracker, long epoch) { + return sendMessages(entries, null, batchSizes, batchIndexesAcks, totalMessages, totalBytes, + totalChunkedMessages, redeliveryTracker, epoch); + } + /** * Dispatch a list of entries to the consumer.
* It is also responsible to release entries data and recycle entries object. * * @return a SendMessageInfo object that contains the detail of what was sent to consumer */ - public Future sendMessages(final List entries, EntryBatchSizes batchSizes, + public Future sendMessages(final List entries, + final List stickyKeyHashes, + EntryBatchSizes batchSizes, EntryBatchIndexesAcks batchIndexesAcks, - int totalMessages, long totalBytes, long totalChunkedMessages, - RedeliveryTracker redeliveryTracker, long epoch) { + int totalMessages, + long totalBytes, + long totalChunkedMessages, + RedeliveryTracker redeliveryTracker, + long epoch) { this.lastConsumedTimestamp = System.currentTimeMillis(); if (entries.isEmpty() || totalMessages == 0) { @@ -323,7 +336,7 @@ public Future sendMessages(final List entries, EntryBatch // because this consumer is possible to disconnect at this time. if (pendingAcks != null) { int batchSize = batchSizes.getBatchSize(i); - int stickyKeyHash = getStickyKeyHash(entry); + int stickyKeyHash = stickyKeyHashes == null ? getStickyKeyHash(entry) : stickyKeyHashes.get(i); long[] ackSet = batchIndexesAcks == null ? null : batchIndexesAcks.getAckSet(i); if (ackSet != null) { unackedMessages -= (batchSize - BitSet.valueOf(ackSet).cardinality()); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java index 2cad253f96ee2..fb7bd22de94a7 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumers.java @@ -126,6 +126,14 @@ protected Map> initialValue() throws Exception { } }; + private static final FastThreadLocal>> localGroupedStickyKeyHashes = + new FastThreadLocal>>() { + @Override + protected Map> initialValue() throws Exception { + return new HashMap<>(); + } + }; + @Override public void sendMessages(List entries) { if (entries.isEmpty()) { @@ -139,28 +147,38 @@ public void sendMessages(List entries) { final Map> groupedEntries = localGroupedEntries.get(); groupedEntries.clear(); + final Map> consumerStickyKeyHashesMap = localGroupedStickyKeyHashes.get(); + consumerStickyKeyHashesMap.clear(); for (Entry entry : entries) { - Consumer consumer = selector.select(peekStickyKey(entry.getDataBuffer())); + byte[] stickyKey = peekStickyKey(entry.getDataBuffer()); + int stickyKeyHash = StickyKeyConsumerSelector.makeStickyKeyHash(stickyKey); + + Consumer consumer = selector.select(stickyKeyHash); if (consumer != null) { - groupedEntries.computeIfAbsent(consumer, k -> new ArrayList<>()).add(entry); + int startingSize = Math.max(10, entries.size() / (2 * consumerSet.size())); + groupedEntries.computeIfAbsent(consumer, k -> new ArrayList<>(startingSize)).add(entry); + consumerStickyKeyHashesMap + .computeIfAbsent(consumer, k -> new ArrayList<>(startingSize)).add(stickyKeyHash); } else { entry.release(); } } for (Map.Entry> entriesByConsumer : groupedEntries.entrySet()) { - Consumer consumer = entriesByConsumer.getKey(); - List entriesForConsumer = entriesByConsumer.getValue(); + final Consumer consumer = entriesByConsumer.getKey(); + final List entriesForConsumer = entriesByConsumer.getValue(); + final List stickyKeysForConsumer = consumerStickyKeyHashesMap.get(consumer); SendMessageInfo sendMessageInfo = SendMessageInfo.getThreadLocal(); EntryBatchSizes batchSizes = EntryBatchSizes.get(entriesForConsumer.size()); filterEntriesForConsumer(entriesForConsumer, batchSizes, sendMessageInfo, null, null, false, consumer); if (consumer.getAvailablePermits() > 0 && consumer.isWritable()) { - consumer.sendMessages(entriesForConsumer, batchSizes, null, sendMessageInfo.getTotalMessages(), + consumer.sendMessages(entriesForConsumer, stickyKeysForConsumer, batchSizes, + null, sendMessageInfo.getTotalMessages(), sendMessageInfo.getTotalBytes(), sendMessageInfo.getTotalChunkedMessages(), - getRedeliveryTracker()); + getRedeliveryTracker(), Commands.DEFAULT_CONSUMER_EPOCH); TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -sendMessageInfo.getTotalMessages()); } else { entriesForConsumer.forEach(e -> { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumersTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumersTest.java index b2638d53ab1c3..6b0f48a57cfe3 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumersTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentStickyKeyDispatcherMultipleConsumersTest.java @@ -128,15 +128,15 @@ public void testSendMessage() throws BrokerServiceException { assertEquals(byteBuf.toString(UTF_8), "message" + index); }; return mockPromise; - }).when(consumerMock).sendMessages(any(List.class), any(EntryBatchSizes.class), any(), - anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class)); + }).when(consumerMock).sendMessages(any(List.class), any(List.class), any(EntryBatchSizes.class), any(), + anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class), anyLong()); try { nonpersistentDispatcher.sendMessages(entries); } catch (Exception e) { fail("Failed to sendMessages.", e); } - verify(consumerMock, times(1)).sendMessages(any(List.class), any(EntryBatchSizes.class), - eq(null), anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class)); + verify(consumerMock, times(1)).sendMessages(any(List.class), any(List.class), any(EntryBatchSizes.class), + eq(null), anyInt(), anyLong(), anyLong(), any(RedeliveryTracker.class), anyLong()); } @Test(timeOut = 10000)