You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by lb...@apache.org on 2016/04/27 12:48:31 UTC
[1/2] camel git commit: CAMEL-9834 : WatchConsumer does not properly
set watchIndex
Repository: camel
Updated Branches:
refs/heads/camel-2.17.x e09eef591 -> 45317409d
CAMEL-9834 : WatchConsumer does not properly set watchIndex
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/ed08a7a3
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/ed08a7a3
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/ed08a7a3
Branch: refs/heads/camel-2.17.x
Commit: ed08a7a34ade56cb177a9e5731a9926e95d6cc0c
Parents: e09eef5
Author: lburgazzoli <lb...@gmail.com>
Authored: Thu Apr 7 14:14:27 2016 +0200
Committer: lburgazzoli <lb...@gmail.com>
Committed: Wed Apr 27 12:45:26 2016 +0200
----------------------------------------------------------------------
components/camel-etcd/pom.xml | 37 +++++++
.../camel/component/etcd/EtcdComponent.java | 25 +----
.../camel/component/etcd/EtcdConfiguration.java | 12 +++
.../camel/component/etcd/EtcdWatchConsumer.java | 104 ++++++++++++++-----
.../camel/component/etcd/EtcdKeysTest.java | 3 +-
.../camel/component/etcd/EtcdStatsTest.java | 2 +-
.../camel/component/etcd/EtcdWatchTest.java | 74 ++++++++-----
parent/pom.xml | 2 +-
8 files changed, 181 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-etcd/pom.xml b/components/camel-etcd/pom.xml
index 3e40304..f7a75ee 100644
--- a/components/camel-etcd/pom.xml
+++ b/components/camel-etcd/pom.xml
@@ -85,4 +85,41 @@
</dependency>
</dependencies>
+
+
+ <profiles>
+ <profile>
+ <id>etcd-skip-tests</id>
+ <activation>
+ <activeByDefault>true</activeByDefault>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>true</skipTests>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ <profile>
+ <id>etcd-tests</id>
+ <activation>
+ <activeByDefault>false</activeByDefault>
+ </activation>
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <skipTests>false</skipTests>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+
+ </profiles>
</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdComponent.java b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdComponent.java
index eab5d35..3b81579 100644
--- a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdComponent.java
+++ b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdComponent.java
@@ -49,6 +49,7 @@ public class EtcdComponent extends UriEndpointComponent {
}
EtcdNamespace namespace = getCamelContext().getTypeConverter().mandatoryConvertTo(EtcdNamespace.class, ns);
+ EtcdConfiguration configuration = loadConfiguration(new EtcdConfiguration(), parameters);
if (namespace != null) {
// path must start with leading slash
@@ -58,29 +59,11 @@ public class EtcdComponent extends UriEndpointComponent {
switch (namespace) {
case stats:
- return new EtcdStatsEndpoint(
- uri,
- this,
- loadConfiguration(new EtcdConfiguration(), parameters),
- namespace,
- path
- );
+ return new EtcdStatsEndpoint(uri, this, configuration, namespace, path);
case watch:
- return new EtcdWatchEndpoint(
- uri,
- this,
- loadConfiguration(new EtcdConfiguration(), parameters),
- namespace,
- path
- );
+ return new EtcdWatchEndpoint(uri, this, configuration, namespace, path);
case keys:
- return new EtcdKeysEndpoint(
- uri,
- this,
- loadConfiguration(new EtcdConfiguration(), parameters),
- namespace,
- path
- );
+ return new EtcdKeysEndpoint(uri, this, configuration, namespace, path);
default:
throw new IllegalStateException("No endpoint for " + remaining);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdConfiguration.java b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdConfiguration.java
index 9fe36cd..45d50f7 100644
--- a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdConfiguration.java
+++ b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdConfiguration.java
@@ -39,6 +39,8 @@ public class EtcdConfiguration {
private Integer timeToLive;
@UriParam
private Long timeout;
+ @UriParam(label = "consumer,advance", defaultValue = "0")
+ private Long fromIndex = 0L;
public String getUris() {
return uris;
@@ -128,4 +130,14 @@ public class EtcdConfiguration {
this.timeout = timeout;
}
+ public Long getFromIndex() {
+ return fromIndex;
+ }
+
+ /**
+ * The index to watch from
+ */
+ public void setFromIndex(Long fromIndex) {
+ this.fromIndex = fromIndex;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
index 4c3e6fa..9f85a0a 100644
--- a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
+++ b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
@@ -16,12 +16,14 @@
*/
package org.apache.camel.component.etcd;
-import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
import mousio.client.promises.ResponsePromise;
import mousio.etcd4j.requests.EtcdKeyGetRequest;
+import mousio.etcd4j.responses.EtcdErrorCode;
+import mousio.etcd4j.responses.EtcdException;
import mousio.etcd4j.responses.EtcdKeysResponse;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
@@ -30,15 +32,18 @@ import org.slf4j.LoggerFactory;
public class EtcdWatchConsumer extends AbstractEtcdConsumer implements ResponsePromise.IsSimplePromiseResponseHandler<EtcdKeysResponse> {
private static final Logger LOGGER = LoggerFactory.getLogger(EtcdWatchConsumer.class);
+ private static final String OUTDATED_EVENT_MSG = "requested index is outdated and cleared";
private final EtcdWatchEndpoint endpoint;
private final EtcdConfiguration configuration;
+ private final AtomicLong index;
public EtcdWatchConsumer(EtcdWatchEndpoint endpoint, Processor processor, EtcdConfiguration configuration, EtcdNamespace namespace, String path) {
super(endpoint, processor, configuration, namespace, path);
this.endpoint = endpoint;
this.configuration = configuration;
+ this.index = new AtomicLong(configuration.getFromIndex());
}
@Override
@@ -59,35 +64,78 @@ public class EtcdWatchConsumer extends AbstractEtcdConsumer implements ResponseP
return;
}
- try {
- EtcdKeysResponse response = promise.get();
-
- Exchange exchange = endpoint.createExchange();
- exchange.getIn().setHeader(EtcdConstants.ETCD_NAMESPACE, getNamespace());
- exchange.getIn().setHeader(EtcdConstants.ETCD_PATH, response.node.key);
- exchange.getIn().setBody(response);
-
- getProcessor().process(exchange);
-
- watch();
- } catch (TimeoutException e) {
- LOGGER.debug("Timeout watching for {}", getPath());
-
- if (configuration.isSendEmptyExchangeOnTimeout()) {
- Exchange exchange = endpoint.createExchange();
- try {
+ Exchange exchange = null;
+ Throwable throwable = promise.getException();
+
+ if (throwable != null && throwable instanceof EtcdException) {
+ EtcdException exception = (EtcdException) throwable;
+ // Etcd only keeps the responses of the most recent 1000 events
+ // across all etcd keys so if we wait for a cleared index, we
+ // get "index is outdated response" like:
+ //
+ // {
+ // "errorCode" : 401,
+ // "message" : "The event in requested index is outdated and cleared",
+ // "cause" : "the requested history has been cleared [1008/8]",
+ // "index" : 2007
+ // }
+ //
+ // So we set the index to the one returned by the exception + 1
+ if (isOutdatedIndexException(exception)) {
+ LOGGER.debug("Outdated index, key: {}, cause={}", getPath(), exception.etcdCause);
+
+ // We set the index to the one returned by the exception + 1.
+ index.set(exception.index + 1);
+
+ // Clean-up the exception so it is not rethrown/handled
+ throwable = null;
+ }
+ } else {
+ try {
+ EtcdKeysResponse response = promise.get();
+
+ exchange = endpoint.createExchange();
+ exchange.getIn().setHeader(EtcdConstants.ETCD_NAMESPACE, getNamespace());
+ exchange.getIn().setHeader(EtcdConstants.ETCD_PATH, response.node.key);
+ exchange.getIn().setBody(response);
+
+ // Watch from the modifiedIndex + 1 of the node we got for ensuring
+ // no events are missed between watch commands
+ index.set(response.node.modifiedIndex + 1);
+ } catch (TimeoutException e) {
+ LOGGER.debug("Timeout watching for {}", getPath());
+
+ if (configuration.isSendEmptyExchangeOnTimeout()) {
+ exchange = endpoint.createExchange();
exchange.getIn().setHeader(EtcdConstants.ETCD_NAMESPACE, getNamespace());
exchange.getIn().setHeader(EtcdConstants.ETCD_TIMEOUT, true);
exchange.getIn().setHeader(EtcdConstants.ETCD_PATH, getPath());
exchange.getIn().setBody(null);
+ }
+
+ throwable = null;
+ } catch (Exception e1) {
+ throwable = e1;
+ }
+ if (exchange != null) {
+ try {
+ throwable = null;
getProcessor().process(exchange);
- } catch (Exception e1) {
- getExceptionHandler().handleException("Error processing exchange", exchange, e1);
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Error processing exchange", exchange, e);
}
}
+ }
+
+ if (throwable != null){
+ handleException("Error processing etcd response", throwable);
+ }
+
+ try {
+ watch();
} catch (Exception e) {
- throw new IllegalArgumentException(e);
+ handleException("Error watching key " + getPath(), e);
}
}
@@ -96,7 +144,7 @@ public class EtcdWatchConsumer extends AbstractEtcdConsumer implements ResponseP
return;
}
- EtcdKeyGetRequest request = getClient().get(getPath()).waitForChange();
+ EtcdKeyGetRequest request = getClient().get(getPath()).waitForChange(index.get());
if (configuration.isRecursive()) {
request.recursive();
}
@@ -104,10 +152,14 @@ public class EtcdWatchConsumer extends AbstractEtcdConsumer implements ResponseP
request.timeout(configuration.getTimeout(), TimeUnit.MILLISECONDS);
}
- try {
- request.send().addListener(this);
- } catch (IOException e) {
- throw new IllegalArgumentException(e);
+ request.send().addListener(this);
+ }
+
+ private boolean isOutdatedIndexException(EtcdException exception) {
+ if (exception.isErrorCode(EtcdErrorCode.EventIndexCleared) && exception.etcdMessage != null) {
+ return exception.etcdMessage.toLowerCase().contains(OUTDATED_EVENT_MSG);
}
+
+ return false;
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdKeysTest.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdKeysTest.java b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdKeysTest.java
index eca2339..cf16017 100644
--- a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdKeysTest.java
+++ b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdKeysTest.java
@@ -27,10 +27,9 @@ import org.apache.camel.Exchange;
import org.apache.camel.Predicate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
-import org.junit.Ignore;
import org.junit.Test;
-@Ignore("Etcd must be started manually")
+//@Ignore("Etcd must be started manually")
public class EtcdKeysTest extends EtcdTest {
@Test(expected = EtcdException.class)
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdStatsTest.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdStatsTest.java b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdStatsTest.java
index b142355..762be01 100644
--- a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdStatsTest.java
+++ b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdStatsTest.java
@@ -26,7 +26,7 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.junit.Ignore;
import org.junit.Test;
-@Ignore("Etcd must be started manually")
+//@Ignore("Etcd must be started manually")
public class EtcdStatsTest extends EtcdTest {
@Test
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdWatchTest.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdWatchTest.java b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdWatchTest.java
index 0bc7ab9..60cea00 100644
--- a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdWatchTest.java
+++ b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/EtcdWatchTest.java
@@ -17,29 +17,52 @@
package org.apache.camel.component.etcd;
import mousio.etcd4j.EtcdClient;
-import org.apache.camel.Exchange;
-import org.apache.camel.Predicate;
+import mousio.etcd4j.responses.EtcdErrorCode;
+import mousio.etcd4j.responses.EtcdException;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
-import org.junit.Ignore;
import org.junit.Test;
-@Ignore("Etcd must be started manually")
+//@Ignore("Etcd must be started manually")
public class EtcdWatchTest extends EtcdTest {
@Test
public void testWatchWithPath() throws Exception {
- testWatch("mock:watch-with-path", "/myKey1", true);
+ testWatch("mock:watch-with-path", "/myKey1", 10);
}
@Test
public void testWatchWithConfigPath() throws Exception {
- testWatch("mock:watch-with-config-path", "/myKey2", true);
+ testWatch("mock:watch-with-config-path", "/myKey2", 10);
}
@Test
public void testWatchRecursive() throws Exception {
- testWatch("mock:watch-recursive", "/recursive/myKey1", true);
+ testWatch("mock:watch-recursive", "/recursive/myKey1", 10);
+ }
+
+ @Test
+ public void testWatchRecovery() throws Exception {
+ final String key = "/myKeyRecovery";
+ final EtcdClient client = getClient();
+
+ try {
+ // Delete the key if present
+ client.delete(key).send().get();
+ } catch (EtcdException e) {
+ if (!e.isErrorCode(EtcdErrorCode.KeyNotFound)) {
+ throw e;
+ }
+ }
+
+ // Fill the vent backlog ( > 1000)
+ for (int i = 0; i < 2000; i++) {
+ client.put(key, "v" + i).send().get();
+ }
+
+ context().startRoute("watchRecovery");
+
+ testWatch("mock:watch-recovery", key, 10);
}
@Test
@@ -49,33 +72,25 @@ public class EtcdWatchTest extends EtcdTest {
mock.expectedHeaderReceived(EtcdConstants.ETCD_NAMESPACE, EtcdNamespace.watch.name());
mock.expectedHeaderReceived(EtcdConstants.ETCD_PATH, "/timeoutKey");
mock.expectedHeaderReceived(EtcdConstants.ETCD_TIMEOUT, true);
- mock.expectedMessagesMatches(new Predicate() {
- @Override
- public boolean matches(Exchange exchange) {
- return exchange.getIn().getBody() == null;
- }
- });
-
+ mock.allMessages().body().isNull();
mock.assertIsSatisfied();
}
- private void testWatch(String mockEndpoint, final String key, boolean updateKey) throws Exception {
+ private void testWatch(String mockEndpoint, final String key, int updates) throws Exception {
+ final String[] values = new String[updates];
+ for (int i = 0; i< updates; i++) {
+ values[i] = key + "=myValue-" + i;
+ }
+
MockEndpoint mock = getMockEndpoint(mockEndpoint);
mock.expectedMessageCount(2);
mock.expectedHeaderReceived(EtcdConstants.ETCD_NAMESPACE, EtcdNamespace.watch.name());
mock.expectedHeaderReceived(EtcdConstants.ETCD_PATH, key);
- mock.expectedMessagesMatches(new Predicate() {
- @Override
- public boolean matches(Exchange exchange) {
- return exchange.getIn().getBody(String.class).startsWith(key + "=myValue-");
- }
- });
+ mock.expectedBodiesReceived(values);
- if (updateKey) {
- EtcdClient client = getClient();
- client.put(key, "myValue-1").send().get();
- Thread.sleep(250);
- client.put(key, "myValue-2").send().get();
+ final EtcdClient client = getClient();
+ for (int i = 0; i< updates; i++) {
+ client.put(key, "myValue-" + i).send().get();
}
mock.assertIsSatisfied();
@@ -88,8 +103,13 @@ public class EtcdWatchTest extends EtcdTest {
from("etcd:watch/myKey1")
.process(NODE_TO_VALUE_IN)
.to("mock:watch-with-path");
- from("etcd:watch/recursive?recursive=true")
+ fromF("etcd:watch/myKeyRecovery?timeout=%d&fromIndex=%d", 1000 * 60 * 5, 1)
+ .id("watchRecovery")
+ .autoStartup(false)
.process(NODE_TO_VALUE_IN)
+ .to("mock:watch-recovery");
+ from("etcd:watch/recursive?recursive=true")
+ .process(NODE_TO_VALUE_IN)
.to("log:org.apache.camel.component.etcd?level=INFO")
.to("mock:watch-recursive");
from("etcd:watch/myKey2")
http://git-wip-us.apache.org/repos/asf/camel/blob/ed08a7a3/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index f52d59b..f8a2774 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -159,7 +159,7 @@
<el-api-1.0-version>1.0.1</el-api-1.0-version>
<!-- embedmongo 1.50.2 do not work -->
<embedmongo-version>1.50.1</embedmongo-version>
- <etcd4j-version>2.10.0</etcd4j-version>
+ <etcd4j-version>2.11.0</etcd4j-version>
<exec-maven-plugin-version>1.4.0</exec-maven-plugin-version>
<ezmorph-bundle-version>1.0.6_1</ezmorph-bundle-version>
<facebook4j-core-version>2.4.2</facebook4j-core-version>
[2/2] camel git commit: CAMEL-9834 : WatchConsumer does not properly
set watchIndex (disable stdout logging)
Posted by lb...@apache.org.
CAMEL-9834 : WatchConsumer does not properly set watchIndex (disable stdout logging)
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/45317409
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/45317409
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/45317409
Branch: refs/heads/camel-2.17.x
Commit: 45317409d5c73b0ccd5480ee3551f8d052225ee6
Parents: ed08a7a
Author: lburgazzoli <lb...@gmail.com>
Authored: Wed Apr 27 12:47:59 2016 +0200
Committer: lburgazzoli <lb...@gmail.com>
Committed: Wed Apr 27 12:47:59 2016 +0200
----------------------------------------------------------------------
components/camel-etcd/src/test/resources/log4j.properties | 8 +++++++-
1 file changed, 7 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/45317409/components/camel-etcd/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/test/resources/log4j.properties b/components/camel-etcd/src/test/resources/log4j.properties
index 3b1bd38..0a364b0 100644
--- a/components/camel-etcd/src/test/resources/log4j.properties
+++ b/components/camel-etcd/src/test/resources/log4j.properties
@@ -1,7 +1,7 @@
#
# The logging properties used
#
-log4j.rootLogger=INFO, out
+log4j.rootLogger=INFO, file
# uncomment the following line to turn on Camel debugging
#log4j.logger.org.apache.camel=DEBUG
@@ -12,3 +12,9 @@ log4j.appender.out.layout=org.apache.log4j.PatternLayout
log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
#log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+# File appender
+log4j.appender.file=org.apache.log4j.FileAppender
+log4j.appender.file.layout=org.apache.log4j.PatternLayout
+log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.file.file=target/camel-etcd-test.log
+