You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by ma...@apache.org on 2021/10/15 05:13:16 UTC
[pulsar] branch master updated: [PIP 79][client] Add lazy-loading
feature to PartitionedProducer (#10279)
This is an automated email from the ASF dual-hosted git repository.
massakam pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 1f8945a [PIP 79][client] Add lazy-loading feature to PartitionedProducer (#10279)
1f8945a is described below
commit 1f8945addb18bf1f94366d2c7f5286ee6f62c675
Author: Yuri Mizushima <yu...@yahoo-corp.jp>
AuthorDate: Fri Oct 15 14:12:21 2021 +0900
[PIP 79][client] Add lazy-loading feature to PartitionedProducer (#10279)
* feat: add lazy loading feature to PartitionedProducerImpl
* feat: add partial round robin routing mode
* test: add tests for lazy-loading
* fix: fix producer closing code at lazy-loading
* test: remove unnecessary handling, fail from test codes
* feat: add enableLazyStartPartitionedProducers config
* test: fix test for lazy-loading config
* fix: address comments
* fix: add partition-change interceptor
* fix: address comments
---
.../apache/pulsar/client/api/ClientErrorsTest.java | 97 +++++++-
.../impl/PartialPartitionedProducerTest.java | 255 +++++++++++++++++++++
.../apache/pulsar/client/api/ProducerBuilder.java | 19 ++
.../client/impl/PartitionedProducerImpl.java | 182 +++++++++------
.../pulsar/client/impl/ProducerBuilderImpl.java | 6 +
.../impl/conf/ProducerConfigurationData.java | 2 +
.../PartialRoundRobinMessageRouterImpl.java | 85 +++++++
.../client/impl/PartitionedProducerImplTest.java | 49 ++++
8 files changed, 628 insertions(+), 67 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
index a38b711..f66b198 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/ClientErrorsTest.java
@@ -33,6 +33,7 @@ import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import org.apache.pulsar.client.impl.ConsumerBase;
+import org.apache.pulsar.client.impl.PartitionedProducerImpl;
import org.apache.pulsar.client.impl.ProducerBase;
import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse.LookupType;
import org.apache.pulsar.common.api.proto.ServerError;
@@ -462,8 +463,100 @@ public class ClientErrorsTest {
mockBrokerService.resetHandleSubscribe();
}
- // if a producer fails to connect while creating partitioned producer, it should close all successful connections of
- // other producers and fail
+ // failed to connect to partition at initialization step if a producer which connects to broker as lazy-loading mode
+ @Test
+ public void testPartitionedProducerFailOnInitialization() throws Throwable {
+ @Cleanup
+ PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getHttpAddress()).build();
+ final AtomicInteger producerCounter = new AtomicInteger(0);
+
+ mockBrokerService.setHandleProducer((ctx, producer) -> {
+ if (producerCounter.incrementAndGet() == 1) {
+ ctx.writeAndFlush(Commands.newError(producer.getRequestId(), ServerError.AuthorizationError, "msg"));
+ return;
+ }
+ ctx.writeAndFlush(Commands.newProducerSuccess(producer.getRequestId(), "default-producer", SchemaVersion.Empty));
+ });
+
+ try {
+ client.newProducer()
+ .enableLazyStartPartitionedProducers(true)
+ .accessMode(ProducerAccessMode.Shared)
+ .topic("persistent://prop/use/ns/multi-part-t1").create();
+ fail("Should have failed with an authorization error");
+ } catch (Exception e) {
+ assertTrue(e instanceof PulsarClientException.AuthorizationException);
+ }
+ assertEquals(producerCounter.get(), 1);
+
+ mockBrokerService.resetHandleProducer();
+ mockBrokerService.resetHandleCloseProducer();
+ client.close();
+ }
+
+ // failed to connect to partition at sending step if a producer which connects to broker as lazy-loading mode
+ @Test
+ public void testPartitionedProducerFailOnSending() throws Throwable {
+ @Cleanup
+ PulsarClient client = PulsarClient.builder().serviceUrl(mockBrokerService.getHttpAddress()).build();
+ final AtomicInteger producerCounter = new AtomicInteger(0);
+ final AtomicInteger closeCounter = new AtomicInteger(0);
+ final String topicName = "persistent://prop/use/ns/multi-part-t1";
+
+ mockBrokerService.setHandleProducer((ctx, producer) -> {
+ if (producerCounter.incrementAndGet() == 2) {
+ ctx.writeAndFlush(Commands.newError(producer.getRequestId(), ServerError.AuthorizationError, "msg"));
+ return;
+ }
+ ctx.writeAndFlush(Commands.newProducerSuccess(producer.getRequestId(), "default-producer", SchemaVersion.Empty));
+ });
+
+ mockBrokerService.setHandleSend((ctx, send, headersAndPayload) ->
+ ctx.writeAndFlush(Commands.newSendReceipt(send.getProducerId(), send.getSequenceId(), send.getHighestSequenceId(), 0L, 0L))
+ );
+
+ mockBrokerService.setHandleCloseProducer((ctx, closeProducer) -> {
+ ctx.writeAndFlush(Commands.newSuccess(closeProducer.getRequestId()));
+ closeCounter.incrementAndGet();
+ });
+
+ final PartitionedProducerImpl<byte[]> producer = (PartitionedProducerImpl<byte[]>) client.newProducer()
+ .enableLazyStartPartitionedProducers(true)
+ .accessMode(ProducerAccessMode.Shared)
+ .topic(topicName)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .create();
+
+ try {
+ producer.send("msg".getBytes());
+ fail("Should have failed with an not connected exception");
+ } catch (Exception e) {
+ assertTrue(e instanceof PulsarClientException.NotConnectedException);
+ assertEquals(producer.getProducers().size(), 1);
+ }
+
+ try {
+ // recreate failed producer
+ for (int i = 0; i < client.getPartitionsForTopic(topicName).get().size(); i++) {
+ producer.send("msg".getBytes());
+ }
+ assertEquals(producer.getProducers().size(), client.getPartitionsForTopic(topicName).get().size());
+ assertEquals(producerCounter.get(), 5);
+ } catch (Exception e) {
+ fail();
+ }
+
+ // should not call close
+ assertEquals(closeCounter.get(), 0);
+
+ mockBrokerService.resetHandleProducer();
+ mockBrokerService.resetHandleCloseProducer();
+ client.close();
+ }
+
+ // if a producer which doesn't connect as lazy-loading mode fails to connect while creating partitioned producer,
+ // it should close all successful connections of other producers and fail
@Test
public void testOneProducerFailShouldCloseAllProducersInPartitionedProducer() throws Exception {
@Cleanup
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PartialPartitionedProducerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PartialPartitionedProducerTest.java
new file mode 100644
index 0000000..7fd992a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PartialPartitionedProducerTest.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.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+import lombok.Cleanup;
+import lombok.extern.slf4j.Slf4j;
+import java.lang.reflect.Field;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.ProducerAccessMode;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.client.impl.customroute.PartialRoundRobinMessageRouterImpl;
+import 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 PartialPartitionedProducerTest extends ProducerConsumerBase {
+ @Override
+ @BeforeClass
+ public void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @Override
+ @AfterClass(alwaysRun = true)
+ public void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testPtWithSinglePartition() throws Throwable {
+ final String topic = BrokerTestUtil.newUniqueName("pt-with-single-routing");
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImpl = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.SinglePartition)
+ .create();
+
+ for (int i = 0; i < 10; i++) {
+ producerImpl.newMessage().value("msg".getBytes()).send();
+ }
+ assertEquals(producerImpl.getProducers().size(), 1);
+ }
+
+ @Test
+ public void testPtWithPartialPartition() throws Throwable {
+ final String topic = BrokerTestUtil.newUniqueName("pt-with-partial-routing");
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImpl = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.CustomPartition)
+ .messageRouter(new PartialRoundRobinMessageRouterImpl(3))
+ .create();
+
+ for (int i = 0; i < 10; i++) {
+ producerImpl.newMessage().value("msg".getBytes()).send();
+ }
+ assertEquals(producerImpl.getProducers().size(), 3);
+ }
+
+ // AddPartitionTest
+ @Test
+ public void testPtLazyLoading() throws Throwable {
+ final String topic = BrokerTestUtil.newUniqueName("pt-lazily");
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImpl = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .create();
+
+ final Supplier<Boolean> send = () -> {
+ for (int i = 0; i < 10; i++) {
+ try {
+ producerImpl.newMessage().value("msg".getBytes()).send();
+ } catch (Throwable e) {
+ return false;
+ }
+ }
+ return true;
+ };
+
+ // create first producer at initialization step
+ assertEquals(producerImpl.getProducers().size(), 1);
+
+ assertTrue(send.get());
+ assertEquals(producerImpl.getProducers().size(), 10);
+ }
+
+ @Test
+ public void testPtLoadingNotSharedMode() throws Throwable {
+ final String topic = BrokerTestUtil.newUniqueName("pt-not-shared-mode");
+ admin.topics().createPartitionedTopic(topic, 10);
+
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImplExclusive = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .accessMode(ProducerAccessMode.Exclusive)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .create();
+
+ // create first producer at initialization step
+ assertEquals(producerImplExclusive.getProducers().size(), 10);
+
+ producerImplExclusive.close();
+
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImplWaitForExclusive = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .accessMode(ProducerAccessMode.WaitForExclusive)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .create();
+
+ assertEquals(producerImplWaitForExclusive.getProducers().size(), 10);
+ }
+
+ // AddPartitionAndLimitTest
+ @Test
+ public void testPtUpdateWithPartialPartition() throws Throwable {
+ final String topic = BrokerTestUtil.newUniqueName("pt-update-with-partial-routing");
+ admin.topics().createPartitionedTopic(topic, 2);
+
+ final Field field = PartitionedProducerImpl.class.getDeclaredField("topicMetadata");
+ field.setAccessible(true);
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImpl = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.CustomPartition)
+ .messageRouter(new PartialRoundRobinMessageRouterImpl(3))
+ .accessMode(ProducerAccessMode.Shared)
+ .autoUpdatePartitions(true)
+ .autoUpdatePartitionsInterval(1, TimeUnit.SECONDS)
+ .create();
+
+ final Supplier<Boolean> send = ()-> {
+ for (int i = 0; i < 10; i++) {
+ try {
+ producerImpl.newMessage().value("msg".getBytes()).send();
+ } catch (Throwable e) {
+ return false;
+ }
+ }
+ return true;
+ };
+
+ // create first producer at initialization step
+ assertEquals(producerImpl.getProducers().size(), 1);
+
+ assertTrue(send.get());
+ assertEquals(producerImpl.getProducers().size(), 2);
+
+ admin.topics().updatePartitionedTopic(topic, 3);
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(((TopicMetadata) field.get(producerImpl)).numPartitions(), 3));
+ assertEquals(producerImpl.getProducers().size(), 2);
+
+ assertTrue(send.get());
+ assertEquals(producerImpl.getProducers().size(), 3);
+
+ admin.topics().updatePartitionedTopic(topic, 4);
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(((TopicMetadata) field.get(producerImpl)).numPartitions(), 4));
+ assertTrue(send.get());
+ assertEquals(producerImpl.getProducers().size(), 3);
+ }
+
+ @Test
+ public void testPtUpdateNotSharedMode() throws Throwable {
+ final String topic = BrokerTestUtil.newUniqueName("pt-update-not-shared");
+ admin.topics().createPartitionedTopic(topic, 2);
+
+ final Field field = PartitionedProducerImpl.class.getDeclaredField("topicMetadata");
+ field.setAccessible(true);
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImplExclusive = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .accessMode(ProducerAccessMode.Exclusive)
+ .autoUpdatePartitions(true)
+ .autoUpdatePartitionsInterval(1, TimeUnit.SECONDS)
+ .create();
+
+ assertEquals(producerImplExclusive.getProducers().size(), 2);
+
+ admin.topics().updatePartitionedTopic(topic, 3);
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(((TopicMetadata) field.get(producerImplExclusive)).numPartitions(), 3));
+ assertEquals(producerImplExclusive.getProducers().size(), 3);
+
+ producerImplExclusive.close();
+
+ @Cleanup
+ final PartitionedProducerImpl<byte[]> producerImplWaitForExclusive = (PartitionedProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic(topic)
+ .enableLazyStartPartitionedProducers(true)
+ .enableBatching(false)
+ .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
+ .accessMode(ProducerAccessMode.WaitForExclusive)
+ .autoUpdatePartitions(true)
+ .autoUpdatePartitionsInterval(1, TimeUnit.SECONDS)
+ .create();
+
+ assertEquals(producerImplWaitForExclusive.getProducers().size(), 3);
+
+ admin.topics().updatePartitionedTopic(topic, 4);
+ Awaitility.await().untilAsserted(() ->
+ assertEquals(((TopicMetadata) field.get(producerImplWaitForExclusive)).numPartitions(), 4));
+ assertEquals(producerImplWaitForExclusive.getProducers().size(), 4);
+ }
+}
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
index ecac0b6..fd91101 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
@@ -576,4 +576,23 @@ public interface ProducerBuilder<T> extends Cloneable {
* @since 2.5.0
*/
ProducerBuilder<T> enableMultiSchema(boolean multiSchema);
+
+ /**
+ * This config affects Shared mode producers of partitioned topics only. It controls whether
+ * producers register and connect immediately to the owner broker of each partition
+ * or start lazily on demand. The internal producer of one partition is always
+ * started eagerly, chosen by the routing policy, but the internal producers of
+ * any additional partitions are started on demand, upon receiving their first
+ * message.
+ * Using this mode can reduce the strain on brokers for topics with large numbers of
+ * partitions and when the SinglePartition or some custom partial partition routing policy
+ * like PartialRoundRobinMessageRouterImpl is used without keyed messages.
+ * Because producer connection can be on demand, this can produce extra send latency
+ * for the first messages of a given partition.
+ *
+ * @param lazyStartPartitionedProducers
+ * true/false as to whether to start partition producers lazily
+ * @return the producer builder instance
+ */
+ ProducerBuilder<T> enableLazyStartPartitionedProducers(boolean lazyStartPartitionedProducers);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
index 5702563..4525531 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
@@ -22,10 +22,10 @@ import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import java.util.Collection;
+import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
@@ -35,11 +35,13 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerAccessMode;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.NotSupportedException;
import org.apache.pulsar.client.api.Schema;
@@ -48,6 +50,7 @@ import org.apache.pulsar.client.api.transaction.Transaction;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,10 +58,11 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
private static final Logger log = LoggerFactory.getLogger(PartitionedProducerImpl.class);
- private List<ProducerImpl<T>> producers;
- private MessageRouter routerPolicy;
+ private final ConcurrentOpenHashMap<Integer, ProducerImpl<T>> producers;
+ private final MessageRouter routerPolicy;
private final ProducerStatsRecorderImpl stats;
private TopicMetadata topicMetadata;
+ private final int firstPartitionIndex;
// timeout related to auto check and subscribe partition increasement
private volatile Timeout partitionsAutoUpdateTimeout = null;
@@ -68,15 +72,29 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
public PartitionedProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf, int numPartitions,
CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T> schema, ProducerInterceptors interceptors) {
super(client, topic, conf, producerCreatedFuture, schema, interceptors);
- this.producers = Lists.newArrayListWithCapacity(numPartitions);
+ this.producers = new ConcurrentOpenHashMap<>();
this.topicMetadata = new TopicMetadataImpl(numPartitions);
this.routerPolicy = getMessageRouter();
stats = client.getConfiguration().getStatsIntervalSeconds() > 0 ? new ProducerStatsRecorderImpl() : null;
+ // MaxPendingMessagesAcrossPartitions doesn't support partial partition such as SinglePartition correctly
int maxPendingMessages = Math.min(conf.getMaxPendingMessages(),
conf.getMaxPendingMessagesAcrossPartitions() / numPartitions);
conf.setMaxPendingMessages(maxPendingMessages);
- start();
+
+ final List<Integer> indexList;
+ if (conf.isLazyStartPartitionedProducers() &&
+ conf.getAccessMode() == ProducerAccessMode.Shared) {
+ // try to create producer at least one partition
+ indexList = Collections.singletonList(routerPolicy
+ .choosePartition(((TypedMessageBuilderImpl<T>) newMessage()).getMessage(), topicMetadata));
+ } else {
+ // try to create producer for all partitions
+ indexList = IntStream.range(0, topicMetadata.numPartitions()).boxed().collect(Collectors.toList());
+ }
+
+ firstPartitionIndex = indexList.get(0);
+ start(indexList);
// start track and auto subscribe partition increasement
if (conf.isAutoUpdatePartitions()) {
@@ -113,25 +131,22 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
@Override
public String getProducerName() {
- return producers.get(0).getProducerName();
+ return producers.get(firstPartitionIndex).getProducerName();
}
@Override
public long getLastSequenceId() {
// Return the highest sequence id across all partitions. This will be correct,
// since there is a single id generator across all partitions for the same producer
- return producers.stream().map(Producer::getLastSequenceId).mapToLong(Long::longValue).max().orElse(-1);
+ return producers.values().stream().map(Producer::getLastSequenceId).mapToLong(Long::longValue).max().orElse(-1);
}
- private void start() {
+ private void start(List<Integer> indexList) {
AtomicReference<Throwable> createFail = new AtomicReference<Throwable>();
AtomicInteger completed = new AtomicInteger();
- for (int partitionIndex = 0; partitionIndex < topicMetadata.numPartitions(); partitionIndex++) {
- String partitionName = TopicName.get(topic).getPartition(partitionIndex).toString();
- ProducerImpl<T> producer = client.newProducerImpl(partitionName, partitionIndex,
- conf, schema, interceptors, new CompletableFuture<>());
- producers.add(producer);
- producer.producerCreatedFuture().handle((prod, createException) -> {
+
+ for (int partitionIndex : indexList) {
+ createProducer(partitionIndex).producerCreatedFuture().handle((prod, createException) -> {
if (createException != null) {
setState(State.Failed);
createFail.compareAndSet(null, createException);
@@ -141,7 +156,7 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
// due to any
// failure in one of the partitions and close the successfully
// created partitions
- if (completed.incrementAndGet() == topicMetadata.numPartitions()) {
+ if (completed.incrementAndGet() == indexList.size()) {
if (createFail.get() == null) {
setState(State.Ready);
log.info("[{}] Created partitioned producer", topic);
@@ -159,7 +174,14 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
return null;
});
}
+ }
+ private ProducerImpl<T> createProducer(final int partitionIndex) {
+ return producers.computeIfAbsent(partitionIndex, (idx) -> {
+ String partitionName = TopicName.get(topic).getPartition(idx).toString();
+ return client.newProducerImpl(partitionName, idx,
+ conf, schema, interceptors, new CompletableFuture<>());
+ });
}
@Override
@@ -169,6 +191,34 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
@Override
CompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message, Transaction txn) {
+ int partition = routerPolicy.choosePartition(message, topicMetadata);
+ checkArgument(partition >= 0 && partition < topicMetadata.numPartitions(),
+ "Illegal partition index chosen by the message routing policy: " + partition);
+
+ if (conf.isLazyStartPartitionedProducers() && !producers.containsKey(partition)) {
+ final ProducerImpl<T> newProducer = createProducer(partition);
+ final State createState = newProducer.producerCreatedFuture().handle((prod, createException) -> {
+ if (createException != null) {
+ log.error("[{}] Could not create internal producer. partitionIndex: {}", topic, partition,
+ createException);
+ try {
+ producers.remove(partition, newProducer);
+ newProducer.close();
+ } catch (PulsarClientException e) {
+ log.error("[{}] Could not close internal producer. partitionIndex: {}", topic, partition, e);
+ }
+ return State.Failed;
+ }
+ if (log.isDebugEnabled()) {
+ log.debug("[{}] Created internal producer. partitionIndex: {}", topic, partition);
+ }
+ return State.Ready;
+ }).join();
+ if (createState == State.Failed) {
+ return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException());
+ }
+ }
+
switch (getState()) {
case Ready:
case Connecting:
@@ -185,34 +235,30 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
return FutureUtil.failedFuture(new PulsarClientException.NotConnectedException());
}
- int partition = routerPolicy.choosePartition(message, topicMetadata);
- checkArgument(partition >= 0 && partition < topicMetadata.numPartitions(),
- "Illegal partition index chosen by the message routing policy: " + partition);
return producers.get(partition).internalSendWithTxnAsync(message, txn);
}
@Override
public CompletableFuture<Void> flushAsync() {
- List<CompletableFuture<Void>> flushFutures =
- producers.stream().map(ProducerImpl::flushAsync).collect(Collectors.toList());
- return CompletableFuture.allOf(flushFutures.toArray(new CompletableFuture[flushFutures.size()]));
+ return CompletableFuture.allOf(
+ producers.values().stream().map(ProducerImpl::flushAsync).toArray(CompletableFuture[]::new));
}
@Override
void triggerFlush() {
- producers.forEach(ProducerImpl::triggerFlush);
+ producers.values().forEach(ProducerImpl::triggerFlush);
}
@Override
public boolean isConnected() {
// returns false if any of the partition is not connected
- return producers.stream().allMatch(ProducerImpl::isConnected);
+ return producers.values().stream().allMatch(ProducerImpl::isConnected);
}
@Override
public long getLastDisconnectedTimestamp() {
long lastDisconnectedTimestamp = 0;
- Optional<ProducerImpl<T>> p = producers.stream().max(Comparator.comparingLong(ProducerImpl::getLastDisconnectedTimestamp));
+ Optional<ProducerImpl<T>> p = producers.values().stream().max(Comparator.comparingLong(ProducerImpl::getLastDisconnectedTimestamp));
if (p.isPresent()) {
lastDisconnectedTimestamp = p.get().getLastDisconnectedTimestamp();
}
@@ -232,9 +278,9 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
}
AtomicReference<Throwable> closeFail = new AtomicReference<Throwable>();
- AtomicInteger completed = new AtomicInteger(topicMetadata.numPartitions());
+ AtomicInteger completed = new AtomicInteger((int) producers.size());
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
- for (Producer<T> producer : producers) {
+ for (Producer<T> producer : producers.values()) {
if (producer != null) {
producer.closeAsync().handle((closed, ex) -> {
if (ex != null) {
@@ -268,14 +314,14 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
return null;
}
stats.reset();
- for (int i = 0; i < topicMetadata.numPartitions(); i++) {
- stats.updateCumulativeStats(producers.get(i).getStats());
- }
+ producers.values().forEach(p -> stats.updateCumulativeStats(p.getStats()));
return stats;
}
public List<ProducerImpl<T>> getProducers() {
- return producers.stream().collect(Collectors.toList());
+ return producers.values().stream()
+ .sorted(Comparator.comparingInt(e -> TopicName.getPartitionIndex(e.getTopic())))
+ .collect(Collectors.toList());
}
@Override
@@ -308,41 +354,47 @@ public class PartitionedProducerImpl<T> extends ProducerBase<T> {
future.complete(null);
return future;
} else if (oldPartitionNumber < currentPartitionNumber) {
- List<CompletableFuture<Producer<T>>> futureList = list
- .subList(oldPartitionNumber, currentPartitionNumber)
- .stream()
- .map(partitionName -> {
- int partitionIndex = TopicName.getPartitionIndex(partitionName);
- ProducerImpl<T> producer =
- new ProducerImpl<>(client,
- partitionName, conf, new CompletableFuture<>(),
- partitionIndex, schema, interceptors);
- producers.add(producer);
- return producer.producerCreatedFuture();
- }).collect(Collectors.toList());
-
- FutureUtil.waitForAll(futureList)
- .thenAccept(finalFuture -> {
- if (log.isDebugEnabled()) {
- log.debug("[{}] success create producers for extended partitions. old: {}, new: {}",
- topic, oldPartitionNumber, currentPartitionNumber);
- }
- topicMetadata = new TopicMetadataImpl(currentPartitionNumber);
- future.complete(null);
- })
- .exceptionally(ex -> {
- // error happened, remove
- log.warn("[{}] fail create producers for extended partitions. old: {}, new: {}",
- topic, oldPartitionNumber, currentPartitionNumber);
- List<ProducerImpl<T>> sublist = producers.subList(oldPartitionNumber, producers.size());
- sublist.forEach(newProducer -> newProducer.closeAsync());
- sublist.clear();
- future.completeExceptionally(ex);
- return null;
- });
- // call interceptor with the metadata change
- onPartitionsChange(topic, currentPartitionNumber);
- return null;
+ if (conf.isLazyStartPartitionedProducers() && conf.getAccessMode() == ProducerAccessMode.Shared) {
+ topicMetadata = new TopicMetadataImpl(currentPartitionNumber);
+ future.complete(null);
+ // call interceptor with the metadata change
+ onPartitionsChange(topic, currentPartitionNumber);
+ return future;
+ } else {
+ List<CompletableFuture<Producer<T>>> futureList = list
+ .subList(oldPartitionNumber, currentPartitionNumber)
+ .stream()
+ .map(partitionName -> {
+ int partitionIndex = TopicName.getPartitionIndex(partitionName);
+ return producers.computeIfAbsent(partitionIndex, (idx) -> new ProducerImpl<>(
+ client, partitionName, conf, new CompletableFuture<>(),
+ idx, schema, interceptors)).producerCreatedFuture();
+ }).collect(Collectors.toList());
+
+ FutureUtil.waitForAll(futureList)
+ .thenAccept(finalFuture -> {
+ if (log.isDebugEnabled()) {
+ log.debug(
+ "[{}] success create producers for extended partitions."
+ + " old: {}, new: {}",
+ topic, oldPartitionNumber, currentPartitionNumber);
+ }
+ topicMetadata = new TopicMetadataImpl(currentPartitionNumber);
+ future.complete(null);
+ })
+ .exceptionally(ex -> {
+ // error happened, remove
+ log.warn("[{}] fail create producers for extended partitions. old: {}, new: {}",
+ topic, oldPartitionNumber, currentPartitionNumber);
+ IntStream.range(oldPartitionNumber, (int) producers.size())
+ .forEach(i -> producers.remove(i).closeAsync());
+ future.completeExceptionally(ex);
+ return null;
+ });
+ // call interceptor with the metadata change
+ onPartitionsChange(topic, currentPartitionNumber);
+ return null;
+ }
} else {
log.error("[{}] not support shrink topic partitions. old: {}, new: {}",
topic, oldPartitionNumber, currentPartitionNumber);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
index d3a1bb1..497b82a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
@@ -322,6 +322,12 @@ public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
return this;
}
+ @Override
+ public ProducerBuilder<T> enableLazyStartPartitionedProducers(boolean lazyStartPartitionedProducers) {
+ conf.setLazyStartPartitionedProducers(lazyStartPartitionedProducers);
+ return this;
+ }
+
private void setMessageRoutingMode() throws PulsarClientException {
if(conf.getMessageRoutingMode() == null && conf.getCustomMessageRouter() == null) {
messageRoutingMode(MessageRoutingMode.RoundRobinPartition);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
index 4b96767..12fb4ab 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ProducerConfigurationData.java
@@ -102,6 +102,8 @@ public class ProducerConfigurationData implements Serializable, Cloneable {
private ProducerAccessMode accessMode = ProducerAccessMode.Shared;
+ private boolean lazyStartPartitionedProducers = false;
+
private SortedMap<String, String> properties = new TreeMap<>();
/**
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/customroute/PartialRoundRobinMessageRouterImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/customroute/PartialRoundRobinMessageRouterImpl.java
new file mode 100644
index 0000000..fe244a6
--- /dev/null
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/customroute/PartialRoundRobinMessageRouterImpl.java
@@ -0,0 +1,85 @@
+/**
+ * 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.customroute;
+
+import static org.apache.pulsar.client.util.MathUtils.signSafeMod;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.TopicMetadata;
+
+public class PartialRoundRobinMessageRouterImpl implements MessageRouter {
+ private final int numPartitionsLimit;
+ private final List<Integer> partialList = new CopyOnWriteArrayList<>();
+ private static final AtomicIntegerFieldUpdater<PartialRoundRobinMessageRouterImpl> PARTITION_INDEX_UPDATER =
+ AtomicIntegerFieldUpdater.newUpdater(PartialRoundRobinMessageRouterImpl.class, "partitionIndex");
+ @SuppressWarnings("unused")
+ private volatile int partitionIndex = 0;
+
+ public PartialRoundRobinMessageRouterImpl(final int numPartitionsLimit) {
+ if (numPartitionsLimit < 1) {
+ throw new IllegalArgumentException("numPartitionsLimit should be greater than or equal to 1");
+ }
+ this.numPartitionsLimit = numPartitionsLimit;
+ }
+
+ /**
+ * Choose a partition based on the topic metadata.
+ * Key hash routing isn't supported.
+ *
+ * @param msg message
+ * @param metadata topic metadata
+ * @return the partition to route the message.
+ */
+ public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+ final List<Integer> newPartialList = new ArrayList<>(getOrCreatePartialList(metadata));
+ return newPartialList
+ .get(signSafeMod(PARTITION_INDEX_UPDATER.getAndIncrement(this), newPartialList.size()));
+ }
+
+ private List<Integer> getOrCreatePartialList(TopicMetadata metadata) {
+ if (partialList.isEmpty()
+ || partialList.size() < numPartitionsLimit && partialList.size() < metadata.numPartitions()) {
+ synchronized (this) {
+ if (partialList.isEmpty()) {
+ partialList.addAll(IntStream.range(0, metadata.numPartitions()).boxed()
+ .collect(Collectors.collectingAndThen(Collectors.toList(), list -> {
+ Collections.shuffle(list);
+ return list.stream();
+ })).limit(numPartitionsLimit).collect(Collectors.toList()));
+ } else if (partialList.size() < numPartitionsLimit && partialList.size() < metadata.numPartitions()) {
+ partialList.addAll(IntStream.range(0, metadata.numPartitions()).boxed()
+ .filter(e -> !partialList.contains(e))
+ .collect(Collectors.collectingAndThen(Collectors.toList(), list -> {
+ Collections.shuffle(list);
+ return list.stream();
+ })).limit(numPartitionsLimit - partialList.size()).collect(Collectors.toList()));
+ }
+ }
+ }
+ return partialList;
+ }
+}
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
index d4c6aa4..1f9496b 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/PartitionedProducerImplTest.java
@@ -18,18 +18,25 @@
*/
package org.apache.pulsar.client.impl;
+import static org.apache.pulsar.client.util.MathUtils.signSafeMod;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import io.netty.channel.EventLoopGroup;
import io.netty.util.Timer;
import io.netty.util.concurrent.DefaultThreadFactory;
+import com.google.api.client.util.Lists;
import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import org.apache.pulsar.client.api.Message;
@@ -40,7 +47,10 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TopicMetadata;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.apache.pulsar.client.impl.customroute.PartialRoundRobinMessageRouterImpl;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.util.netty.EventLoopUtil;
+import org.assertj.core.util.Sets;
import org.testng.annotations.BeforeTest;
import org.testng.annotations.Test;
@@ -107,6 +117,45 @@ public class PartitionedProducerImplTest {
assertTrue(messageRouter instanceof CustomMessageRouter);
}
+ @Test
+ public void testPartialPartition() {
+ final MessageRouter router = new PartialRoundRobinMessageRouterImpl(3);
+ final Set<Integer> actualSet = Sets.newHashSet();
+ final Message<byte[]> msg = MessageImpl
+ .create(new MessageMetadata(), ByteBuffer.wrap(new byte[0]), Schema.BYTES, null);
+
+ for (int i = 0; i < 10; i++) {
+ final TopicMetadata metadata = new TopicMetadataImpl(10);
+ actualSet.add(router.choosePartition(msg, metadata));
+ }
+ assertEquals(actualSet.size(), 3);
+
+ try {
+ new PartialRoundRobinMessageRouterImpl(0);
+ fail();
+ } catch (Exception e) {
+ assertEquals(e.getClass(), IllegalArgumentException.class);
+ }
+ }
+
+ @Test
+ public void testPartialPartitionWithKey() {
+ final MessageRouter router = new PartialRoundRobinMessageRouterImpl(3);
+ final Hash hash = Murmur3_32Hash.getInstance();
+ final List<Integer> expectedHashList = Lists.newArrayList();
+ final List<Integer> actualHashList = Lists.newArrayList();
+
+ for (int i = 0; i < 10; i++) {
+ final String key = String.valueOf(i);
+ final Message<byte[]> msg = MessageImpl
+ .create(new MessageMetadata().setPartitionKey(key), ByteBuffer.wrap(new byte[0]), Schema.BYTES, null);
+ final TopicMetadata metadata = new TopicMetadataImpl(10);
+ expectedHashList.add(signSafeMod(hash.makeHash(key), 10));
+ actualHashList.add(router.choosePartition(msg, metadata));
+ }
+ assertNotEquals(actualHashList, expectedHashList);
+ }
+
private MessageRouter getMessageRouter(ProducerConfigurationData producerConfigurationData)
throws NoSuchFieldException, IllegalAccessException {
PartitionedProducerImpl impl = new PartitionedProducerImpl(