You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2020/03/23 05:15:57 UTC
[pulsar] branch master updated: Do not retry unretriable errors
(#6584)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ca7bb99 Do not retry unretriable errors (#6584)
ca7bb99 is described below
commit ca7bb998436a32f7f919567d4b62d42b994363f8
Author: Sanjeev Kulkarni <sa...@gmail.com>
AuthorDate: Sun Mar 22 22:15:44 2020 -0700
Do not retry unretriable errors (#6584)
* Do not retry unretriable errors
* Fix test
* Cleanup logic
Co-authored-by: Sanjeev Kulkarni <sa...@splunk.com>
---
.../org/apache/pulsar/client/api/PulsarClientException.java | 3 +--
.../java/org/apache/pulsar/client/impl/ConnectionHandler.java | 8 +++++++-
.../main/java/org/apache/pulsar/client/impl/ConsumerImpl.java | 10 ++++++++--
.../main/java/org/apache/pulsar/client/impl/ProducerImpl.java | 11 ++++++++---
.../java/org/apache/pulsar/client/impl/PulsarClientImpl.java | 2 +-
5 files changed, 25 insertions(+), 9 deletions(-)
diff --git a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
index 8703caa..b9ca6f7 100644
--- a/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
+++ b/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/PulsarClientException.java
@@ -721,8 +721,7 @@ public class PulsarClientException extends IOException {
}
public static boolean isRetriableError(Throwable t) {
- if (t instanceof TooManyRequestsException
- || t instanceof AuthorizationException
+ if (t instanceof AuthorizationException
|| t instanceof InvalidServiceURL
|| t instanceof InvalidConfigurationException
|| t instanceof NotFoundException
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
index 0df0aae..e035283 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
@@ -75,7 +75,13 @@ public class ConnectionHandler {
private Void handleConnectionError(Throwable exception) {
log.warn("[{}] [{}] Error connecting to broker: {}", state.topic, state.getHandlerName(), exception.getMessage());
- connection.connectionFailed(new PulsarClientException(exception));
+ if (exception instanceof PulsarClientException) {
+ connection.connectionFailed((PulsarClientException) exception);
+ } else if (exception.getCause() instanceof PulsarClientException) {
+ connection.connectionFailed((PulsarClientException)exception.getCause());
+ } else {
+ connection.connectionFailed(new PulsarClientException(exception));
+ }
State state = this.state.getState();
if (state == State.Uninitialized || state == State.Connecting || state == State.Ready) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 5312675..ea52568 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -719,9 +719,15 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
@Override
public void connectionFailed(PulsarClientException exception) {
- if (System.currentTimeMillis() > subscribeTimeout && subscribeFuture.completeExceptionally(exception)) {
+ boolean nonRetriableError = !PulsarClientException.isRetriableError(exception);
+ boolean timeout = System.currentTimeMillis() > subscribeTimeout;
+ if ((nonRetriableError || timeout) && subscribeFuture.completeExceptionally(exception)) {
setState(State.Failed);
- log.info("[{}] Consumer creation failed for consumer {}", topic, consumerId);
+ if (nonRetriableError) {
+ log.info("[{}] Consumer creation failed for consumer {} with unretriableError {}", topic, consumerId, exception);
+ } else {
+ log.info("[{}] Consumer creation failed for consumer {} after timeout", topic, consumerId);
+ }
client.cleanupConsumer(this);
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 989a284..5b8996c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -1195,9 +1195,14 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
@Override
public void connectionFailed(PulsarClientException exception) {
- if (System.currentTimeMillis() > createProducerTimeout
- && producerCreatedFuture.completeExceptionally(exception)) {
- log.info("[{}] Producer creation failed for producer {}", topic, producerId);
+ boolean nonRetriableError = !PulsarClientException.isRetriableError(exception);
+ boolean producerTimeout = System.currentTimeMillis() > createProducerTimeout;
+ if ((nonRetriableError || producerTimeout) && producerCreatedFuture.completeExceptionally(exception)) {
+ if (nonRetriableError) {
+ log.info("[{}] Producer creation failed for producer {} with unretriableError = {}", topic, producerId, exception);
+ } else {
+ log.info("[{}] Producer creation failed for producer {} after producerTimeout", topic, producerId);
+ }
setState(State.Failed);
client.cleanupProducer(this);
}
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 31a4ed0..91ca682 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
@@ -679,7 +679,7 @@ public class PulsarClientImpl implements PulsarClient {
lookup.getPartitionedTopicMetadata(topicName).thenAccept(future::complete).exceptionally(e -> {
long nextDelay = Math.min(backoff.next(), remainingTime.get());
// skip retry scheduler when set lookup throttle in client or server side which will lead to `TooManyRequestsException`
- boolean isLookupThrottling = !PulsarClientException.isRetriableError(e.getCause());
+ boolean isLookupThrottling = !PulsarClientException.isRetriableError(e.getCause()) || e.getCause() instanceof PulsarClientException.TooManyRequestsException;
if (nextDelay <= 0 || isLookupThrottling) {
future.completeExceptionally(e);
return null;