[
https://issues.apache.org/jira/browse/CAMEL-12111?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16314523#comment-16314523
]
ASF GitHub Bot commented on CAMEL-12111:
----------------------------------------
davsclaus closed pull request #2164: CAMEL-12111: Fix reconnect if broker is
down on startup. Also fix so…
URL: https://github.com/apache/camel/pull/2164
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
index e96367c6e2f..ffef62d003c 100644
---
a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
+++
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java
@@ -295,10 +295,10 @@ public void reconnect() throws IOException,
TimeoutException {
if (isChannelOpen()) {
// The connection is good, so nothing to do
return;
- } else if (!isChannelOpen() &&
this.consumer.getEndpoint().getAutomaticRecoveryEnabled()) {
+ } else if (channel != null && !channel.isOpen() &&
isAutomaticRecoveryEnabled()) {
// Still need to wait for channel to re-open
throw new IOException("Waiting for channel to re-open.");
- } else if (!this.consumer.getEndpoint().getAutomaticRecoveryEnabled())
{
+ } else if (channel == null || !isAutomaticRecoveryEnabled()) {
log.info("Attempting to open a new rabbitMQ channel");
Connection conn = consumer.getConnection();
channel = openChannel(conn);
@@ -307,6 +307,11 @@ public void reconnect() throws IOException,
TimeoutException {
}
}
+ private boolean isAutomaticRecoveryEnabled() {
+ return this.consumer.getEndpoint().getAutomaticRecoveryEnabled() !=
null
+ && this.consumer.getEndpoint().getAutomaticRecoveryEnabled();
+ }
+
private boolean isChannelOpen() {
return channel != null && channel.isOpen();
}
diff --git
a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
index 95a660925ee..c0aaa6d5d03 100644
---
a/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
+++
b/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitMQConsumer.java
@@ -74,7 +74,7 @@ protected synchronized Connection getConnection() throws
IOException, TimeoutExc
if (this.conn == null) {
openConnection();
return this.conn;
- } else if (!this.conn.isOpen() &&
this.endpoint.getAutomaticRecoveryEnabled()) {
+ } else if (this.conn.isOpen() || (!this.conn.isOpen() &&
isAutomaticRecoveryEnabled())) {
return this.conn;
} else {
log.debug("The existing connection is closed");
@@ -83,16 +83,24 @@ protected synchronized Connection getConnection() throws
IOException, TimeoutExc
}
}
-
+ private boolean isAutomaticRecoveryEnabled() {
+ return this.endpoint.getAutomaticRecoveryEnabled() != null
+ && this.endpoint.getAutomaticRecoveryEnabled();
+ }
/**
- * Add a consumer thread for given channel
+ * Create the consumers but don't start yet
*/
- private void startConsumers() throws IOException {
-
+ private void createConsumers() throws IOException {
// Create consumers but don't start yet
for (int i = 0; i < endpoint.getConcurrentConsumers(); i++) {
createConsumer();
}
+ }
+
+ /**
+ * Start the consumers (already created)
+ */
+ private void startConsumers() {
// Try starting consumers (which will fail if RabbitMQ can't connect)
try {
@@ -160,6 +168,7 @@ protected void doResume() throws Exception {
protected void doStart() throws Exception {
executor = endpoint.createExecutor();
log.debug("Using executor {}", executor);
+ createConsumers();
startConsumers();
}
@@ -211,9 +220,6 @@ public Void call() throws Exception {
Thread.sleep(connectionRetryInterval);
}
}
- if (!connectionFailed) {
- startConsumers();
- }
stop();
return null;
}
----------------------------------------------------------------
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:
[email protected]
> Reconnect doesn't work if camel is started with rabbit broker initially
> inaccessible and automaticRecoveryEnabled=true or not set
> ---------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-12111
> URL: https://issues.apache.org/jira/browse/CAMEL-12111
> Project: Camel
> Issue Type: Bug
> Components: camel-rabbitmq
> Affects Versions: 2.20.1
> Reporter: Jeremy M Isikoff
> Fix For: 2.20.2, 2.21.0
>
>
> Starting a camel rabbit consumer with the rabbitmq broker down and then
> bringing the broker up used to work before 2.20.1 with
> automaticRecoveryEnabled=true.
> The logic added in line 298 of the RabbitConsumer and line 301 now only will
> recover in the case if automaticRecoveryEnabled=false by allowing it through
> to line 303 and opening a connection for the FIRST TIME.
> In addition line 298 null pointers on '&& null' if automaticRecoveryEnabled
> wasn't specified at all.
> https://github.com/apache/camel/blob/bad9ed4bb2ce1e258039d96ac80c71a746b0520d/components/camel-rabbitmq/src/main/java/org/apache/camel/component/rabbitmq/RabbitConsumer.java#L298
> As of now you can not have automaticRecoveryEnabled either not set or true
> and handle have camel work in the startup with broker down and then up use
> case.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
