| /* |
| * 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.rocketmq.client.java.impl.producer; |
| |
| import static com.google.common.base.Preconditions.checkNotNull; |
| |
| import apache.rocketmq.v2.ClientType; |
| import apache.rocketmq.v2.Code; |
| import apache.rocketmq.v2.EndTransactionRequest; |
| import apache.rocketmq.v2.EndTransactionResponse; |
| import apache.rocketmq.v2.HeartbeatRequest; |
| import apache.rocketmq.v2.NotifyClientTerminationRequest; |
| import apache.rocketmq.v2.RecoverOrphanedTransactionCommand; |
| import apache.rocketmq.v2.SendMessageRequest; |
| import apache.rocketmq.v2.SendMessageResponse; |
| import apache.rocketmq.v2.Status; |
| import com.google.common.math.IntMath; |
| import com.google.common.util.concurrent.FutureCallback; |
| import com.google.common.util.concurrent.Futures; |
| import com.google.common.util.concurrent.ListenableFuture; |
| import com.google.common.util.concurrent.ListeningExecutorService; |
| import com.google.common.util.concurrent.MoreExecutors; |
| import com.google.common.util.concurrent.SettableFuture; |
| import java.time.Duration; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.TimeUnit; |
| import java.util.function.Function; |
| import java.util.stream.Collectors; |
| import net.javacrumbs.futureconverter.java8guava.FutureConverter; |
| import org.apache.rocketmq.client.apis.ClientConfiguration; |
| import org.apache.rocketmq.client.apis.ClientException; |
| import org.apache.rocketmq.client.apis.message.Message; |
| import org.apache.rocketmq.client.apis.message.MessageId; |
| import org.apache.rocketmq.client.apis.producer.Producer; |
| import org.apache.rocketmq.client.apis.producer.SendReceipt; |
| import org.apache.rocketmq.client.apis.producer.Transaction; |
| import org.apache.rocketmq.client.apis.producer.TransactionChecker; |
| import org.apache.rocketmq.client.apis.producer.TransactionResolution; |
| import org.apache.rocketmq.client.java.exception.InternalErrorException; |
| import org.apache.rocketmq.client.java.exception.TooManyRequestsException; |
| import org.apache.rocketmq.client.java.hook.MessageHookPoints; |
| import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus; |
| import org.apache.rocketmq.client.java.hook.MessageInterceptorContextImpl; |
| import org.apache.rocketmq.client.java.impl.ClientImpl; |
| import org.apache.rocketmq.client.java.impl.Settings; |
| import org.apache.rocketmq.client.java.message.GeneralMessage; |
| import org.apache.rocketmq.client.java.message.GeneralMessageImpl; |
| import org.apache.rocketmq.client.java.message.MessageType; |
| import org.apache.rocketmq.client.java.message.MessageViewImpl; |
| import org.apache.rocketmq.client.java.message.PublishingMessageImpl; |
| import org.apache.rocketmq.client.java.retry.ExponentialBackoffRetryPolicy; |
| import org.apache.rocketmq.client.java.retry.RetryPolicy; |
| import org.apache.rocketmq.client.java.route.Endpoints; |
| import org.apache.rocketmq.client.java.route.MessageQueueImpl; |
| import org.apache.rocketmq.client.java.route.TopicRouteData; |
| import org.apache.rocketmq.client.java.rpc.RpcFuture; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Default implementation of {@link Producer} |
| * |
| * @see Producer |
| */ |
| @SuppressWarnings({"UnstableApiUsage", "NullableProblems"}) |
| class ProducerImpl extends ClientImpl implements Producer { |
| private static final Logger log = LoggerFactory.getLogger(ProducerImpl.class); |
| |
| protected final PublishingSettings publishingSettings; |
| final ConcurrentMap<String/* topic */, PublishingLoadBalancer> publishingRouteDataCache; |
| private final TransactionChecker checker; |
| |
| /** |
| * The caller is supposed to have validated the arguments and handled throwing exception or |
| * logging warnings already, so we avoid repeating args check here. |
| */ |
| ProducerImpl(ClientConfiguration clientConfiguration, Set<String> topics, int maxAttempts, |
| TransactionChecker checker) { |
| super(clientConfiguration, topics); |
| ExponentialBackoffRetryPolicy retryPolicy = ExponentialBackoffRetryPolicy.immediatelyRetryPolicy(maxAttempts); |
| this.publishingSettings = new PublishingSettings(clientConfiguration.getNamespace(), clientId, endpoints, |
| retryPolicy, clientConfiguration.getRequestTimeout(), topics); |
| this.checker = checker; |
| this.publishingRouteDataCache = new ConcurrentHashMap<>(); |
| } |
| |
| @Override |
| protected void startUp() throws Exception { |
| try { |
| log.info("Begin to start the rocketmq producer, clientId={}", clientId); |
| super.startUp(); |
| log.info("The rocketmq producer starts successfully, clientId={}", clientId); |
| } catch (Throwable t) { |
| log.error("Failed to start the rocketmq producer, try to shutdown it, clientId={}", clientId, t); |
| shutDown(); |
| throw t; |
| } |
| } |
| |
| @Override |
| protected void shutDown() throws InterruptedException { |
| log.info("Begin to shutdown the rocketmq producer, clientId={}", clientId); |
| super.shutDown(); |
| log.info("Shutdown the rocketmq producer successfully, clientId={}", clientId); |
| } |
| |
| @Override |
| public void onRecoverOrphanedTransactionCommand(Endpoints endpoints, RecoverOrphanedTransactionCommand command) { |
| final String transactionId = command.getTransactionId(); |
| final String messageId = command.getMessage().getSystemProperties().getMessageId(); |
| if (null == checker) { |
| log.error("No transaction checker registered, ignore it, messageId={}, transactionId={}, endpoints={}," |
| + " clientId={}", messageId, transactionId, endpoints, clientId); |
| return; |
| } |
| MessageViewImpl messageView; |
| try { |
| messageView = MessageViewImpl.fromProtobuf(command.getMessage()); |
| } catch (Throwable t) { |
| log.error("[Bug] Failed to decode message during orphaned transaction message recovery, messageId={}, " |
| + "transactionId={}, endpoints={}, clientId={}", messageId, transactionId, endpoints, clientId, t); |
| return; |
| } |
| ListenableFuture<TransactionResolution> future; |
| try { |
| final ListeningExecutorService service = MoreExecutors.listeningDecorator(telemetryCommandExecutor); |
| final Callable<TransactionResolution> task = () -> checker.check(messageView); |
| future = service.submit(task); |
| } catch (Throwable t) { |
| final SettableFuture<TransactionResolution> future0 = SettableFuture.create(); |
| future0.setException(t); |
| future = future0; |
| } |
| Futures.addCallback(future, new FutureCallback<TransactionResolution>() { |
| @Override |
| public void onSuccess(TransactionResolution resolution) { |
| try { |
| if (null == resolution || TransactionResolution.UNKNOWN.equals(resolution)) { |
| return; |
| } |
| final GeneralMessage generalMessage = new GeneralMessageImpl(messageView); |
| endTransaction(endpoints, generalMessage, messageView.getMessageId(), |
| transactionId, resolution); |
| } catch (Throwable t) { |
| log.error("Exception raised while ending the transaction, messageId={}, transactionId={}, " |
| + "endpoints={}, clientId={}", messageId, transactionId, endpoints, clientId, t); |
| } |
| } |
| |
| @Override |
| public void onFailure(Throwable t) { |
| log.error("Exception raised while checking the transaction, messageId={}, transactionId={}, " |
| + "endpoints={}, clientId={}", messageId, transactionId, endpoints, clientId, t); |
| |
| } |
| }, MoreExecutors.directExecutor()); |
| } |
| |
| @Override |
| public Settings getSettings() { |
| return publishingSettings; |
| } |
| |
| @Override |
| public NotifyClientTerminationRequest wrapNotifyClientTerminationRequest() { |
| return NotifyClientTerminationRequest.newBuilder().build(); |
| } |
| |
| @Override |
| public HeartbeatRequest wrapHeartbeatRequest() { |
| return HeartbeatRequest.newBuilder().setClientType(ClientType.PRODUCER).build(); |
| } |
| |
| /** |
| * @see Producer#send(Message) |
| */ |
| @Override |
| public SendReceipt send(Message message) throws ClientException { |
| final ListenableFuture<SendReceipt> future = Futures.transform(send(Collections.singletonList(message), false), |
| sendReceipts -> sendReceipts.iterator().next(), MoreExecutors.directExecutor()); |
| return handleClientFuture(future); |
| } |
| |
| /** |
| * @see Producer#send(Message, Transaction) |
| */ |
| @Override |
| public SendReceipt send(Message message, Transaction transaction) throws ClientException { |
| if (!(transaction instanceof TransactionImpl)) { |
| throw new IllegalArgumentException("Failed downcasting for transaction"); |
| } |
| TransactionImpl transactionImpl = (TransactionImpl) transaction; |
| final PublishingMessageImpl publishingMessage; |
| try { |
| publishingMessage = transactionImpl.tryAddMessage(message); |
| } catch (Throwable t) { |
| throw new ClientException(t); |
| } |
| final ListenableFuture<List<SendReceiptImpl>> future = send(Collections.singletonList(publishingMessage), true); |
| final List<SendReceiptImpl> receipts = handleClientFuture(future); |
| final SendReceiptImpl sendReceipt = receipts.iterator().next(); |
| ((TransactionImpl) transaction).tryAddReceipt(publishingMessage, sendReceipt); |
| return sendReceipt; |
| } |
| |
| /** |
| * @see Producer#sendAsync(Message) |
| */ |
| @Override |
| public CompletableFuture<SendReceipt> sendAsync(Message message) { |
| final ListenableFuture<SendReceipt> future = Futures.transform(send(Collections.singletonList(message), false), |
| sendReceipts -> sendReceipts.iterator().next(), MoreExecutors.directExecutor()); |
| return FutureConverter.toCompletableFuture(future); |
| } |
| |
| /** |
| * @see Producer#beginTransaction() |
| */ |
| @Override |
| public Transaction beginTransaction() { |
| checkNotNull(checker, "Transaction checker should not be null"); |
| if (!this.isRunning()) { |
| log.error("Unable to begin a transaction because producer is not running, state={}, clientId={}", |
| this.state(), clientId); |
| throw new IllegalStateException("Producer is not running now"); |
| } |
| return new TransactionImpl(this); |
| } |
| |
| @Override |
| public void close() { |
| this.stopAsync().awaitTerminated(); |
| } |
| |
| public void endTransaction(Endpoints endpoints, GeneralMessage generalMessage, MessageId messageId, |
| String transactionId, final TransactionResolution resolution) throws ClientException { |
| final EndTransactionRequest.Builder builder = EndTransactionRequest.newBuilder() |
| .setMessageId(messageId.toString()).setTransactionId(transactionId) |
| .setTopic(apache.rocketmq.v2.Resource.newBuilder() |
| .setResourceNamespace(clientConfiguration.getNamespace()) |
| .setName(generalMessage.getTopic()) |
| .build()); |
| switch (resolution) { |
| case COMMIT: |
| builder.setResolution(apache.rocketmq.v2.TransactionResolution.COMMIT); |
| break; |
| case ROLLBACK: |
| default: |
| builder.setResolution(apache.rocketmq.v2.TransactionResolution.ROLLBACK); |
| } |
| final Duration requestTimeout = clientConfiguration.getRequestTimeout(); |
| final EndTransactionRequest request = builder.build(); |
| final List<GeneralMessage> generalMessages = Collections.singletonList(generalMessage); |
| MessageHookPoints messageHookPoints = TransactionResolution.COMMIT.equals(resolution) ? |
| MessageHookPoints.COMMIT_TRANSACTION : MessageHookPoints.ROLLBACK_TRANSACTION; |
| final MessageInterceptorContextImpl context = new MessageInterceptorContextImpl(messageHookPoints); |
| doBefore(context, generalMessages); |
| |
| final RpcFuture<EndTransactionRequest, EndTransactionResponse> future = |
| this.getClientManager().endTransaction(endpoints, request, requestTimeout); |
| Futures.addCallback(future, new FutureCallback<EndTransactionResponse>() { |
| @Override |
| public void onSuccess(EndTransactionResponse response) { |
| final Status status = response.getStatus(); |
| final Code code = status.getCode(); |
| MessageHookPointsStatus hookPointsStatus = Code.OK.equals(code) ? MessageHookPointsStatus.OK : |
| MessageHookPointsStatus.ERROR; |
| final MessageInterceptorContextImpl context0 = new MessageInterceptorContextImpl(context, |
| hookPointsStatus); |
| doAfter(context0, generalMessages); |
| } |
| |
| @Override |
| public void onFailure(Throwable t) { |
| final MessageInterceptorContextImpl context0 = new MessageInterceptorContextImpl(context, |
| MessageHookPointsStatus.ERROR); |
| doAfter(context0, generalMessages); |
| } |
| }, MoreExecutors.directExecutor()); |
| final EndTransactionResponse response = handleClientFuture(future); |
| final Status status = response.getStatus(); |
| final Code code = status.getCode(); |
| if (!Code.OK.equals(code)) { |
| throw new ClientException(code.getNumber(), future.getContext().getRequestId(), status.getMessage()); |
| } |
| } |
| |
| /** |
| * Isolate specified {@link Endpoints}. |
| */ |
| private void isolate(Endpoints endpoints) { |
| isolated.add(endpoints); |
| } |
| |
| private RetryPolicy getRetryPolicy() { |
| return publishingSettings.getRetryPolicy(); |
| } |
| |
| /** |
| * Take message queue(s) from route for message publishing. |
| */ |
| private List<MessageQueueImpl> takeMessageQueues(PublishingLoadBalancer result) { |
| return result.takeMessageQueues(isolated, this.getRetryPolicy().getMaxAttempts()); |
| } |
| |
| private ListenableFuture<List<SendReceiptImpl>> send(List<Message> messages, boolean txEnabled) { |
| SettableFuture<List<SendReceiptImpl>> future = SettableFuture.create(); |
| |
| // Check producer state before message publishing. |
| if (!this.isRunning()) { |
| final IllegalStateException e = new IllegalStateException("Producer is not running now"); |
| future.setException(e); |
| log.error("Unable to send message because producer is not running, state={}, clientId={}", |
| this.state(), clientId); |
| return future; |
| } |
| |
| List<PublishingMessageImpl> pubMessages = new ArrayList<>(); |
| for (Message message : messages) { |
| try { |
| final PublishingMessageImpl pubMessage = new PublishingMessageImpl(message, publishingSettings, |
| txEnabled); |
| pubMessages.add(pubMessage); |
| } catch (Throwable t) { |
| // Failed to refine message, no need to proceed. |
| log.error("Failed to refine message to send, clientId={}, message={}", clientId, message, t); |
| future.setException(t); |
| return future; |
| } |
| } |
| |
| // Collect topics to send message. |
| final Set<String> topics = pubMessages.stream().map(Message::getTopic).collect(Collectors.toSet()); |
| if (1 < topics.size()) { |
| // Messages have different topics, no need to proceed. |
| final IllegalArgumentException e = new IllegalArgumentException("Messages to send have different topics"); |
| future.setException(e); |
| log.error("Messages to be sent have different topics, no need to proceed, topic(s)={}, clientId={}", |
| topics, clientId); |
| return future; |
| } |
| |
| final String topic = topics.iterator().next(); |
| // Collect message types. |
| final Set<MessageType> messageTypes = pubMessages.stream() |
| .map(PublishingMessageImpl::getMessageType) |
| .collect(Collectors.toSet()); |
| if (1 < messageTypes.size()) { |
| // Messages have different message type, no need to proceed. |
| final IllegalArgumentException e = new IllegalArgumentException("Messages to send have different types, " |
| + "please check"); |
| future.setException(e); |
| log.error("Messages to be sent have different message types, no need to proceed, topic={}, messageType" |
| + "(s)={}, clientId={}", topic, messageTypes, clientId, e); |
| return future; |
| } |
| |
| final MessageType messageType = messageTypes.iterator().next(); |
| final String messageGroup; |
| |
| // Message group must be same if message type is FIFO, or no need to proceed. |
| if (MessageType.FIFO.equals(messageType)) { |
| final Set<String> messageGroups = pubMessages.stream() |
| .map(PublishingMessageImpl::getMessageGroup).filter(Optional::isPresent) |
| .map(Optional::get).collect(Collectors.toSet()); |
| |
| if (1 < messageGroups.size()) { |
| final IllegalArgumentException e = new IllegalArgumentException("FIFO messages to send have different " |
| + "message groups, messageGroups=" + messageGroups); |
| future.setException(e); |
| log.error("FIFO messages to be sent have different message groups, no need to proceed, topic={}, " |
| + "messageGroups={}, clientId={}", topic, messageGroups, clientId, e); |
| return future; |
| } |
| messageGroup = messageGroups.iterator().next(); |
| } else { |
| messageGroup = null; |
| } |
| |
| this.topics.add(topic); |
| // Get publishing topic route. |
| final ListenableFuture<PublishingLoadBalancer> routeFuture = getPublishingLoadBalancer(topic); |
| return Futures.transformAsync(routeFuture, result -> { |
| // Prepare the candidate message queue(s) for retry-sending in advance. |
| final List<MessageQueueImpl> candidates = null == messageGroup ? takeMessageQueues(result) : |
| Collections.singletonList(result.takeMessageQueueByMessageGroup(messageGroup)); |
| final SettableFuture<List<SendReceiptImpl>> future0 = SettableFuture.create(); |
| send0(future0, topic, messageType, candidates, pubMessages, 1); |
| return future0; |
| }, MoreExecutors.directExecutor()); |
| } |
| |
| /** |
| * The caller is supposed to make sure different messages have the same message type and same topic. |
| */ |
| private SendMessageRequest wrapSendMessageRequest(List<PublishingMessageImpl> pubMessages, MessageQueueImpl mq) { |
| final List<apache.rocketmq.v2.Message> messages = pubMessages.stream() |
| .map(publishingMessage -> publishingMessage.toProtobuf(clientConfiguration.getNamespace(), mq)) |
| .collect(Collectors.toList()); |
| return SendMessageRequest.newBuilder().addAllMessages(messages).build(); |
| } |
| |
| ListenableFuture<List<SendReceiptImpl>> send0(Endpoints endpoints, List<PublishingMessageImpl> pubMessages, |
| MessageQueueImpl mq) { |
| final SendMessageRequest request = wrapSendMessageRequest(pubMessages, mq); |
| final RpcFuture<SendMessageRequest, SendMessageResponse> future0 = |
| this.getClientManager().sendMessage(endpoints, request, clientConfiguration.getRequestTimeout()); |
| return Futures.transformAsync(future0, |
| response -> Futures.immediateFuture(SendReceiptImpl.processResponseInvocation(mq, response, future0)), |
| MoreExecutors.directExecutor()); |
| } |
| |
| /** |
| * Warning: please DO NOT modify the signature of this method, it is used by OpenTelemetry instrumentation. |
| */ |
| private void send0(SettableFuture<List<SendReceiptImpl>> future0, String topic, MessageType messageType, |
| final List<MessageQueueImpl> candidates, final List<PublishingMessageImpl> messages, final int attempt) { |
| // Calculate the current message queue. |
| final MessageQueueImpl mq = candidates.get(IntMath.mod(attempt - 1, candidates.size())); |
| final List<MessageType> acceptMessageTypes = mq.getAcceptMessageTypes(); |
| if (publishingSettings.isValidateMessageType() && !acceptMessageTypes.contains(messageType)) { |
| final IllegalArgumentException e = new IllegalArgumentException("Current message type not match with " |
| + "topic accept message types, topic=" + topic + ", actualMessageType=" + messageType + ", " |
| + "acceptMessageTypes=" + acceptMessageTypes); |
| future0.setException(e); |
| return; |
| } |
| final Endpoints endpoints = mq.getBroker().getEndpoints(); |
| final ListenableFuture<List<SendReceiptImpl>> future = send0(endpoints, messages, mq); |
| final int maxAttempts = this.getRetryPolicy().getMaxAttempts(); |
| |
| // Intercept before message publishing. |
| final List<GeneralMessage> generalMessages = messages.stream().map((Function<PublishingMessageImpl, |
| GeneralMessage>) GeneralMessageImpl::new).collect(Collectors.toList()); |
| final MessageInterceptorContextImpl context = new MessageInterceptorContextImpl(MessageHookPoints.SEND); |
| doBefore(context, generalMessages); |
| |
| Futures.addCallback(future, new FutureCallback<List<SendReceiptImpl>>() { |
| @Override |
| public void onSuccess(List<SendReceiptImpl> sendReceipts) { |
| // Should never reach here. |
| if (sendReceipts.size() != messages.size()) { |
| final InternalErrorException e = new InternalErrorException("[Bug] due to an" |
| + " unknown reason from remote, received send receipt's quantity " + sendReceipts.size() |
| + " is not equal to sent message's quantity " + messages.size()); |
| future0.setException(e); |
| |
| // Intercept after message publishing. |
| final MessageInterceptorContextImpl context0 = new MessageInterceptorContextImpl(context, |
| MessageHookPointsStatus.ERROR); |
| doAfter(context0, generalMessages); |
| |
| return; |
| } |
| // Intercept after message publishing. |
| final MessageInterceptorContextImpl context0 = new MessageInterceptorContextImpl(context, |
| MessageHookPointsStatus.OK); |
| doAfter(context0, generalMessages); |
| |
| // No need more attempts. |
| future0.set(sendReceipts); |
| // Resend message(s) successfully. |
| if (1 < attempt) { |
| // Collect messageId(s) for logging. |
| List<MessageId> messageIds = new ArrayList<>(); |
| for (SendReceipt receipt : sendReceipts) { |
| messageIds.add(receipt.getMessageId()); |
| } |
| log.info("Resend message successfully, topic={}, messageId(s)={}, maxAttempts={}, " |
| + "attempt={}, endpoints={}, clientId={}", topic, messageIds, maxAttempts, attempt, |
| endpoints, clientId); |
| } |
| // Send message(s) successfully on first attempt, return directly. |
| } |
| |
| @Override |
| public void onFailure(Throwable t) { |
| // Intercept after message publishing. |
| final MessageInterceptorContextImpl context0 = new MessageInterceptorContextImpl(context, |
| MessageHookPointsStatus.ERROR); |
| doAfter(context0, generalMessages); |
| |
| // Collect messageId(s) for logging. |
| List<MessageId> messageIds = new ArrayList<>(); |
| for (PublishingMessageImpl message : messages) { |
| messageIds.add(message.getMessageId()); |
| } |
| // Isolate endpoints because of sending failure. |
| isolate(endpoints); |
| if (attempt >= maxAttempts) { |
| // No need more attempts. |
| future0.setException(t); |
| log.error("Failed to send message(s) finally, run out of attempt times, maxAttempts={}, " + |
| "attempt={}, topic={}, messageId(s)={}, endpoints={}, clientId={}", |
| maxAttempts, attempt, topic, messageIds, endpoints, clientId, t); |
| return; |
| } |
| // No need more attempts for transactional message. |
| if (MessageType.TRANSACTION.equals(messageType)) { |
| future0.setException(t); |
| log.error("Failed to send transactional message finally, maxAttempts=1, attempt={}, " + |
| "topic={}, messageId(s)={}, endpoints={}, clientId={}", attempt, topic, messageIds, |
| endpoints, clientId, t); |
| return; |
| } |
| // Try to do more attempts. |
| int nextAttempt = 1 + attempt; |
| // Retry immediately if the request is not throttled. |
| if (!(t instanceof TooManyRequestsException)) { |
| log.warn("Failed to send message, would attempt to resend right now, maxAttempts={}, " |
| + "attempt={}, topic={}, messageId(s)={}, endpoints={}, clientId={}", maxAttempts, attempt, |
| topic, messageIds, endpoints, clientId, t); |
| send0(future0, topic, messageType, candidates, messages, nextAttempt); |
| return; |
| } |
| final Duration delay = ProducerImpl.this.getRetryPolicy().getNextAttemptDelay(nextAttempt); |
| log.warn("Failed to send message due to too many requests, would attempt to resend after {}, " |
| + "maxAttempts={}, attempt={}, topic={}, messageId(s)={}, endpoints={}, clientId={}", delay, |
| maxAttempts, attempt, topic, messageIds, endpoints, clientId, t); |
| ProducerImpl.this.getClientManager().getScheduler().schedule(() -> send0(future0, topic, messageType, |
| candidates, messages, nextAttempt), delay.toNanos(), TimeUnit.NANOSECONDS); |
| } |
| }, clientCallbackExecutor); |
| } |
| |
| private PublishingLoadBalancer updatePublishingLoadBalancer(String topic, TopicRouteData topicRouteData) { |
| PublishingLoadBalancer publishingLoadBalancer = publishingRouteDataCache.get(topic); |
| publishingLoadBalancer = null == publishingLoadBalancer ? new PublishingLoadBalancer(topicRouteData) : |
| publishingLoadBalancer.update(topicRouteData); |
| publishingRouteDataCache.put(topic, publishingLoadBalancer); |
| return publishingLoadBalancer; |
| } |
| |
| @Override |
| public void onTopicRouteDataUpdate0(String topic, TopicRouteData topicRouteData) { |
| updatePublishingLoadBalancer(topic, topicRouteData); |
| } |
| |
| private ListenableFuture<PublishingLoadBalancer> getPublishingLoadBalancer(final String topic) { |
| final PublishingLoadBalancer loadBalancer = publishingRouteDataCache.get(topic); |
| if (null != loadBalancer) { |
| return Futures.immediateFuture(loadBalancer); |
| } |
| return Futures.transform(getRouteData(topic), topicRouteData -> updatePublishingLoadBalancer(topic, |
| topicRouteData), MoreExecutors.directExecutor()); |
| } |
| } |
