You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/06/24 12:52:04 UTC
[GitHub] [pulsar] RobertIndie commented on a diff in pull request #16202: [feature][broker] PIP 37: Support chunking with Shared subscription
RobertIndie commented on code in PR #16202:
URL: https://github.com/apache/pulsar/pull/16202#discussion_r906026279
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingSharedTest.java:
##########
@@ -0,0 +1,248 @@
+/**
+ * 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.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import io.netty.buffer.ByteBuf;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker-impl")
+public class MessageChunkingSharedTest extends ProducerConsumerBase {
+
+ private static final int MAX_MESSAGE_SIZE = 100;
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testSingleConsumer() throws Exception {
+ final String topic = "my-property/my-ns/test-single-consumer";
+ @Cleanup final Producer<String> producer = createProducer(topic);
+ @Cleanup final Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .receiverQueueSize(5)
+ .subscribe();
+
+ final List<String> values = new ArrayList<>();
+ values.add(createChunkedMessage(1)); // non-chunk
+ values.add(createChunkedMessage(10)); // number of chunks > receiver queue size
+ values.add(createChunkedMessage(4)); // number of chunks < receiver queue size
+ for (String value : values) {
+ final MessageId messageId = producer.send(value);
+ log.info("Sent {} bytes to {}", value.length(), messageId);
+ }
+
+ final List<String> receivedValues = new ArrayList<>();
+ for (int i = 0; i < values.size(); i++) {
+ final Message<String> message = consumer.receive(3, TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+ receivedValues.add(message.getValue());
+ log.info("Received {} bytes from {}", message.getValue().length(), message.getMessageId());
+ consumer.acknowledge(message);
+ }
+ assertEquals(receivedValues, values);
+ }
+
+ @Test
+ public void testMultiConsumers() throws Exception {
+ final String topic = "my-property/my-ns/test-multi-consumers";
+ @Cleanup final Producer<String> producer = createProducer(topic);
+ final ConsumerBuilder<String> consumerBuilder = pulsarClient.newConsumer(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscriptionType(SubscriptionType.Shared)
+ .receiverQueueSize(5);
+
+ final List<String> receivedValues1 = Collections.synchronizedList(new ArrayList<>());
+ @Cleanup final Consumer<String> consumer1 = consumerBuilder
+ .messageListener((MessageListener<String>) (consumer, msg) -> receivedValues1.add(msg.getValue()))
+ .subscribe();
+ final List<String> receivedValues2 = Collections.synchronizedList(new ArrayList<>());
+ @Cleanup final Consumer<String> consumer2 = consumerBuilder
+ .messageListener((MessageListener<String>) (consumer, msg) -> receivedValues2.add(msg.getValue()))
+ .subscribe();
+
+ final Set<String> values = new HashSet<>();
+ for (int i = 0; i < 10; i++) {
+ values.add(createChunkedMessage(4));
+ }
+ for (String value : values) {
+ producer.send(value);
+ }
+
+ Awaitility.await().atMost(Duration.ofSeconds(3))
+ .until(() -> receivedValues1.size() + receivedValues2.size() >= values.size());
Review Comment:
Maybe check if they are equal is enough here?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java:
##########
@@ -607,19 +645,72 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
dispatchRateLimiter.get().tryDispatchPermit(permits, totalBytesSent);
}
}
+ }
+
+ private void sendChunkedMessagesToConsumers(ReadType readType,
+ List<Entry> entries,
+ MessageMetadata[] metadataArray) {
+ final List<EntryAndMetadata> originalEntryAndMetadataList = new ArrayList<>(metadataArray.length);
+ for (int i = 0; i < metadataArray.length; i++) {
+ originalEntryAndMetadataList.add(EntryAndMetadata.create(entries.get(i), metadataArray[i]));
+ }
- if (entriesToDispatch > 0) {
+ final Map<Consumer, List<EntryAndMetadata>> assignResult =
+ assignor.assign(originalEntryAndMetadataList, consumerList.size());
+ long totalMessagesSent = 0;
+ long totalBytesSent = 0;
+ long totalEntries = 0;
+ final AtomicInteger numConsumers = new AtomicInteger(assignResult.size());
+ for (Map.Entry<Consumer, List<EntryAndMetadata>> current : assignResult.entrySet()) {
+ final Consumer consumer = current.getKey();
+ final List<EntryAndMetadata> entryAndMetadataList = current.getValue();
+ final int messagesForC = Math.min(consumer.getAvailablePermits(), entryAndMetadataList.size());
if (log.isDebugEnabled()) {
- log.debug("[{}] No consumers found with available permits, storing {} positions for later replay", name,
- entries.size() - start);
+ log.debug("[{}] select consumer {} with messages num {}, read type is {}",
+ name, consumer.consumerName(), messagesForC, readType);
}
- entries.subList(start, entries.size()).forEach(entry -> {
- long stickyKeyHash = getStickyKeyHash(entry);
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
- entry.release();
+ if (messagesForC < entryAndMetadataList.size()) {
Review Comment:
When will messagesForC or `consumer.getAvailablePermits()` be less than `entryAndMetadataList.size()`?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryAndMetadata.java:
##########
@@ -0,0 +1,117 @@
+/**
+ * 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.pulsar.broker.service;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.buffer.ByteBuf;
+import java.nio.charset.StandardCharsets;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
+import org.apache.pulsar.common.protocol.Commands;
+
+public class EntryAndMetadata implements Entry {
+
+ private final Entry entry;
+ @Getter
+ @Nullable
+ private final MessageMetadata metadata;
+
+ private EntryAndMetadata(final Entry entry, @Nullable final MessageMetadata metadata) {
+ this.entry = entry;
+ this.metadata = metadata;
+ }
+
+ public static EntryAndMetadata create(final Entry entry, final MessageMetadata metadata) {
+ return new EntryAndMetadata(entry, metadata);
+ }
+
+ @VisibleForTesting
+ static EntryAndMetadata create(final Entry entry) {
+ return create(entry, Commands.peekAndCopyMessageMetadata(entry.getDataBuffer(), "", -1));
+ }
+
+ public byte[] getStickyKey() {
+ if (metadata != null) {
+ if (metadata.hasOrderingKey()) {
+ return metadata.getOrderingKey();
+ } else if (metadata.hasPartitionKey()) {
+ return metadata.getPartitionKey().getBytes(StandardCharsets.UTF_8);
+ } else {
+ return "NONE_KEY".getBytes(StandardCharsets.UTF_8);
+ }
+ } else {
+ return "NONE_KEY".getBytes(StandardCharsets.UTF_8);
+ }
+ }
Review Comment:
```suggestion
}
}
return "NONE_KEY".getBytes(StandardCharsets.UTF_8);
}
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java:
##########
@@ -129,6 +133,8 @@ public PersistentDispatcherMultipleConsumers(PersistentTopic topic, ManagedCurso
: RedeliveryTrackerDisabled.REDELIVERY_TRACKER_DISABLED;
this.readBatchSize = serviceConfig.getDispatcherMaxReadBatchSize();
this.initializeDispatchRateLimiterIfNeeded();
+ this.assignor = new SharedConsumerAssignor(this::getNextConsumer,
+ entry -> addMessageToReplay(entry.getLedgerId(), entry.getEntryId()));
Review Comment:
Should we release this entry after addMessageToReplay?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org