zhaijack commented on a change in pull request #1103: PIP-13-1/3: Provide `TopicsConsumer` to consume from several topics under same namespace URL: https://github.com/apache/incubator-pulsar/pull/1103#discussion_r167915437
########## File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java ########## @@ -0,0 +1,881 @@ +/** + * 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 com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkState; + +import com.google.common.collect.Lists; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import org.apache.pulsar.client.api.Consumer; +import org.apache.pulsar.client.api.ConsumerConfiguration; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.SubscriptionType; +import org.apache.pulsar.client.util.ConsumerName; +import org.apache.pulsar.client.util.FutureUtil; +import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType; +import org.apache.pulsar.common.naming.DestinationName; +import org.apache.pulsar.common.naming.NamespaceName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class TopicsConsumerImpl extends ConsumerBase { + + // All topics should be in same namespace + protected NamespaceName namespaceName; + + // Map <topic+partition, consumer>, when get do ACK, consumer will by find by topic name + private final ConcurrentHashMap<String, ConsumerImpl> consumers; + + // Map <topic, partitionNumber>, store partition number for each topic + private final ConcurrentHashMap<String, Integer> topics; + + // Queue of partition consumers on which we have stopped calling receiveAsync() because the + // shared incoming queue was full + private final ConcurrentLinkedQueue<ConsumerImpl> pausedConsumers; + + // Threshold for the shared queue. When the size of the shared queue goes below the threshold, we are going to + // resume receiving from the paused consumer partitions + private final int sharedQueueResumeThreshold; + + // sum of topicPartitions, simple topic has 1, partitioned topic equals to partition number. + AtomicInteger numberTopicPartitions; + + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + private final ConsumerStats stats; + private final UnAckedMessageTracker unAckedMessageTracker; + private final ConsumerConfiguration internalConfig; + + TopicsConsumerImpl(PulsarClientImpl client, Collection<String> topics, String subscription, + ConsumerConfiguration conf, ExecutorService listenerExecutor, + CompletableFuture<Consumer> subscribeFuture) { + super(client, "TopicsConsumerFakeTopicName" + ConsumerName.generateRandomName(), subscription, + conf, Math.max(2, conf.getReceiverQueueSize()), listenerExecutor, + subscribeFuture); + + checkArgument(conf.getReceiverQueueSize() > 0, + "Receiver queue size needs to be greater than 0 for Topics Consumer"); + + this.topics = new ConcurrentHashMap<>(); + this.consumers = new ConcurrentHashMap<>(); + this.pausedConsumers = new ConcurrentLinkedQueue<>(); + this.sharedQueueResumeThreshold = maxReceiverQueueSize / 2; + this.numberTopicPartitions = new AtomicInteger(0); + + if (conf.getAckTimeoutMillis() != 0) { + this.unAckedMessageTracker = new UnAckedMessageTracker(client, this, conf.getAckTimeoutMillis()); + } else { + this.unAckedMessageTracker = UnAckedMessageTracker.UNACKED_MESSAGE_TRACKER_DISABLED; + } + + this.internalConfig = getInternalConsumerConfig(); + this.stats = client.getConfiguration().getStatsIntervalSeconds() > 0 ? new ConsumerStats() : null; + + if (topics.isEmpty()) { + this.namespaceName = null; + setState(State.Ready); + // We have successfully created N consumers, so we can start receiving messages now + subscribeFuture().complete(TopicsConsumerImpl.this); + return; + } + + checkArgument(topics.isEmpty() || !topicsNameInvalid(topics), "Topics should have same namespace."); + this.namespaceName = DestinationName.get(topics.stream().findFirst().get()).getNamespaceObject(); + + List<CompletableFuture<Void>> futures = Lists.newArrayListWithExpectedSize(topics.size()); + topics.forEach(topic -> futures.add(subscribeAsync(topic))); + FutureUtil.waitForAll(futures) + .thenAccept(finalFuture -> { + try { + if (numberTopicPartitions.get() > maxReceiverQueueSize) { + setMaxReceiverQueueSize(numberTopicPartitions.get()); + } + setState(State.Ready); + // We have successfully created N consumers, so we can start receiving messages now + startReceivingMessages(consumers.values().stream().collect(Collectors.toList())); + subscribeFuture().complete(TopicsConsumerImpl.this); + log.info("[{}] [{}] Created topics consumer with {} sub-consumers", + topic, subscription, numberTopicPartitions.get()); + } catch (PulsarClientException e) { + log.warn("[{}] Failed startReceivingMessages while subscribe topics: {}", topic, e.getMessage()); + subscribeFuture.completeExceptionally(e); + }}) + .exceptionally(ex -> { + log.warn("[{}] Failed to subscribe topics: {}", topic, ex.getMessage()); + subscribeFuture.completeExceptionally(ex); + return null; + }); + } + + // Check topics are valid. + // - each topic is valid, + // - every topic has same namespace. + private static boolean topicsNameInvalid(Collection<String> topics) { + checkState(topics != null && topics.size() > 1, + "topics should should contain more than 1 topics"); + + final String namespace = DestinationName.get(topics.stream().findFirst().get()).getNamespace(); + + Optional<String> result = topics.stream() + .filter(topic -> { + boolean topicInvalid = !DestinationName.isValid(topic); + if (topicInvalid) { + return true; + } + + String newNamespace = DestinationName.get(topic).getNamespace(); + if (!namespace.equals(newNamespace)) { + return true; + } else { + return false; + } + }).findFirst(); + if (result.isPresent()) { + log.warn("[{}] Received invalid topic name. {}/{}", result.get()); + } + return result.isPresent(); + } + + private void startReceivingMessages(List<ConsumerImpl> newConsumers) throws PulsarClientException { + if (log.isDebugEnabled()) { + log.debug("[{}] startReceivingMessages for {} new consumers in topics consumer", + topic, newConsumers.size()); + } + if (getState() == State.Ready) { + newConsumers.forEach(consumer -> { + consumer.sendFlowPermitsToBroker(consumer.cnx(), conf.getReceiverQueueSize()); + receiveMessageFromConsumer(consumer); + }); + } + } + + private void receiveMessageFromConsumer(ConsumerImpl consumer) { + consumer.receiveAsync().thenAccept(message -> { + if (log.isDebugEnabled()) { + log.debug("[{}] [{}] Receive message from sub consumer:{}", + topic, subscription, consumer.getTopic()); + } + // Process the message, add to the queue and trigger listener or async callback + messageReceived(consumer, message); + + // we're modifying pausedConsumers + lock.writeLock().lock(); + try { + int size = incomingMessages.size(); + if (size >= maxReceiverQueueSize + || (size > sharedQueueResumeThreshold && !pausedConsumers.isEmpty())) { + // mark this consumer to be resumed later: if No more space left in shared queue, + // or if any consumer is already paused (to create fair chance for already paused consumers) + pausedConsumers.add(consumer); + } else { + // Schedule next receiveAsync() if the incoming queue is not full. Use a different thread to avoid + // recursion and stack overflow + client.eventLoopGroup().execute(() -> { + receiveMessageFromConsumer(consumer); + }); + } + } finally { + lock.writeLock().unlock(); + } + }); + } + + private void messageReceived(ConsumerImpl consumer, Message message) { + checkArgument(message instanceof MessageImpl); + lock.writeLock().lock(); + try { + TopicMessageImpl topicMessage = new TopicMessageImpl(consumer.getTopic(), message); + unAckedMessageTracker.add(topicMessage.getMessageId()); + + if (log.isDebugEnabled()) { + log.debug("[{}][{}] Received message from topics-consumer {}", + topic, subscription, message.getMessageId()); + } + + // if asyncReceive is waiting : return message to callback without adding to incomingMessages queue + if (!pendingReceives.isEmpty()) { + CompletableFuture<Message> receivedFuture = pendingReceives.poll(); + listenerExecutor.execute(() -> receivedFuture.complete(topicMessage)); + } else { + // Enqueue the message so that it can be retrieved when application calls receive() + // Waits for the queue to have space for the message + // This should never block cause PartitonedConsumerImpl should always use GrowableArrayBlockingQueue + incomingMessages.put(topicMessage); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + lock.writeLock().unlock(); + } + + if (listener != null) { + // Trigger the notification on the message listener in a separate thread to avoid blocking the networking + // thread while the message processing happens + listenerExecutor.execute(() -> { + Message msg; + try { + msg = internalReceive(); + } catch (PulsarClientException e) { + log.warn("[{}] [{}] Failed to dequeue the message for listener", topic, subscription, e); + return; + } + + try { + if (log.isDebugEnabled()) { + log.debug("[{}][{}] Calling message listener for message {}", + topic, subscription, message.getMessageId()); + } + listener.received(TopicsConsumerImpl.this, msg); + } catch (Throwable t) { + log.error("[{}][{}] Message listener error in processing message: {}", + topic, subscription, message, t); + } + }); + } + } + + private void resumeReceivingFromPausedConsumersIfNeeded() { + lock.readLock().lock(); + try { + if (incomingMessages.size() <= sharedQueueResumeThreshold && !pausedConsumers.isEmpty()) { + while (true) { + ConsumerImpl consumer = pausedConsumers.poll(); + if (consumer == null) { + break; + } + + // if messages are readily available on consumer we will attempt to writeLock on the same thread + client.eventLoopGroup().execute(() -> { + receiveMessageFromConsumer(consumer); + }); + } + } + } finally { + lock.readLock().unlock(); + } + } + + @Override + protected Message internalReceive() throws PulsarClientException { + Message message; + try { + message = incomingMessages.take(); + checkState(message instanceof TopicMessageImpl); + unAckedMessageTracker.add(message.getMessageId()); + resumeReceivingFromPausedConsumersIfNeeded(); + return message; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarClientException(e); + } + } + + @Override + protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClientException { + Message message; + try { + message = incomingMessages.poll(timeout, unit); + if (message != null) { + checkArgument(message instanceof TopicMessageImpl); + unAckedMessageTracker.add(message.getMessageId()); + } + resumeReceivingFromPausedConsumersIfNeeded(); + return message; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new PulsarClientException(e); + } + } + + @Override + protected CompletableFuture<Message> internalReceiveAsync() { + CompletableFuture<Message> result = new CompletableFuture<>(); + Message message; + try { + lock.writeLock().lock(); + message = incomingMessages.poll(0, TimeUnit.SECONDS); + if (message == null) { + pendingReceives.add(result); + } else { + checkState(message instanceof TopicMessageImpl); + unAckedMessageTracker.add(message.getMessageId()); + resumeReceivingFromPausedConsumersIfNeeded(); + result.complete(message); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + result.completeExceptionally(new PulsarClientException(e)); + } finally { + lock.writeLock().unlock(); + } + + return result; + } + + @Override + protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType, + Map<String,Long> properties) { + checkArgument(messageId instanceof TopicMessageIdImpl); + + if (getState() != State.Ready) { + return FutureUtil.failedFuture(new PulsarClientException("Consumer already closed")); + } + + if (ackType == AckType.Cumulative) { + return FutureUtil.failedFuture(new PulsarClientException.NotSupportedException( + "Cumulative acknowledge not supported for topics consumer")); + } else { + ConsumerImpl consumer = consumers.get(messageId.getTopicName()); + + MessageId innerId = ((TopicMessageIdImpl)messageId).getInnerMessageId(); + return consumer.doAcknowledge(innerId, ackType, properties) + .thenRun(() -> + unAckedMessageTracker.remove(messageId)); + } + } + + @Override + public CompletableFuture<Void> unsubscribeAsync() { + if (getState() == State.Closing || getState() == State.Closed) { + return FutureUtil.failedFuture( + new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed")); + } + setState(State.Closing); + + AtomicReference<Throwable> unsubscribeFail = new AtomicReference<Throwable>(); + AtomicInteger completed = new AtomicInteger(numberTopicPartitions.get()); + CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>(); + + consumers.values().stream().forEach(consumer -> { + if (consumer != null) { + consumer.unsubscribeAsync().handle((unsubscribed, ex) -> { + if (ex != null) { + unsubscribeFail.compareAndSet(null, ex); + } + if (completed.decrementAndGet() == 0) { + if (unsubscribeFail.get() == null) { + setState(State.Closed); + unAckedMessageTracker.close(); + unsubscribeFuture.complete(null); + log.info("[{}] [{}] [{}] Unsubscribed Topics Consumer", + topic, subscription, consumerName); + } else { + setState(State.Failed); + unsubscribeFuture.completeExceptionally(unsubscribeFail.get()); + log.error("[{}] [{}] [{}] Could not unsubscribe Topics Consumer", + topic, subscription, consumerName, unsubscribeFail.get().getCause()); + } + } + + return null; + }); + } + }); + + return unsubscribeFuture; + } + + @Override + public CompletableFuture<Void> closeAsync() { + if (getState() == State.Closing || getState() == State.Closed) { + unAckedMessageTracker.close(); + return CompletableFuture.completedFuture(null); + } + setState(State.Closing); + + AtomicReference<Throwable> closeFail = new AtomicReference<Throwable>(); + AtomicInteger completed = new AtomicInteger(numberTopicPartitions.get()); + CompletableFuture<Void> closeFuture = new CompletableFuture<>(); + + consumers.values().stream().forEach(consumer -> { Review comment: Thanks. will change it. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
