camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lburgazz...@apache.org
Subject [1/2] camel git commit: CAMEL-9834 : WatchConsumer does not properly set watchIndex
Date Wed, 27 Apr 2016 10:48:31 GMT
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 <lburgazzoli@gmail.com>
Authored: Thu Apr 7 14:14:27 2016 +0200
Committer: lburgazzoli <lburgazzoli@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>


Mime
View raw message