From 171ec08123a243293aa0768dce4d328c0a5532f4 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 25 Nov 2019 16:04:56 +0800 Subject: [PATCH 1/6] [Issue 5597][pulsar-client-java]retry when getPartitionedTopicMetadata failed Signed-off-by: xiaolong.ran --- .../client/PulsarBrokerStatsClientTest.java | 63 ++ .../apache/pulsar/client/impl/HttpClient.java | 6 +- .../pulsar/client/impl/PulsarClientImpl.java | 30 +- .../client/impl/PulsarClientImpl.java.orig | 783 ++++++++++++++++++ 4 files changed, 878 insertions(+), 4 deletions(-) create mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java index fca14857f1b32..00d95d1538470 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java @@ -31,6 +31,8 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats; import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats.CursorStats; import org.slf4j.Logger; @@ -47,6 +49,7 @@ import static org.mockito.Mockito.spy; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; public class PulsarBrokerStatsClientTest extends ProducerConsumerBase { @@ -132,5 +135,65 @@ public void testTopicInternalStats() throws Exception { log.info("-- Exiting {} test --", methodName); } + @Test + public void testGetPartitionedTopicMetaData() throws Exception { + log.info("-- Starting {} test --", methodName); + + final String topicName = "persistent://my-property/my-ns/my-topic1"; + final String subscriptionName = "my-subscriber-name"; + + + + try { + String url = "http://localhost:51000,localhost:" + BROKER_WEBSERVICE_PORT; + if (isTcpLookup) { + url = "pulsar://localhost:51000,localhost:" + BROKER_PORT; + } + PulsarClient client = newPulsarClient(url, 0); + + Consumer consumer = client.newConsumer().topic(topicName).subscriptionName(subscriptionName) + .acknowledgmentGroupTime(0, TimeUnit.SECONDS).subscribe(); + Producer producer = client.newProducer().topic(topicName).create(); + + consumer.close(); + producer.close(); + client.close(); + } catch (PulsarClientException pce) { + log.error("create producer or consumer error: ", pce); + fail(); + } + + log.info("-- Exiting {} test --", methodName); + } + + @Test (timeOut = 4000) + public void testGetPartitionedTopicDataTimeout() { + log.info("-- Starting {} test --", methodName); + + final String topicName = "persistent://my-property/my-ns/my-topic1"; + + String url = "http://localhost:51000,localhost:51001"; + if (isTcpLookup) { + url = "pulsar://localhost:51000,localhost:51001"; + } + + PulsarClient client; + try { + client = PulsarClient.builder() + .serviceUrl(url) + .statsInterval(0, TimeUnit.SECONDS) + .operationTimeout(3, TimeUnit.SECONDS) + .build(); + + Producer producer = client.newProducer().topic(topicName).create(); + + fail(); + } catch (PulsarClientException pce) { + log.error("create producer error: ", pce); + } + + log.info("-- Exiting {} test --", methodName); + } + private static final Logger log = LoggerFactory.getLogger(PulsarBrokerStatsClientTest.class); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java index 96c62d2e9b6ef..845c741eb43c4 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpClient.java @@ -21,6 +21,7 @@ import java.io.Closeable; import java.io.IOException; import java.net.HttpURLConnection; +import java.net.URI; import java.net.URL; import java.util.Map; import java.util.Map.Entry; @@ -127,8 +128,9 @@ public void close() throws IOException { public CompletableFuture get(String path, Class clazz) { final CompletableFuture future = new CompletableFuture<>(); try { - String requestUrl = new URL(serviceNameResolver.resolveHostUri().toURL(), path).toString(); - String remoteHostName = serviceNameResolver.resolveHostUri().getHost(); + URI hostUri = serviceNameResolver.resolveHostUri(); + String requestUrl = new URL(hostUri.toURL(), path).toString(); + String remoteHostName = hostUri.getHost(); AuthenticationDataProvider authData = authentication.getAuthData(remoteHostName); CompletableFuture> authFuture = new CompletableFuture<>(); diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index af3adda183594..ffd580c3a5c59 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -647,17 +647,43 @@ public CompletableFuture getNumberOfPartitions(String topic) { public CompletableFuture getPartitionedTopicMetadata(String topic) { - CompletableFuture metadataFuture; + CompletableFuture metadataFuture = new CompletableFuture<>(); try { TopicName topicName = TopicName.get(topic); - metadataFuture = lookup.getPartitionedTopicMetadata(topicName); + AtomicLong opTimeoutMs = new AtomicLong(conf.getOperationTimeoutMs()); + Backoff backoff = new BackoffBuilder() + .setInitialTime(100, TimeUnit.MILLISECONDS) + .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS) + .setMax(0, TimeUnit.MILLISECONDS) + .create(); + getPartitionedTopicMetadata(topicName, backoff, opTimeoutMs, metadataFuture); } catch (IllegalArgumentException e) { return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(e.getMessage())); } return metadataFuture; } + private void getPartitionedTopicMetadata(TopicName topicName, + Backoff backoff, + AtomicLong remainingTime, + CompletableFuture future) { + lookup.getPartitionedTopicMetadata(topicName).thenAccept(future::complete).exceptionally(e -> { + long nextDelay = Math.min(backoff.next(), remainingTime.get()); + if (nextDelay <= 0) { + future.completeExceptionally(new PulsarClientException + .TimeoutException("Could not getPartitionedTopicMetadata within configured timeout.")); + return null; + } + + timer.newTimeout( task -> { + remainingTime.addAndGet(-nextDelay); + getPartitionedTopicMetadata(topicName, backoff, remainingTime, future); + }, nextDelay, TimeUnit.MILLISECONDS); + return null; + }); + } + @Override public CompletableFuture> getPartitionsForTopic(String topic) { return getPartitionedTopicMetadata(topic).thenApply(metadata -> { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig new file mode 100644 index 0000000000000..af3adda183594 --- /dev/null +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig @@ -0,0 +1,783 @@ +/** + * 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.apache.commons.lang3.StringUtils.isBlank; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; + +import io.netty.channel.EventLoopGroup; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import io.netty.util.concurrent.DefaultThreadFactory; + +import java.time.Clock; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import org.apache.commons.lang3.StringUtils; +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.ConsumerBuilder; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Reader; +import org.apache.pulsar.client.api.ReaderBuilder; +import org.apache.pulsar.client.api.RegexSubscriptionMode; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.api.schema.SchemaInfoProvider; +import org.apache.pulsar.client.api.AuthenticationFactory; +import org.apache.pulsar.client.api.transaction.TransactionBuilder; +import org.apache.pulsar.client.impl.ConsumerImpl.SubscriptionMode; +import org.apache.pulsar.client.impl.conf.ClientConfigurationData; +import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; +import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; +import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; +import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; +import org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema; +import org.apache.pulsar.client.impl.schema.generic.MultiVersionSchemaInfoProvider; +import org.apache.pulsar.client.impl.transaction.TransactionBuilderImpl; +import org.apache.pulsar.client.util.ExecutorProvider; +import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace.Mode; +import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.TopicDomain; +import org.apache.pulsar.common.naming.TopicName; +import org.apache.pulsar.common.partition.PartitionedTopicMetadata; +import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.util.FutureUtil; +import org.apache.pulsar.common.util.netty.EventLoopUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class PulsarClientImpl implements PulsarClient { + + private static final Logger log = LoggerFactory.getLogger(PulsarClientImpl.class); + + private final ClientConfigurationData conf; + private LookupService lookup; + private final ConnectionPool cnxPool; + private final Timer timer; + private final ExecutorProvider externalExecutorProvider; + + enum State { + Open, Closing, Closed + } + + private AtomicReference state = new AtomicReference<>(); + private final IdentityHashMap, Boolean> producers; + private final IdentityHashMap, Boolean> consumers; + + private final AtomicLong producerIdGenerator = new AtomicLong(); + private final AtomicLong consumerIdGenerator = new AtomicLong(); + private final AtomicLong requestIdGenerator = new AtomicLong(); + + private final EventLoopGroup eventLoopGroup; + + private final LoadingCache schemaProviderLoadingCache = CacheBuilder.newBuilder().maximumSize(100000) + .expireAfterAccess(30, TimeUnit.MINUTES).build(new CacheLoader() { + + @Override + public SchemaInfoProvider load(String topicName) { + return newSchemaProvider(topicName); + } + }); + + private final Clock clientClock; + + public PulsarClientImpl(ClientConfigurationData conf) throws PulsarClientException { + this(conf, getEventLoopGroup(conf)); + } + + public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException { + this(conf, eventLoopGroup, new ConnectionPool(conf, eventLoopGroup)); + } + + public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool) + throws PulsarClientException { + if (conf == null || isBlank(conf.getServiceUrl()) || eventLoopGroup == null) { + throw new PulsarClientException.InvalidConfigurationException("Invalid client configuration"); + } + this.eventLoopGroup = eventLoopGroup; + setAuth(conf); + this.conf = conf; + this.clientClock = conf.getClock(); + conf.getAuthentication().start(); + this.cnxPool = cnxPool; + externalExecutorProvider = new ExecutorProvider(conf.getNumListenerThreads(), getThreadFactory("pulsar-external-listener")); + if (conf.getServiceUrl().startsWith("http")) { + lookup = new HttpLookupService(conf, eventLoopGroup); + } else { + lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.isUseTls(), externalExecutorProvider.getExecutor()); + } + timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS); + producers = Maps.newIdentityHashMap(); + consumers = Maps.newIdentityHashMap(); + state.set(State.Open); + } + + private void setAuth(ClientConfigurationData conf) throws PulsarClientException { + if (StringUtils.isBlank(conf.getAuthPluginClassName()) || StringUtils.isBlank( conf.getAuthParams())) { + return; + } + + conf.setAuthentication(AuthenticationFactory.create(conf.getAuthPluginClassName(), conf.getAuthParams())); + } + + public ClientConfigurationData getConfiguration() { + return conf; + } + + @VisibleForTesting + public Clock getClientClock() { + return clientClock; + } + + @Override + public ProducerBuilder newProducer() { + return new ProducerBuilderImpl<>(this, Schema.BYTES); + } + + @Override + public ProducerBuilder newProducer(Schema schema) { + return new ProducerBuilderImpl<>(this, schema); + } + + @Override + public ConsumerBuilder newConsumer() { + return new ConsumerBuilderImpl<>(this, Schema.BYTES); + } + + @Override + public ConsumerBuilder newConsumer(Schema schema) { + return new ConsumerBuilderImpl<>(this, schema); + } + + @Override + public ReaderBuilder newReader() { + return new ReaderBuilderImpl<>(this, Schema.BYTES); + } + + @Override + public ReaderBuilder newReader(Schema schema) { + return new ReaderBuilderImpl<>(this, schema); + } + + public CompletableFuture> createProducerAsync(ProducerConfigurationData conf) { + return createProducerAsync(conf, Schema.BYTES, null); + } + + public CompletableFuture> createProducerAsync(ProducerConfigurationData conf, Schema schema) { + return createProducerAsync(conf, schema, null); + } + + public CompletableFuture> createProducerAsync(ProducerConfigurationData conf, Schema schema, + ProducerInterceptors interceptors) { + if (conf == null) { + return FutureUtil.failedFuture( + new PulsarClientException.InvalidConfigurationException("Producer configuration undefined")); + } + + if (schema instanceof AutoConsumeSchema) { + return FutureUtil.failedFuture( + new PulsarClientException.InvalidConfigurationException("AutoConsumeSchema is only used by consumers to detect schemas automatically")); + } + + if (state.get() != State.Open) { + return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed : state = " + state.get())); + } + + String topic = conf.getTopicName(); + + if (!TopicName.isValid(topic)) { + return FutureUtil.failedFuture( + new PulsarClientException.InvalidTopicNameException("Invalid topic name: '" + topic + "'")); + } + + if (schema instanceof AutoProduceBytesSchema) { + AutoProduceBytesSchema autoProduceBytesSchema = (AutoProduceBytesSchema) schema; + if (autoProduceBytesSchema.schemaInitialized()) { + return createProducerAsync(topic, conf, schema, interceptors); + } + return lookup.getSchema(TopicName.get(conf.getTopicName())) + .thenCompose(schemaInfoOptional -> { + if (schemaInfoOptional.isPresent()) { + autoProduceBytesSchema.setSchema(Schema.getSchema(schemaInfoOptional.get())); + } else { + autoProduceBytesSchema.setSchema(Schema.BYTES); + } + return createProducerAsync(topic, conf, schema, interceptors); + }); + } else { + return createProducerAsync(topic, conf, schema, interceptors); + } + + } + + private CompletableFuture> createProducerAsync(String topic, + ProducerConfigurationData conf, + Schema schema, + ProducerInterceptors interceptors) { + CompletableFuture> producerCreatedFuture = new CompletableFuture<>(); + + getPartitionedTopicMetadata(topic).thenAccept(metadata -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); + } + + ProducerBase producer; + if (metadata.partitions > 0) { + producer = new PartitionedProducerImpl<>(PulsarClientImpl.this, topic, conf, metadata.partitions, + producerCreatedFuture, schema, interceptors); + } else { + producer = new ProducerImpl<>(PulsarClientImpl.this, topic, conf, producerCreatedFuture, -1, schema, interceptors); + } + + synchronized (producers) { + producers.put(producer, Boolean.TRUE); + } + }).exceptionally(ex -> { + log.warn("[{}] Failed to get partitioned topic metadata: {}", topic, ex.getMessage()); + producerCreatedFuture.completeExceptionally(ex); + return null; + }); + + return producerCreatedFuture; + } + + public CompletableFuture> subscribeAsync(ConsumerConfigurationData conf) { + return subscribeAsync(conf, Schema.BYTES, null); + } + + public CompletableFuture> subscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { + if (state.get() != State.Open) { + return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed")); + } + + if (conf == null) { + return FutureUtil.failedFuture( + new PulsarClientException.InvalidConfigurationException("Consumer configuration undefined")); + } + + if (!conf.getTopicNames().stream().allMatch(TopicName::isValid)) { + return FutureUtil.failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name")); + } + + if (isBlank(conf.getSubscriptionName())) { + return FutureUtil + .failedFuture(new PulsarClientException.InvalidConfigurationException("Empty subscription name")); + } + + if (conf.isReadCompacted() && (!conf.getTopicNames().stream() + .allMatch(topic -> TopicName.get(topic).getDomain() == TopicDomain.persistent) + || (conf.getSubscriptionType() != SubscriptionType.Exclusive + && conf.getSubscriptionType() != SubscriptionType.Failover))) { + return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException( + "Read compacted can only be used with exclusive of failover persistent subscriptions")); + } + + if (conf.getConsumerEventListener() != null && conf.getSubscriptionType() != SubscriptionType.Failover) { + return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException( + "Active consumer listener is only supported for failover subscription")); + } + + if (conf.getTopicsPattern() != null) { + // If use topicsPattern, we should not use topic(), and topics() method. + if (!conf.getTopicNames().isEmpty()){ + return FutureUtil + .failedFuture(new IllegalArgumentException("Topic names list must be null when use topicsPattern")); + } + return patternTopicSubscribeAsync(conf, schema, interceptors); + } else if (conf.getTopicNames().size() == 1) { + return singleTopicSubscribeAsync(conf, schema, interceptors); + } else { + return multiTopicSubscribeAsync(conf, schema, interceptors); + } + } + + private CompletableFuture> singleTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { + return preProcessSchemaBeforeSubscribe(this, schema, conf.getSingleTopic()) + .thenCompose(ignored -> doSingleTopicSubscribeAsync(conf, schema, interceptors)); + } + + private CompletableFuture> doSingleTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { + CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); + + String topic = conf.getSingleTopic(); + + getPartitionedTopicMetadata(topic).thenAccept(metadata -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); + } + + ConsumerBase consumer; + // gets the next single threaded executor from the list of executors + ExecutorService listenerThread = externalExecutorProvider.getExecutor(); + if (metadata.partitions > 0) { + consumer = MultiTopicsConsumerImpl.createPartitionedConsumer(PulsarClientImpl.this, conf, + listenerThread, consumerSubscribedFuture, metadata.partitions, schema, interceptors); + } else { + int partitionIndex = TopicName.getPartitionIndex(topic); + consumer = ConsumerImpl.newConsumerImpl(PulsarClientImpl.this, topic, conf, listenerThread, partitionIndex, false, + consumerSubscribedFuture, SubscriptionMode.Durable, null, schema, interceptors, + true /* createTopicIfDoesNotExist */); + } + + synchronized (consumers) { + consumers.put(consumer, Boolean.TRUE); + } + }).exceptionally(ex -> { + log.warn("[{}] Failed to get partitioned topic metadata", topic, ex); + consumerSubscribedFuture.completeExceptionally(ex); + return null; + }); + + return consumerSubscribedFuture; + } + + private CompletableFuture> multiTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { + CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); + + ConsumerBase consumer = new MultiTopicsConsumerImpl<>(PulsarClientImpl.this, conf, + externalExecutorProvider.getExecutor(), consumerSubscribedFuture, schema, interceptors, + true /* createTopicIfDoesNotExist */); + + synchronized (consumers) { + consumers.put(consumer, Boolean.TRUE); + } + + return consumerSubscribedFuture; + } + + public CompletableFuture> patternTopicSubscribeAsync(ConsumerConfigurationData conf) { + return patternTopicSubscribeAsync(conf, Schema.BYTES, null); + } + + private CompletableFuture> patternTopicSubscribeAsync(ConsumerConfigurationData conf, + Schema schema, ConsumerInterceptors interceptors) { + String regex = conf.getTopicsPattern().pattern(); + Mode subscriptionMode = convertRegexSubscriptionMode(conf.getRegexSubscriptionMode()); + TopicName destination = TopicName.get(regex); + NamespaceName namespaceName = destination.getNamespaceObject(); + + CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); + lookup.getTopicsUnderNamespace(namespaceName, subscriptionMode) + .thenAccept(topics -> { + if (log.isDebugEnabled()) { + log.debug("Get topics under namespace {}, topics.size: {}", namespaceName.toString(), topics.size()); + topics.forEach(topicName -> + log.debug("Get topics under namespace {}, topic: {}", namespaceName.toString(), topicName)); + } + + List topicsList = topicsPatternFilter(topics, conf.getTopicsPattern()); + conf.getTopicNames().addAll(topicsList); + ConsumerBase consumer = new PatternMultiTopicsConsumerImpl(conf.getTopicsPattern(), + PulsarClientImpl.this, + conf, + externalExecutorProvider.getExecutor(), + consumerSubscribedFuture, + schema, subscriptionMode, interceptors); + + synchronized (consumers) { + consumers.put(consumer, Boolean.TRUE); + } + }) + .exceptionally(ex -> { + log.warn("[{}] Failed to get topics under namespace", namespaceName); + consumerSubscribedFuture.completeExceptionally(ex); + return null; + }); + + return consumerSubscribedFuture; + } + + // get topics that match 'topicsPattern' from original topics list + // return result should contain only topic names, without partition part + public static List topicsPatternFilter(List original, Pattern topicsPattern) { + final Pattern shortenedTopicsPattern = topicsPattern.toString().contains("://") + ? Pattern.compile(topicsPattern.toString().split("\\:\\/\\/")[1]) : topicsPattern; + + return original.stream() + .map(TopicName::get) + .map(TopicName::toString) + .filter(topic -> shortenedTopicsPattern.matcher(topic.split("\\:\\/\\/")[1]).matches()) + .collect(Collectors.toList()); + } + + public CompletableFuture> createReaderAsync(ReaderConfigurationData conf) { + return createReaderAsync(conf, Schema.BYTES); + } + + public CompletableFuture> createReaderAsync(ReaderConfigurationData conf, Schema schema) { + return preProcessSchemaBeforeSubscribe(this, schema, conf.getTopicName()) + .thenCompose(ignored -> doCreateReaderAsync(conf, schema)); + } + + CompletableFuture> doCreateReaderAsync(ReaderConfigurationData conf, Schema schema) { + if (state.get() != State.Open) { + return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed")); + } + + if (conf == null) { + return FutureUtil.failedFuture( + new PulsarClientException.InvalidConfigurationException("Consumer configuration undefined")); + } + + String topic = conf.getTopicName(); + + if (!TopicName.isValid(topic)) { + return FutureUtil.failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name")); + } + + if (conf.getStartMessageId() == null) { + return FutureUtil + .failedFuture(new PulsarClientException.InvalidConfigurationException("Invalid startMessageId")); + } + + CompletableFuture> readerFuture = new CompletableFuture<>(); + + getPartitionedTopicMetadata(topic).thenAccept(metadata -> { + if (log.isDebugEnabled()) { + log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); + } + + if (metadata.partitions > 0) { + readerFuture.completeExceptionally( + new PulsarClientException("Topic reader cannot be created on a partitioned topic")); + return; + } + + CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); + // gets the next single threaded executor from the list of executors + ExecutorService listenerThread = externalExecutorProvider.getExecutor(); + ReaderImpl reader = new ReaderImpl<>(PulsarClientImpl.this, conf, listenerThread, consumerSubscribedFuture, schema); + + synchronized (consumers) { + consumers.put(reader.getConsumer(), Boolean.TRUE); + } + + consumerSubscribedFuture.thenRun(() -> { + readerFuture.complete(reader); + }).exceptionally(ex -> { + log.warn("[{}] Failed to get create topic reader", topic, ex); + readerFuture.completeExceptionally(ex); + return null; + }); + }).exceptionally(ex -> { + log.warn("[{}] Failed to get partitioned topic metadata", topic, ex); + readerFuture.completeExceptionally(ex); + return null; + }); + + return readerFuture; + } + + /** + * Read the schema information for a given topic. + * + * If the topic does not exist or it has no schema associated, it will return an empty response + */ + public CompletableFuture> getSchema(String topic) { + TopicName topicName; + try { + topicName = TopicName.get(topic); + } catch (Throwable t) { + return FutureUtil + .failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name: " + topic)); + } + + return lookup.getSchema(topicName); + } + + @Override + public void close() throws PulsarClientException { + try { + closeAsync().get(); + } catch (Exception e) { + throw PulsarClientException.unwrap(e); + } + } + + @Override + public CompletableFuture closeAsync() { + log.info("Client closing. URL: {}", lookup.getServiceUrl()); + if (!state.compareAndSet(State.Open, State.Closing)) { + return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed")); + } + + final CompletableFuture closeFuture = new CompletableFuture<>(); + List> futures = Lists.newArrayList(); + + synchronized (producers) { + // Copy to a new list, because the closing will trigger a removal from the map + // and invalidate the iterator + List> producersToClose = Lists.newArrayList(producers.keySet()); + producersToClose.forEach(p -> futures.add(p.closeAsync())); + } + + synchronized (consumers) { + List> consumersToClose = Lists.newArrayList(consumers.keySet()); + consumersToClose.forEach(c -> futures.add(c.closeAsync())); + } + + FutureUtil.waitForAll(futures).thenRun(() -> { + // All producers & consumers are now closed, we can stop the client safely + try { + shutdown(); + closeFuture.complete(null); + state.set(State.Closed); + } catch (PulsarClientException e) { + closeFuture.completeExceptionally(e); + } + }).exceptionally(exception -> { + closeFuture.completeExceptionally(exception); + return null; + }); + + return closeFuture; + } + + @Override + public void shutdown() throws PulsarClientException { + try { + lookup.close(); + cnxPool.close(); + timer.stop(); + externalExecutorProvider.shutdownNow(); + conf.getAuthentication().close(); + } catch (Throwable t) { + log.warn("Failed to shutdown Pulsar client", t); + throw PulsarClientException.unwrap(t); + } + } + + @Override + public synchronized void updateServiceUrl(String serviceUrl) throws PulsarClientException { + log.info("Updating service URL to {}", serviceUrl); + + conf.setServiceUrl(serviceUrl); + lookup.updateServiceUrl(serviceUrl); + cnxPool.closeAllConnections(); + } + + protected CompletableFuture getConnection(final String topic) { + TopicName topicName = TopicName.get(topic); + return lookup.getBroker(topicName) + .thenCompose(pair -> cnxPool.getConnection(pair.getLeft(), pair.getRight())); + } + + /** visible for pulsar-functions **/ + public Timer timer() { + return timer; + } + + ExecutorProvider externalExecutorProvider() { + return externalExecutorProvider; + } + + long newProducerId() { + return producerIdGenerator.getAndIncrement(); + } + + long newConsumerId() { + return consumerIdGenerator.getAndIncrement(); + } + + public long newRequestId() { + return requestIdGenerator.getAndIncrement(); + } + + public ConnectionPool getCnxPool() { + return cnxPool; + } + + public EventLoopGroup eventLoopGroup() { + return eventLoopGroup; + } + + public LookupService getLookup() { + return lookup; + } + + public void reloadLookUp() throws PulsarClientException { + if (conf.getServiceUrl().startsWith("http")) { + lookup = new HttpLookupService(conf, eventLoopGroup); + } else { + lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.isUseTls(), externalExecutorProvider.getExecutor()); + } + } + + public CompletableFuture getNumberOfPartitions(String topic) { + return getPartitionedTopicMetadata(topic).thenApply(metadata -> metadata.partitions); + } + + public CompletableFuture getPartitionedTopicMetadata(String topic) { + + CompletableFuture metadataFuture; + + try { + TopicName topicName = TopicName.get(topic); + metadataFuture = lookup.getPartitionedTopicMetadata(topicName); + } catch (IllegalArgumentException e) { + return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(e.getMessage())); + } + return metadataFuture; + } + + @Override + public CompletableFuture> getPartitionsForTopic(String topic) { + return getPartitionedTopicMetadata(topic).thenApply(metadata -> { + if (metadata.partitions > 0) { + TopicName topicName = TopicName.get(topic); + List partitions = new ArrayList<>(metadata.partitions); + for (int i = 0; i < metadata.partitions; i++) { + partitions.add(topicName.getPartition(i).toString()); + } + return partitions; + } else { + return Collections.singletonList(topic); + } + }); + } + + private static EventLoopGroup getEventLoopGroup(ClientConfigurationData conf) { + ThreadFactory threadFactory = getThreadFactory("pulsar-client-io"); + return EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), threadFactory); + } + + private static ThreadFactory getThreadFactory(String poolName) { + return new DefaultThreadFactory(poolName, Thread.currentThread().isDaemon()); + } + + void cleanupProducer(ProducerBase producer) { + synchronized (producers) { + producers.remove(producer); + } + } + + void cleanupConsumer(ConsumerBase consumer) { + synchronized (consumers) { + consumers.remove(consumer); + } + } + + @VisibleForTesting + int producersCount() { + synchronized (producers) { + return producers.size(); + } + } + + @VisibleForTesting + int consumersCount() { + synchronized (consumers) { + return consumers.size(); + } + } + + private static Mode convertRegexSubscriptionMode(RegexSubscriptionMode regexSubscriptionMode) { + switch (regexSubscriptionMode) { + case PersistentOnly: + return Mode.PERSISTENT; + case NonPersistentOnly: + return Mode.NON_PERSISTENT; + case AllTopics: + return Mode.ALL; + default: + return null; + } + } + + private SchemaInfoProvider newSchemaProvider(String topicName) { + return new MultiVersionSchemaInfoProvider(TopicName.get(topicName), this); + } + + private LoadingCache getSchemaProviderLoadingCache() { + return schemaProviderLoadingCache; + } + + @SuppressWarnings("unchecked") + protected CompletableFuture preProcessSchemaBeforeSubscribe(PulsarClientImpl pulsarClientImpl, + Schema schema, + String topicName) { + if (schema != null && schema.supportSchemaVersioning()) { + final SchemaInfoProvider schemaInfoProvider; + try { + schemaInfoProvider = pulsarClientImpl.getSchemaProviderLoadingCache().get(topicName); + } catch (ExecutionException e) { + log.error("Failed to load schema info provider for topic {}", topicName, e); + return FutureUtil.failedFuture(e.getCause()); + } + + if (schema.requireFetchingSchemaInfo()) { + return schemaInfoProvider.getLatestSchema().thenCompose(schemaInfo -> { + if (null == schemaInfo) { + if (!(schema instanceof AutoConsumeSchema)) { + // no schema info is found + return FutureUtil.failedFuture( + new PulsarClientException.NotFoundException( + "No latest schema found for topic " + topicName)); + } + } + try { + log.info("Configuring schema for topic {} : {}", topicName, schemaInfo); + schema.configureSchemaInfo(topicName, "topic", schemaInfo); + } catch (RuntimeException re) { + return FutureUtil.failedFuture(re); + } + schema.setSchemaInfoProvider(schemaInfoProvider); + return CompletableFuture.completedFuture(null); + }); + } else { + schema.setSchemaInfoProvider(schemaInfoProvider); + } + } + return CompletableFuture.completedFuture(null); + } + + // + // Transaction related API + // + + // This method should be exposed in the PulsarClient interface. Only expose it when all the transaction features + // are completed. + // @Override + public TransactionBuilder newTransaction() { + return new TransactionBuilderImpl(this); + } + +} From eca62d80eb81b755bad83b7c5b2ccecfbf15fe8b Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 25 Nov 2019 16:10:34 +0800 Subject: [PATCH 2/6] remove unuse code Signed-off-by: xiaolong.ran --- .../client/impl/PulsarClientImpl.java.orig | 783 ------------------ 1 file changed, 783 deletions(-) delete mode 100644 pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig deleted file mode 100644 index af3adda183594..0000000000000 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java.orig +++ /dev/null @@ -1,783 +0,0 @@ -/** - * 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.apache.commons.lang3.StringUtils.isBlank; - -import com.google.common.annotations.VisibleForTesting; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import com.google.common.collect.Lists; -import com.google.common.collect.Maps; - -import io.netty.channel.EventLoopGroup; -import io.netty.util.HashedWheelTimer; -import io.netty.util.Timer; -import io.netty.util.concurrent.DefaultThreadFactory; - -import java.time.Clock; -import java.util.ArrayList; -import java.util.Collections; -import java.util.IdentityHashMap; -import java.util.List; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.AtomicReference; -import java.util.regex.Pattern; -import java.util.stream.Collectors; - -import org.apache.commons.lang3.StringUtils; -import org.apache.pulsar.client.api.Consumer; -import org.apache.pulsar.client.api.ConsumerBuilder; -import org.apache.pulsar.client.api.Producer; -import org.apache.pulsar.client.api.ProducerBuilder; -import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.api.Reader; -import org.apache.pulsar.client.api.ReaderBuilder; -import org.apache.pulsar.client.api.RegexSubscriptionMode; -import org.apache.pulsar.client.api.Schema; -import org.apache.pulsar.client.api.SubscriptionType; -import org.apache.pulsar.client.api.schema.SchemaInfoProvider; -import org.apache.pulsar.client.api.AuthenticationFactory; -import org.apache.pulsar.client.api.transaction.TransactionBuilder; -import org.apache.pulsar.client.impl.ConsumerImpl.SubscriptionMode; -import org.apache.pulsar.client.impl.conf.ClientConfigurationData; -import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; -import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; -import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; -import org.apache.pulsar.client.impl.schema.AutoConsumeSchema; -import org.apache.pulsar.client.impl.schema.AutoProduceBytesSchema; -import org.apache.pulsar.client.impl.schema.generic.MultiVersionSchemaInfoProvider; -import org.apache.pulsar.client.impl.transaction.TransactionBuilderImpl; -import org.apache.pulsar.client.util.ExecutorProvider; -import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace.Mode; -import org.apache.pulsar.common.naming.NamespaceName; -import org.apache.pulsar.common.naming.TopicDomain; -import org.apache.pulsar.common.naming.TopicName; -import org.apache.pulsar.common.partition.PartitionedTopicMetadata; -import org.apache.pulsar.common.schema.SchemaInfo; -import org.apache.pulsar.common.util.FutureUtil; -import org.apache.pulsar.common.util.netty.EventLoopUtil; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class PulsarClientImpl implements PulsarClient { - - private static final Logger log = LoggerFactory.getLogger(PulsarClientImpl.class); - - private final ClientConfigurationData conf; - private LookupService lookup; - private final ConnectionPool cnxPool; - private final Timer timer; - private final ExecutorProvider externalExecutorProvider; - - enum State { - Open, Closing, Closed - } - - private AtomicReference state = new AtomicReference<>(); - private final IdentityHashMap, Boolean> producers; - private final IdentityHashMap, Boolean> consumers; - - private final AtomicLong producerIdGenerator = new AtomicLong(); - private final AtomicLong consumerIdGenerator = new AtomicLong(); - private final AtomicLong requestIdGenerator = new AtomicLong(); - - private final EventLoopGroup eventLoopGroup; - - private final LoadingCache schemaProviderLoadingCache = CacheBuilder.newBuilder().maximumSize(100000) - .expireAfterAccess(30, TimeUnit.MINUTES).build(new CacheLoader() { - - @Override - public SchemaInfoProvider load(String topicName) { - return newSchemaProvider(topicName); - } - }); - - private final Clock clientClock; - - public PulsarClientImpl(ClientConfigurationData conf) throws PulsarClientException { - this(conf, getEventLoopGroup(conf)); - } - - public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException { - this(conf, eventLoopGroup, new ConnectionPool(conf, eventLoopGroup)); - } - - public PulsarClientImpl(ClientConfigurationData conf, EventLoopGroup eventLoopGroup, ConnectionPool cnxPool) - throws PulsarClientException { - if (conf == null || isBlank(conf.getServiceUrl()) || eventLoopGroup == null) { - throw new PulsarClientException.InvalidConfigurationException("Invalid client configuration"); - } - this.eventLoopGroup = eventLoopGroup; - setAuth(conf); - this.conf = conf; - this.clientClock = conf.getClock(); - conf.getAuthentication().start(); - this.cnxPool = cnxPool; - externalExecutorProvider = new ExecutorProvider(conf.getNumListenerThreads(), getThreadFactory("pulsar-external-listener")); - if (conf.getServiceUrl().startsWith("http")) { - lookup = new HttpLookupService(conf, eventLoopGroup); - } else { - lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.isUseTls(), externalExecutorProvider.getExecutor()); - } - timer = new HashedWheelTimer(getThreadFactory("pulsar-timer"), 1, TimeUnit.MILLISECONDS); - producers = Maps.newIdentityHashMap(); - consumers = Maps.newIdentityHashMap(); - state.set(State.Open); - } - - private void setAuth(ClientConfigurationData conf) throws PulsarClientException { - if (StringUtils.isBlank(conf.getAuthPluginClassName()) || StringUtils.isBlank( conf.getAuthParams())) { - return; - } - - conf.setAuthentication(AuthenticationFactory.create(conf.getAuthPluginClassName(), conf.getAuthParams())); - } - - public ClientConfigurationData getConfiguration() { - return conf; - } - - @VisibleForTesting - public Clock getClientClock() { - return clientClock; - } - - @Override - public ProducerBuilder newProducer() { - return new ProducerBuilderImpl<>(this, Schema.BYTES); - } - - @Override - public ProducerBuilder newProducer(Schema schema) { - return new ProducerBuilderImpl<>(this, schema); - } - - @Override - public ConsumerBuilder newConsumer() { - return new ConsumerBuilderImpl<>(this, Schema.BYTES); - } - - @Override - public ConsumerBuilder newConsumer(Schema schema) { - return new ConsumerBuilderImpl<>(this, schema); - } - - @Override - public ReaderBuilder newReader() { - return new ReaderBuilderImpl<>(this, Schema.BYTES); - } - - @Override - public ReaderBuilder newReader(Schema schema) { - return new ReaderBuilderImpl<>(this, schema); - } - - public CompletableFuture> createProducerAsync(ProducerConfigurationData conf) { - return createProducerAsync(conf, Schema.BYTES, null); - } - - public CompletableFuture> createProducerAsync(ProducerConfigurationData conf, Schema schema) { - return createProducerAsync(conf, schema, null); - } - - public CompletableFuture> createProducerAsync(ProducerConfigurationData conf, Schema schema, - ProducerInterceptors interceptors) { - if (conf == null) { - return FutureUtil.failedFuture( - new PulsarClientException.InvalidConfigurationException("Producer configuration undefined")); - } - - if (schema instanceof AutoConsumeSchema) { - return FutureUtil.failedFuture( - new PulsarClientException.InvalidConfigurationException("AutoConsumeSchema is only used by consumers to detect schemas automatically")); - } - - if (state.get() != State.Open) { - return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed : state = " + state.get())); - } - - String topic = conf.getTopicName(); - - if (!TopicName.isValid(topic)) { - return FutureUtil.failedFuture( - new PulsarClientException.InvalidTopicNameException("Invalid topic name: '" + topic + "'")); - } - - if (schema instanceof AutoProduceBytesSchema) { - AutoProduceBytesSchema autoProduceBytesSchema = (AutoProduceBytesSchema) schema; - if (autoProduceBytesSchema.schemaInitialized()) { - return createProducerAsync(topic, conf, schema, interceptors); - } - return lookup.getSchema(TopicName.get(conf.getTopicName())) - .thenCompose(schemaInfoOptional -> { - if (schemaInfoOptional.isPresent()) { - autoProduceBytesSchema.setSchema(Schema.getSchema(schemaInfoOptional.get())); - } else { - autoProduceBytesSchema.setSchema(Schema.BYTES); - } - return createProducerAsync(topic, conf, schema, interceptors); - }); - } else { - return createProducerAsync(topic, conf, schema, interceptors); - } - - } - - private CompletableFuture> createProducerAsync(String topic, - ProducerConfigurationData conf, - Schema schema, - ProducerInterceptors interceptors) { - CompletableFuture> producerCreatedFuture = new CompletableFuture<>(); - - getPartitionedTopicMetadata(topic).thenAccept(metadata -> { - if (log.isDebugEnabled()) { - log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); - } - - ProducerBase producer; - if (metadata.partitions > 0) { - producer = new PartitionedProducerImpl<>(PulsarClientImpl.this, topic, conf, metadata.partitions, - producerCreatedFuture, schema, interceptors); - } else { - producer = new ProducerImpl<>(PulsarClientImpl.this, topic, conf, producerCreatedFuture, -1, schema, interceptors); - } - - synchronized (producers) { - producers.put(producer, Boolean.TRUE); - } - }).exceptionally(ex -> { - log.warn("[{}] Failed to get partitioned topic metadata: {}", topic, ex.getMessage()); - producerCreatedFuture.completeExceptionally(ex); - return null; - }); - - return producerCreatedFuture; - } - - public CompletableFuture> subscribeAsync(ConsumerConfigurationData conf) { - return subscribeAsync(conf, Schema.BYTES, null); - } - - public CompletableFuture> subscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { - if (state.get() != State.Open) { - return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed")); - } - - if (conf == null) { - return FutureUtil.failedFuture( - new PulsarClientException.InvalidConfigurationException("Consumer configuration undefined")); - } - - if (!conf.getTopicNames().stream().allMatch(TopicName::isValid)) { - return FutureUtil.failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name")); - } - - if (isBlank(conf.getSubscriptionName())) { - return FutureUtil - .failedFuture(new PulsarClientException.InvalidConfigurationException("Empty subscription name")); - } - - if (conf.isReadCompacted() && (!conf.getTopicNames().stream() - .allMatch(topic -> TopicName.get(topic).getDomain() == TopicDomain.persistent) - || (conf.getSubscriptionType() != SubscriptionType.Exclusive - && conf.getSubscriptionType() != SubscriptionType.Failover))) { - return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException( - "Read compacted can only be used with exclusive of failover persistent subscriptions")); - } - - if (conf.getConsumerEventListener() != null && conf.getSubscriptionType() != SubscriptionType.Failover) { - return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException( - "Active consumer listener is only supported for failover subscription")); - } - - if (conf.getTopicsPattern() != null) { - // If use topicsPattern, we should not use topic(), and topics() method. - if (!conf.getTopicNames().isEmpty()){ - return FutureUtil - .failedFuture(new IllegalArgumentException("Topic names list must be null when use topicsPattern")); - } - return patternTopicSubscribeAsync(conf, schema, interceptors); - } else if (conf.getTopicNames().size() == 1) { - return singleTopicSubscribeAsync(conf, schema, interceptors); - } else { - return multiTopicSubscribeAsync(conf, schema, interceptors); - } - } - - private CompletableFuture> singleTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { - return preProcessSchemaBeforeSubscribe(this, schema, conf.getSingleTopic()) - .thenCompose(ignored -> doSingleTopicSubscribeAsync(conf, schema, interceptors)); - } - - private CompletableFuture> doSingleTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { - CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); - - String topic = conf.getSingleTopic(); - - getPartitionedTopicMetadata(topic).thenAccept(metadata -> { - if (log.isDebugEnabled()) { - log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); - } - - ConsumerBase consumer; - // gets the next single threaded executor from the list of executors - ExecutorService listenerThread = externalExecutorProvider.getExecutor(); - if (metadata.partitions > 0) { - consumer = MultiTopicsConsumerImpl.createPartitionedConsumer(PulsarClientImpl.this, conf, - listenerThread, consumerSubscribedFuture, metadata.partitions, schema, interceptors); - } else { - int partitionIndex = TopicName.getPartitionIndex(topic); - consumer = ConsumerImpl.newConsumerImpl(PulsarClientImpl.this, topic, conf, listenerThread, partitionIndex, false, - consumerSubscribedFuture, SubscriptionMode.Durable, null, schema, interceptors, - true /* createTopicIfDoesNotExist */); - } - - synchronized (consumers) { - consumers.put(consumer, Boolean.TRUE); - } - }).exceptionally(ex -> { - log.warn("[{}] Failed to get partitioned topic metadata", topic, ex); - consumerSubscribedFuture.completeExceptionally(ex); - return null; - }); - - return consumerSubscribedFuture; - } - - private CompletableFuture> multiTopicSubscribeAsync(ConsumerConfigurationData conf, Schema schema, ConsumerInterceptors interceptors) { - CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); - - ConsumerBase consumer = new MultiTopicsConsumerImpl<>(PulsarClientImpl.this, conf, - externalExecutorProvider.getExecutor(), consumerSubscribedFuture, schema, interceptors, - true /* createTopicIfDoesNotExist */); - - synchronized (consumers) { - consumers.put(consumer, Boolean.TRUE); - } - - return consumerSubscribedFuture; - } - - public CompletableFuture> patternTopicSubscribeAsync(ConsumerConfigurationData conf) { - return patternTopicSubscribeAsync(conf, Schema.BYTES, null); - } - - private CompletableFuture> patternTopicSubscribeAsync(ConsumerConfigurationData conf, - Schema schema, ConsumerInterceptors interceptors) { - String regex = conf.getTopicsPattern().pattern(); - Mode subscriptionMode = convertRegexSubscriptionMode(conf.getRegexSubscriptionMode()); - TopicName destination = TopicName.get(regex); - NamespaceName namespaceName = destination.getNamespaceObject(); - - CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); - lookup.getTopicsUnderNamespace(namespaceName, subscriptionMode) - .thenAccept(topics -> { - if (log.isDebugEnabled()) { - log.debug("Get topics under namespace {}, topics.size: {}", namespaceName.toString(), topics.size()); - topics.forEach(topicName -> - log.debug("Get topics under namespace {}, topic: {}", namespaceName.toString(), topicName)); - } - - List topicsList = topicsPatternFilter(topics, conf.getTopicsPattern()); - conf.getTopicNames().addAll(topicsList); - ConsumerBase consumer = new PatternMultiTopicsConsumerImpl(conf.getTopicsPattern(), - PulsarClientImpl.this, - conf, - externalExecutorProvider.getExecutor(), - consumerSubscribedFuture, - schema, subscriptionMode, interceptors); - - synchronized (consumers) { - consumers.put(consumer, Boolean.TRUE); - } - }) - .exceptionally(ex -> { - log.warn("[{}] Failed to get topics under namespace", namespaceName); - consumerSubscribedFuture.completeExceptionally(ex); - return null; - }); - - return consumerSubscribedFuture; - } - - // get topics that match 'topicsPattern' from original topics list - // return result should contain only topic names, without partition part - public static List topicsPatternFilter(List original, Pattern topicsPattern) { - final Pattern shortenedTopicsPattern = topicsPattern.toString().contains("://") - ? Pattern.compile(topicsPattern.toString().split("\\:\\/\\/")[1]) : topicsPattern; - - return original.stream() - .map(TopicName::get) - .map(TopicName::toString) - .filter(topic -> shortenedTopicsPattern.matcher(topic.split("\\:\\/\\/")[1]).matches()) - .collect(Collectors.toList()); - } - - public CompletableFuture> createReaderAsync(ReaderConfigurationData conf) { - return createReaderAsync(conf, Schema.BYTES); - } - - public CompletableFuture> createReaderAsync(ReaderConfigurationData conf, Schema schema) { - return preProcessSchemaBeforeSubscribe(this, schema, conf.getTopicName()) - .thenCompose(ignored -> doCreateReaderAsync(conf, schema)); - } - - CompletableFuture> doCreateReaderAsync(ReaderConfigurationData conf, Schema schema) { - if (state.get() != State.Open) { - return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed")); - } - - if (conf == null) { - return FutureUtil.failedFuture( - new PulsarClientException.InvalidConfigurationException("Consumer configuration undefined")); - } - - String topic = conf.getTopicName(); - - if (!TopicName.isValid(topic)) { - return FutureUtil.failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name")); - } - - if (conf.getStartMessageId() == null) { - return FutureUtil - .failedFuture(new PulsarClientException.InvalidConfigurationException("Invalid startMessageId")); - } - - CompletableFuture> readerFuture = new CompletableFuture<>(); - - getPartitionedTopicMetadata(topic).thenAccept(metadata -> { - if (log.isDebugEnabled()) { - log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions); - } - - if (metadata.partitions > 0) { - readerFuture.completeExceptionally( - new PulsarClientException("Topic reader cannot be created on a partitioned topic")); - return; - } - - CompletableFuture> consumerSubscribedFuture = new CompletableFuture<>(); - // gets the next single threaded executor from the list of executors - ExecutorService listenerThread = externalExecutorProvider.getExecutor(); - ReaderImpl reader = new ReaderImpl<>(PulsarClientImpl.this, conf, listenerThread, consumerSubscribedFuture, schema); - - synchronized (consumers) { - consumers.put(reader.getConsumer(), Boolean.TRUE); - } - - consumerSubscribedFuture.thenRun(() -> { - readerFuture.complete(reader); - }).exceptionally(ex -> { - log.warn("[{}] Failed to get create topic reader", topic, ex); - readerFuture.completeExceptionally(ex); - return null; - }); - }).exceptionally(ex -> { - log.warn("[{}] Failed to get partitioned topic metadata", topic, ex); - readerFuture.completeExceptionally(ex); - return null; - }); - - return readerFuture; - } - - /** - * Read the schema information for a given topic. - * - * If the topic does not exist or it has no schema associated, it will return an empty response - */ - public CompletableFuture> getSchema(String topic) { - TopicName topicName; - try { - topicName = TopicName.get(topic); - } catch (Throwable t) { - return FutureUtil - .failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name: " + topic)); - } - - return lookup.getSchema(topicName); - } - - @Override - public void close() throws PulsarClientException { - try { - closeAsync().get(); - } catch (Exception e) { - throw PulsarClientException.unwrap(e); - } - } - - @Override - public CompletableFuture closeAsync() { - log.info("Client closing. URL: {}", lookup.getServiceUrl()); - if (!state.compareAndSet(State.Open, State.Closing)) { - return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed")); - } - - final CompletableFuture closeFuture = new CompletableFuture<>(); - List> futures = Lists.newArrayList(); - - synchronized (producers) { - // Copy to a new list, because the closing will trigger a removal from the map - // and invalidate the iterator - List> producersToClose = Lists.newArrayList(producers.keySet()); - producersToClose.forEach(p -> futures.add(p.closeAsync())); - } - - synchronized (consumers) { - List> consumersToClose = Lists.newArrayList(consumers.keySet()); - consumersToClose.forEach(c -> futures.add(c.closeAsync())); - } - - FutureUtil.waitForAll(futures).thenRun(() -> { - // All producers & consumers are now closed, we can stop the client safely - try { - shutdown(); - closeFuture.complete(null); - state.set(State.Closed); - } catch (PulsarClientException e) { - closeFuture.completeExceptionally(e); - } - }).exceptionally(exception -> { - closeFuture.completeExceptionally(exception); - return null; - }); - - return closeFuture; - } - - @Override - public void shutdown() throws PulsarClientException { - try { - lookup.close(); - cnxPool.close(); - timer.stop(); - externalExecutorProvider.shutdownNow(); - conf.getAuthentication().close(); - } catch (Throwable t) { - log.warn("Failed to shutdown Pulsar client", t); - throw PulsarClientException.unwrap(t); - } - } - - @Override - public synchronized void updateServiceUrl(String serviceUrl) throws PulsarClientException { - log.info("Updating service URL to {}", serviceUrl); - - conf.setServiceUrl(serviceUrl); - lookup.updateServiceUrl(serviceUrl); - cnxPool.closeAllConnections(); - } - - protected CompletableFuture getConnection(final String topic) { - TopicName topicName = TopicName.get(topic); - return lookup.getBroker(topicName) - .thenCompose(pair -> cnxPool.getConnection(pair.getLeft(), pair.getRight())); - } - - /** visible for pulsar-functions **/ - public Timer timer() { - return timer; - } - - ExecutorProvider externalExecutorProvider() { - return externalExecutorProvider; - } - - long newProducerId() { - return producerIdGenerator.getAndIncrement(); - } - - long newConsumerId() { - return consumerIdGenerator.getAndIncrement(); - } - - public long newRequestId() { - return requestIdGenerator.getAndIncrement(); - } - - public ConnectionPool getCnxPool() { - return cnxPool; - } - - public EventLoopGroup eventLoopGroup() { - return eventLoopGroup; - } - - public LookupService getLookup() { - return lookup; - } - - public void reloadLookUp() throws PulsarClientException { - if (conf.getServiceUrl().startsWith("http")) { - lookup = new HttpLookupService(conf, eventLoopGroup); - } else { - lookup = new BinaryProtoLookupService(this, conf.getServiceUrl(), conf.isUseTls(), externalExecutorProvider.getExecutor()); - } - } - - public CompletableFuture getNumberOfPartitions(String topic) { - return getPartitionedTopicMetadata(topic).thenApply(metadata -> metadata.partitions); - } - - public CompletableFuture getPartitionedTopicMetadata(String topic) { - - CompletableFuture metadataFuture; - - try { - TopicName topicName = TopicName.get(topic); - metadataFuture = lookup.getPartitionedTopicMetadata(topicName); - } catch (IllegalArgumentException e) { - return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(e.getMessage())); - } - return metadataFuture; - } - - @Override - public CompletableFuture> getPartitionsForTopic(String topic) { - return getPartitionedTopicMetadata(topic).thenApply(metadata -> { - if (metadata.partitions > 0) { - TopicName topicName = TopicName.get(topic); - List partitions = new ArrayList<>(metadata.partitions); - for (int i = 0; i < metadata.partitions; i++) { - partitions.add(topicName.getPartition(i).toString()); - } - return partitions; - } else { - return Collections.singletonList(topic); - } - }); - } - - private static EventLoopGroup getEventLoopGroup(ClientConfigurationData conf) { - ThreadFactory threadFactory = getThreadFactory("pulsar-client-io"); - return EventLoopUtil.newEventLoopGroup(conf.getNumIoThreads(), threadFactory); - } - - private static ThreadFactory getThreadFactory(String poolName) { - return new DefaultThreadFactory(poolName, Thread.currentThread().isDaemon()); - } - - void cleanupProducer(ProducerBase producer) { - synchronized (producers) { - producers.remove(producer); - } - } - - void cleanupConsumer(ConsumerBase consumer) { - synchronized (consumers) { - consumers.remove(consumer); - } - } - - @VisibleForTesting - int producersCount() { - synchronized (producers) { - return producers.size(); - } - } - - @VisibleForTesting - int consumersCount() { - synchronized (consumers) { - return consumers.size(); - } - } - - private static Mode convertRegexSubscriptionMode(RegexSubscriptionMode regexSubscriptionMode) { - switch (regexSubscriptionMode) { - case PersistentOnly: - return Mode.PERSISTENT; - case NonPersistentOnly: - return Mode.NON_PERSISTENT; - case AllTopics: - return Mode.ALL; - default: - return null; - } - } - - private SchemaInfoProvider newSchemaProvider(String topicName) { - return new MultiVersionSchemaInfoProvider(TopicName.get(topicName), this); - } - - private LoadingCache getSchemaProviderLoadingCache() { - return schemaProviderLoadingCache; - } - - @SuppressWarnings("unchecked") - protected CompletableFuture preProcessSchemaBeforeSubscribe(PulsarClientImpl pulsarClientImpl, - Schema schema, - String topicName) { - if (schema != null && schema.supportSchemaVersioning()) { - final SchemaInfoProvider schemaInfoProvider; - try { - schemaInfoProvider = pulsarClientImpl.getSchemaProviderLoadingCache().get(topicName); - } catch (ExecutionException e) { - log.error("Failed to load schema info provider for topic {}", topicName, e); - return FutureUtil.failedFuture(e.getCause()); - } - - if (schema.requireFetchingSchemaInfo()) { - return schemaInfoProvider.getLatestSchema().thenCompose(schemaInfo -> { - if (null == schemaInfo) { - if (!(schema instanceof AutoConsumeSchema)) { - // no schema info is found - return FutureUtil.failedFuture( - new PulsarClientException.NotFoundException( - "No latest schema found for topic " + topicName)); - } - } - try { - log.info("Configuring schema for topic {} : {}", topicName, schemaInfo); - schema.configureSchemaInfo(topicName, "topic", schemaInfo); - } catch (RuntimeException re) { - return FutureUtil.failedFuture(re); - } - schema.setSchemaInfoProvider(schemaInfoProvider); - return CompletableFuture.completedFuture(null); - }); - } else { - schema.setSchemaInfoProvider(schemaInfoProvider); - } - } - return CompletableFuture.completedFuture(null); - } - - // - // Transaction related API - // - - // This method should be exposed in the PulsarClient interface. Only expose it when all the transaction features - // are completed. - // @Override - public TransactionBuilder newTransaction() { - return new TransactionBuilderImpl(this); - } - -} From b6b6521b473a06aa6ba55c1909de7e15bc6635d7 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Tue, 26 Nov 2019 19:38:04 +0800 Subject: [PATCH 3/6] fix ci error Signed-off-by: xiaolong.ran --- .../org/apache/pulsar/client/impl/ConnectionTimeoutTest.java | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java index eac23c5211107..b67ae75b5c03e 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java @@ -25,6 +25,7 @@ import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; import org.testng.Assert; import org.testng.annotations.Test; @@ -48,9 +49,7 @@ public void testLowTimeout() throws Exception { Assert.fail("Shouldn't be able to connect to anything"); } catch (Exception e) { Assert.assertFalse(defaultFuture.isDone()); - Assert.assertEquals(e.getCause().getCause().getCause().getClass(), - ConnectTimeoutException.class); - Assert.assertTrue((System.nanoTime() - startNanos) < TimeUnit.SECONDS.toNanos(3)); + Assert.assertEquals(e.getCause().getClass(), PulsarClientException.TimeoutException.class); } } } From 1d945390ead89bb85a7cf74c9ed012960554a121 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Wed, 27 Nov 2019 11:46:53 +0800 Subject: [PATCH 4/6] fix ci error Signed-off-by: xiaolong.ran --- .../org/apache/pulsar/client/impl/PulsarClientImpl.java | 7 +++---- .../apache/pulsar/client/impl/ConnectionTimeoutTest.java | 3 ++- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index ffd580c3a5c59..63b3bd4abfe41 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -653,9 +653,9 @@ public CompletableFuture getPartitionedTopicMetadata(S TopicName topicName = TopicName.get(topic); AtomicLong opTimeoutMs = new AtomicLong(conf.getOperationTimeoutMs()); Backoff backoff = new BackoffBuilder() - .setInitialTime(100, TimeUnit.MILLISECONDS) + .setInitialTime(TimeUnit.MILLISECONDS.toNanos(100), TimeUnit.NANOSECONDS) .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS) - .setMax(0, TimeUnit.MILLISECONDS) + .setMax(TimeUnit.SECONDS.toNanos(60), TimeUnit.NANOSECONDS) .create(); getPartitionedTopicMetadata(topicName, backoff, opTimeoutMs, metadataFuture); } catch (IllegalArgumentException e) { @@ -671,8 +671,7 @@ private void getPartitionedTopicMetadata(TopicName topicName, lookup.getPartitionedTopicMetadata(topicName).thenAccept(future::complete).exceptionally(e -> { long nextDelay = Math.min(backoff.next(), remainingTime.get()); if (nextDelay <= 0) { - future.completeExceptionally(new PulsarClientException - .TimeoutException("Could not getPartitionedTopicMetadata within configured timeout.")); + future.completeExceptionally(e); return null; } diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java index b67ae75b5c03e..fb072e573891c 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConnectionTimeoutTest.java @@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import org.apache.http.conn.ConnectionPoolTimeoutException; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; @@ -49,7 +50,7 @@ public void testLowTimeout() throws Exception { Assert.fail("Shouldn't be able to connect to anything"); } catch (Exception e) { Assert.assertFalse(defaultFuture.isDone()); - Assert.assertEquals(e.getCause().getClass(), PulsarClientException.TimeoutException.class); + Assert.assertEquals(e.getCause().getCause().getCause().getClass(), ConnectTimeoutException.class); } } } From 4cde91210195563dd729261e4a6164e4a82771be Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 2 Dec 2019 13:11:15 +0800 Subject: [PATCH 5/6] fix ci error Signed-off-by: xiaolong.ran --- .../java/org/apache/pulsar/client/impl/PulsarClientImpl.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java index 517764123e7f9..f053da8608e51 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java @@ -657,9 +657,9 @@ public CompletableFuture getPartitionedTopicMetadata(S TopicName topicName = TopicName.get(topic); AtomicLong opTimeoutMs = new AtomicLong(conf.getOperationTimeoutMs()); Backoff backoff = new BackoffBuilder() - .setInitialTime(TimeUnit.MILLISECONDS.toNanos(100), TimeUnit.NANOSECONDS) + .setInitialTime(100, TimeUnit.NANOSECONDS) .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS) - .setMax(TimeUnit.SECONDS.toNanos(60), TimeUnit.NANOSECONDS) + .setMax(0, TimeUnit.NANOSECONDS) .create(); getPartitionedTopicMetadata(topicName, backoff, opTimeoutMs, metadataFuture); } catch (IllegalArgumentException e) { From d58344376de3a55c4d0e582332b77933171031b7 Mon Sep 17 00:00:00 2001 From: "xiaolong.ran" Date: Mon, 2 Dec 2019 15:37:27 +0800 Subject: [PATCH 6/6] fix ci error Signed-off-by: xiaolong.ran --- .../pulsar/stats/client/PulsarBrokerStatsClientTest.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java index 00d95d1538470..f826453799f8c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/stats/client/PulsarBrokerStatsClientTest.java @@ -142,12 +142,10 @@ public void testGetPartitionedTopicMetaData() throws Exception { final String topicName = "persistent://my-property/my-ns/my-topic1"; final String subscriptionName = "my-subscriber-name"; - - try { - String url = "http://localhost:51000,localhost:" + BROKER_WEBSERVICE_PORT; + String url = "http://localhost:" + BROKER_WEBSERVICE_PORT; if (isTcpLookup) { - url = "pulsar://localhost:51000,localhost:" + BROKER_PORT; + url = "pulsar://localhost:" + BROKER_PORT; } PulsarClient client = newPulsarClient(url, 0);