camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ningji...@apache.org
Subject svn commit: r1380174 - in /camel/trunk/components/camel-zookeeper/src: main/java/org/apache/camel/component/zookeeper/policy/ test/java/org/apache/camel/component/zookeeper/ test/java/org/apache/camel/component/zookeeper/policy/
Date Mon, 03 Sep 2012 08:58:31 GMT
Author: ningjiang
Date: Mon Sep  3 08:58:31 2012
New Revision: 1380174

URL: http://svn.apache.org/viewvc?rev=1380174&view=rev
Log:
CAMEL-5546 Applied the patch with thanks to Andrew

Added:
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ElectionWatcher.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperElection.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/ZookeeperElectionTest.java
Modified:
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperRoutePolicy.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperTestSupport.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverRoutePolicyTest.java

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ElectionWatcher.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ElectionWatcher.java?rev=1380174&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ElectionWatcher.java
(added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ElectionWatcher.java
Mon Sep  3 08:58:31 2012
@@ -0,0 +1,27 @@
+/**
+ * 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.camel.component.zookeeper.policy;
+
+
+public interface ElectionWatcher {
+    /**
+     * This method is called when there is a potential change to the master.
+     * Implementations should call "isMaster" on their ZookeeperElection
+     * instance to re-validate their status.
+     */
+    void electionResultChanged();
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperElection.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperElection.java?rev=1380174&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperElection.java
(added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperElection.java
Mon Sep  3 08:58:31 2012
@@ -0,0 +1,252 @@
+/**
+ * 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.camel.component.zookeeper.policy;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.Processor;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.zookeeper.SequenceComparator;
+import org.apache.camel.component.zookeeper.ZooKeeperEndpoint;
+import org.apache.camel.component.zookeeper.ZooKeeperMessage;
+import org.apache.camel.impl.JavaUuidGenerator;
+import org.apache.camel.spi.UuidGenerator;
+
+import org.apache.zookeeper.CreateMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * <code>ZooKeeperElection</code> uses the leader election capabilities of a
+ * ZooKeeper cluster to control which nodes are enabled. It is typically used in
+ * fail-over scenarios controlling identical instances of an application across
+ * a cluster of Camel based servers. <p> The election is configured with a 'top
+ * n' number of servers that should be marked as master, for a simple
+ * master/slave scenario this would be 1. Each instance will execute the
+ * election algorithm to obtain its position in the hierarchy of servers, if it
+ * is within the 'top n' servers then the node is enabled and isMaster() will
+ * return 'true'. If not it waits for a change in the leader hierarchy and then
+ * reruns this scenario to see if it is now in the top n. <p> All instances of
+ * the election must also be configured with the same path on the ZooKeeper
+ * cluster where the election will be carried out. It is good practice for this
+ * to indicate the application e.g. <tt>/someapplication/someroute/</tt> note
+ * that these nodes should exist before using the election. <p> See <a
+ * href="http://hadoop.apache.org/zookeeper/docs/current/recipes.html#sc_leaderElection">
+ * for more on how Leader election</a> is archived with ZooKeeper.
+ */
+public class ZooKeeperElection {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperElection.class);
+    private final ProducerTemplate producerTemplate;
+    private final CamelContext camelContext;
+    private final String uri;
+    private final String candidateName;
+    private final Lock lock = new ReentrantLock();
+    private final CountDownLatch electionComplete = new CountDownLatch(1);
+    private AtomicBoolean masterNode = new AtomicBoolean();
+    private volatile boolean isCandidateCreated;
+    private int enabledCount = 1;
+    private UuidGenerator uuidGenerator = new JavaUuidGenerator();
+    private final List<ElectionWatcher> watchers = new ArrayList<ElectionWatcher>();
+
+    public ZooKeeperElection(CamelContext camelContext, String uri, int enabledCount) {
+        this(camelContext.createProducerTemplate(), camelContext, uri, enabledCount);
+    }
+
+    public ZooKeeperElection(ProducerTemplate producerTemplate, CamelContext camelContext,
String uri, int enabledCount) {
+        this.camelContext = camelContext;
+        this.producerTemplate = producerTemplate;
+        this.uri = uri;
+        this.enabledCount = enabledCount;
+        this.candidateName = createCandidateName();
+    }
+
+    public boolean isMaster() {
+        if (!isCandidateCreated) {
+            testAndCreateCandidateNode();
+            awaitElectionResults();
+
+        }
+        return masterNode.get();
+    }
+
+    private String createCandidateName() {
+        StringBuilder builder = new StringBuilder();
+        try {
+            /* UUID would be enough, also using hostname for human readability */
+            builder.append(InetAddress.getLocalHost().getCanonicalHostName());
+        } catch (UnknownHostException ex) {
+            LOG.warn("Failed to get the local hostname.", ex);
+            builder.append("unknown-host");
+        }
+        builder.append("-").append(uuidGenerator.generateUuid());
+        return builder.toString();
+    }
+
+    private void testAndCreateCandidateNode() {
+        try {
+            lock.lock();
+            if (!isCandidateCreated) {
+                createCandidateNode(camelContext);
+                isCandidateCreated = true;
+            }
+        } catch (Exception e) {
+            handleException(e);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void awaitElectionResults() {
+        while (electionComplete.getCount() > 0) {
+            try {
+                LOG.debug("Awaiting election results...");
+                electionComplete.await();
+            } catch (InterruptedException e1) {
+            }
+        }
+    }
+
+    private ZooKeeperEndpoint createCandidateNode(CamelContext camelContext) {
+        LOG.info("Initializing ZookeeperElection with uri '{}'", uri);
+        ZooKeeperEndpoint zep = camelContext.getEndpoint(uri, ZooKeeperEndpoint.class);
+        zep.getConfiguration().setCreate(true);
+        String fullpath = createFullPathToCandidate(zep);
+        Exchange e = zep.createExchange();
+        e.setPattern(ExchangePattern.InOut);
+        e.getIn().setHeader(ZooKeeperMessage.ZOOKEEPER_NODE, fullpath);
+        e.getIn().setHeader(ZooKeeperMessage.ZOOKEEPER_CREATE_MODE, CreateMode.EPHEMERAL_SEQUENTIAL);
+        producerTemplate.send(zep, e);
+
+        if (e.isFailed()) {
+            LOG.error("Error setting up election node " + fullpath, e.getException());
+        } else {
+            LOG.info("Candidate node '{}' has been created", fullpath);
+            try {
+                if (zep != null) {
+                    camelContext.addRoutes(new ElectoralMonitorRoute(zep));
+                }
+            } catch (Exception ex) {
+                LOG.error("Error configuring ZookeeperElection", ex);
+            }
+        }
+        return zep;
+
+    }
+
+    private String createFullPathToCandidate(ZooKeeperEndpoint zep) {
+        String fullpath = zep.getConfiguration().getPath();
+        if (!fullpath.endsWith("/")) {
+            fullpath += "/";
+        }
+        fullpath += candidateName;
+        return fullpath;
+    }
+
+    private void handleException(Exception e) {
+        throw new RuntimeException(e.getMessage(), e);
+    }
+
+    private void notifyElectionWatchers() {
+        for (ElectionWatcher watcher : watchers) {
+            try {
+                watcher.electionResultChanged();
+            } catch (Exception e) {
+                LOG.warn("Election watcher " + watcher + " of type " + watcher.getClass()
+ " threw an exception.", e);
+            }
+        }
+    }
+
+    public boolean addElectionWatcher(ElectionWatcher e) {
+        return watchers.add(e);
+    }
+
+    public boolean removeElectionWatcher(ElectionWatcher o) {
+        return watchers.remove(o);
+    }
+
+    private class ElectoralMonitorRoute extends RouteBuilder {
+
+        private SequenceComparator comparator = new SequenceComparator();
+        private ZooKeeperEndpoint zep;
+
+        public ElectoralMonitorRoute(ZooKeeperEndpoint zep) {
+            this.zep = zep;
+            zep.getConfiguration().setListChildren(true);
+            zep.getConfiguration().setSendEmptyMessageOnDelete(true);
+            zep.getConfiguration().setRepeat(true);
+        }
+
+        @Override
+        public void configure() throws Exception {
+
+            /**
+             * TODO: this is cheap cheerful but suboptimal; it suffers from the
+             * 'herd effect' that on any change to the candidates list every
+             * policy instance will ask for the entire candidate list again.
+             * This is fine for small numbers of nodes (for scenarios like
+             * Master-Slave it is perfect) but could get noisy if large numbers
+             * of nodes were involved. <p> Better would be to find the position
+             * of this node in the list and watch the node in the position ahead
+             * node ahead of this and only request the candidate list when its
+             * status changes. This will require enhancing the consumer to allow
+             * custom operation lists.
+             */
+            from(zep).id("election-route-" + candidateName.substring(0, 8)).sort(body(),
comparator).process(new Processor() {
+                @Override
+                public void process(Exchange e) throws Exception {
+                    @SuppressWarnings("unchecked")
+                    List<String> candidates = e.getIn().getMandatoryBody(List.class);
+
+                    int location = Math.abs(Collections.binarySearch(candidates, candidateName));
+                    /**
+                     * check if the item at this location starts with this nodes
+                     * candidate name
+                     */
+                    if (isOurCandidateAtLocationInCandidatesList(candidates, location)) {
+
+                        masterNode.set(location <= enabledCount);
+                        LOG.debug(
+                                "This node is number '{}' on the candidate list, election
is configured for the top '{}'. this node will be {}",
+                                new Object[]{location, enabledCount, masterNode.get() ? "enabled"
: "disabled"}
+                        );
+                    }
+                    electionComplete.countDown();
+
+                    notifyElectionWatchers();
+                }
+
+                private boolean isOurCandidateAtLocationInCandidatesList(List<String>
candidates, int location) {
+                    return location <= candidates.size() && candidates.get(location
- 1).startsWith(candidateName);
+                }
+            });
+        }
+    }
+}

Modified: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperRoutePolicy.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperRoutePolicy.java?rev=1380174&r1=1380173&r2=1380174&view=diff
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperRoutePolicy.java
(original)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperRoutePolicy.java
Mon Sep  3 08:58:31 2012
@@ -16,32 +16,15 @@
  */
 package org.apache.camel.component.zookeeper.policy;
 
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.Collections;
-import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
-import org.apache.camel.ExchangePattern;
-import org.apache.camel.Processor;
-import org.apache.camel.ProducerTemplate;
 import org.apache.camel.Route;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.zookeeper.SequenceComparator;
-import org.apache.camel.component.zookeeper.ZooKeeperEndpoint;
-import org.apache.camel.component.zookeeper.ZooKeeperMessage;
-import org.apache.camel.impl.JavaUuidGenerator;
 import org.apache.camel.impl.RoutePolicySupport;
-import org.apache.camel.spi.UuidGenerator;
-
-import org.apache.zookeeper.CreateMode;
 
 /**
  * <code>ZooKeeperRoutePolicy</code> uses the leader election capabilities of
a
@@ -65,81 +48,57 @@ import org.apache.zookeeper.CreateMode;
  * See <a href="http://hadoop.apache.org/zookeeper/docs/current/recipes.html#sc_leaderElection">
  *     for more on how Leader election</a> is archived with ZooKeeper.
  */
-public class ZooKeeperRoutePolicy extends RoutePolicySupport {
+public class ZooKeeperRoutePolicy extends RoutePolicySupport implements ElectionWatcher {
 
     private final String uri;
     private final int enabledCount;
-    private String candidateName;
     private final Lock lock = new ReentrantLock();
-    private final CountDownLatch electionComplete = new CountDownLatch(1);
     private final Set<Route> suspendedRoutes = new CopyOnWriteArraySet<Route>();
     private final AtomicBoolean shouldProcessExchanges = new AtomicBoolean();
-    private ProducerTemplate template;
     private volatile boolean shouldStopConsumer = true;
-    private final UuidGenerator uuidGenerator = new JavaUuidGenerator();
-    private volatile boolean isCandidateCreated;
+
+    private final Lock electionLock = new ReentrantLock();
+    private ZooKeeperElection election;
 
     public ZooKeeperRoutePolicy(String uri, int enabledCount) {
         this.uri = uri;
         this.enabledCount = enabledCount;
-        createCandidateName();
     }
 
-    private void createCandidateName() {
-        /** UUID would be enough, also using hostname for human readability */
-        StringBuilder b = new StringBuilder(fetchHostname());
-        b.append("-").append(uuidGenerator.generateUuid());
-        this.candidateName = b.toString();
-    }
-
-    private String fetchHostname() {
-        try {
-            return InetAddress.getLocalHost().getCanonicalHostName();
-        } catch (UnknownHostException ex) {
-            log.warn("Unable to determine the local hostname, using a default.", ex);
-            return "default";
-        }
+    public ZooKeeperRoutePolicy(ZooKeeperElection election) {
+        this.election = election;
+        this.uri = null;
+        this.enabledCount = -1;
     }
 
     @Override
     public void onExchangeBegin(Route route, Exchange exchange) {
-        testAndCreateCandidateNode(route);
+        ensureElectionIsCreated(route);
 
-        awaitElectionResults();
-        if (!shouldProcessExchanges.get()) {
+        if (election.isMaster()) {
             if (shouldStopConsumer) {
-                stopConsumer(route);
+                startConsumer(route);
             }
-
-            IllegalStateException e = new IllegalStateException("Zookeeper based route policy
prohibits processing exchanges, stopping route and failing the exchange");
-            exchange.setException(e);
-
         } else {
             if (shouldStopConsumer) {
-                startConsumer(route);
+                stopConsumer(route);
             }
-        }
-    }
 
-    private void testAndCreateCandidateNode(Route route) {
-        try {
-            lock.lock();
-            if (!isCandidateCreated) {
-                createCandidateNode(route.getRouteContext().getCamelContext());
-                isCandidateCreated = true;
-            }
-        } catch (Exception e) {
-            handleException(e);
-        } finally {
-            lock.unlock();
+            IllegalStateException e = new IllegalStateException("Zookeeper based route policy
prohibits processing exchanges, stopping route and failing the exchange");
+            exchange.setException(e);
         }
     }
 
-    private void awaitElectionResults() {
-        while (electionComplete.getCount() > 0) {
+    private void ensureElectionIsCreated(Route route) {
+        if (election == null) {
+            electionLock.lock();
             try {
-                electionComplete.await();
-            } catch (InterruptedException e1) {
+                if (election == null) { // re-test
+                    election = new ZooKeeperElection(route.getRouteContext().getCamelContext(),
uri, enabledCount);
+                    election.addElectionWatcher(this);
+                }
+            } finally {
+                electionLock.unlock();
             }
         }
     }
@@ -173,6 +132,13 @@ public class ZooKeeperRoutePolicy extend
         }
     }
 
+    @Override
+    public void electionResultChanged() {
+        if (election.isMaster()) {
+            startAllStoppedConsumers();
+        }
+    }
+
     private void startAllStoppedConsumers() {
         try {
             lock.lock();
@@ -200,99 +166,4 @@ public class ZooKeeperRoutePolicy extend
     public void setShouldStopConsumer(boolean shouldStopConsumer) {
         this.shouldStopConsumer = shouldStopConsumer;
     }
-
-    private ZooKeeperEndpoint createCandidateNode(CamelContext camelContext) {
-        this.template = camelContext.createProducerTemplate();
-        log.info("Initializing ZookeeperRoutePolicy with uri {}", uri);
-
-        ZooKeeperEndpoint zep = camelContext.getEndpoint(uri, ZooKeeperEndpoint.class);
-        zep.getConfiguration().setCreate(true);
-        String fullpath = createFullPathToCandidate(zep);
-        Exchange e = zep.createExchange();
-        e.setPattern(ExchangePattern.InOut);
-        e.getIn().setHeader(ZooKeeperMessage.ZOOKEEPER_NODE, fullpath);
-        e.getIn().setHeader(ZooKeeperMessage.ZOOKEEPER_CREATE_MODE, CreateMode.EPHEMERAL_SEQUENTIAL);
-        template.send(zep, e);
-
-        if (e.isFailed()) {
-            log.warn("Error setting up election node " + fullpath, e.getException());
-        } else {
-            log.info("Candidate node {} has been created", fullpath);
-            try {
-                camelContext.addRoutes(new ElectoralMonitorRoute(zep));
-            } catch (Exception ex) {
-                log.warn("Error configuring ZookeeperRoutePolicy. This exception is ignored.",
ex);
-            }
-        }
-        return zep;
-
-    }
-
-    private String createFullPathToCandidate(ZooKeeperEndpoint zep) {
-        String fullpath = zep.getConfiguration().getPath();
-        if (!fullpath.endsWith("/")) {
-            fullpath += "/";
-        }
-        fullpath += candidateName;
-        return fullpath;
-    }
-
-    private class ElectoralMonitorRoute extends RouteBuilder {
-
-        private SequenceComparator comparator = new SequenceComparator();
-        
-        private ZooKeeperEndpoint zep;
-
-        public ElectoralMonitorRoute(ZooKeeperEndpoint zep) {
-            this.zep = zep;
-            zep.getConfiguration().setListChildren(true);
-            zep.getConfiguration().setRepeat(true);
-        }
-
-        @Override
-        public void configure() throws Exception {
-
-            /**
-             * TODO: this is cheap cheerful but suboptimal; it suffers from the
-             * 'herd effect' that on any change to the candidates list every
-             * policy instance will ask for the entire candidate list again.
-             * This is fine for small numbers of nodes (for scenarios
-             * like Master-Slave it is perfect) but could get noisy if
-             * large numbers of nodes were involved.
-             * <p>
-             * Better would be to find the position of this node in the list and
-             * watch the node in the position ahead node ahead of this and only
-             * request the candidate list when its status changes. This will
-             * require enhancing the consumer to allow custom operation lists.
-             */
-            from(zep).sort(body(), comparator).process(new Processor() {
-
-                public void process(Exchange e) throws Exception {
-                    @SuppressWarnings("unchecked")
-                    List<String> candidates = e.getIn().getMandatoryBody(List.class);
-
-                    int location = Math.abs(Collections.binarySearch(candidates, candidateName));
-                    /**
-                     * check if the item at this location starts with this nodes
-                     * candidate name
-                     */
-                    if (isOurCandidateAtLocationInCandidatesList(candidates, location)) {
-
-                        shouldProcessExchanges.set(location <= enabledCount);
-                        if (log.isDebugEnabled()) {
-                            log.debug("This node is number {} on the candidate list, route
is configured for the top {}. Exchange processing will be {}",
-                                    new Object[]{location, enabledCount, shouldProcessExchanges.get()
? "enabled" : "disabled"});
-                        }
-                        startAllStoppedConsumers();
-                    }
-                    electionComplete.countDown();
-                }
-
-                private boolean isOurCandidateAtLocationInCandidatesList(List<String>
candidates, int location) {
-                    return location <= candidates.size() && candidates.get(location
- 1).startsWith(candidateName);
-                }
-            });
-        }
-    }
-
 }

Modified: camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperTestSupport.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperTestSupport.java?rev=1380174&r1=1380173&r2=1380174&view=diff
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperTestSupport.java
(original)
+++ camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperTestSupport.java
Mon Sep  3 08:58:31 2012
@@ -206,6 +206,15 @@ public class ZooKeeperTestSupport extend
             }
         }
 
+        public void deleteAll(String node) throws Exception {
+            delay(200);
+            log.debug("Deleting {} and it's immediate children", node);
+            for (String child : zk.getChildren(node, false)) {
+                delete(node + "/" + child);
+            }
+            delete(node);
+        }
+
         public void delete(String node) throws Exception {
             delay(200);
             log.debug("Deleting node " + node);

Modified: camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverRoutePolicyTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverRoutePolicyTest.java?rev=1380174&r1=1380173&r2=1380174&view=diff
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverRoutePolicyTest.java
(original)
+++ camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverRoutePolicyTest.java
Mon Sep  3 08:58:31 2012
@@ -27,8 +27,11 @@ import org.apache.camel.component.zookee
 import org.apache.camel.impl.DefaultCamelContext;
 import org.apache.commons.logging.LogFactory;
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class FailoverRoutePolicyTest extends ZooKeeperTestSupport {
+    private static final Logger LOG = LoggerFactory.getLogger(FailoverRoutePolicyTest.class);
 
     protected CamelContext createCamelContext() throws Exception {
         disableJMX();
@@ -76,6 +79,7 @@ public class FailoverRoutePolicyTest ext
                 template.sendBody("vm:" + routename, ExchangePattern.InOut, message);
             } catch (Exception e) {
                 if (expected > 0) {
+                    LOG.error(e.getMessage(), e);
                     fail("Expected messages...");
                 }
             }

Added: camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/ZookeeperElectionTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/ZookeeperElectionTest.java?rev=1380174&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/ZookeeperElectionTest.java
(added)
+++ camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/ZookeeperElectionTest.java
Mon Sep  3 08:58:31 2012
@@ -0,0 +1,121 @@
+/**
+ * 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.camel.component.zookeeper.policy;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.camel.component.zookeeper.ZooKeeperTestSupport;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ZookeeperElectionTest extends ZooKeeperTestSupport {
+    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperElectionTest.class);
+
+    private static final String NODE_BASE_KEY = "/someapp";
+    private static final String NODE_PARTICULAR_KEY = "/someapp/somepolicy";
+    private static final String ELECTION_URI = "zookeeper:localhost:39913/someapp/somepolicy";
+
+    @Before
+    public void before() throws Exception {
+        // set up the parent used to control the election
+        client.createPersistent(NODE_BASE_KEY, "App node to contain policy election nodes...");
+        client.createPersistent(NODE_PARTICULAR_KEY, "Policy node used by route policy to
control routes...");
+    }
+
+    @After
+    public void after() throws Exception {
+        client.deleteAll(NODE_PARTICULAR_KEY);
+        client.delete(NODE_BASE_KEY);
+    }
+
+    @Test
+    public void masterCanBeElected() throws Exception {
+        ZooKeeperElection candidate = new ZooKeeperElection(template, context, ELECTION_URI,
1);
+        assertTrue("The only election candidate was not elected as master.", candidate.isMaster());
+    }
+
+    @Test
+    public void masterAndSlave() throws Exception {
+        final DefaultCamelContext candidateOneContext = createNewContext();
+        final DefaultCamelContext candidateTwoContext = createNewContext();
+
+        ZooKeeperElection electionCandidate1 = createElectionCandidate(candidateOneContext,
1);
+        assertTrue("The first candidate was not elected.", electionCandidate1.isMaster());
+        ZooKeeperElection electionCandidate2 = createElectionCandidate(candidateTwoContext,
1);
+        assertFalse("The second candidate should not have been elected.", electionCandidate2.isMaster());
+    }
+
+    @Test
+    public void testMasterGoesAway() throws Exception {
+        final DefaultCamelContext candidateOneContext = createNewContext();
+        final DefaultCamelContext candidateTwoContext = createNewContext();
+
+        ZooKeeperElection electionCandidate1 = createElectionCandidate(candidateOneContext,
1);
+        assertTrue("The first candidate was not elected.", electionCandidate1.isMaster());
+        ZooKeeperElection electionCandidate2 = createElectionCandidate(candidateTwoContext,
1);
+        assertFalse("The second candidate should not have been elected.", electionCandidate2.isMaster());
+
+        LOG.debug("About to shutdown the first candidate.");
+
+        candidateOneContext.stop(); // the first candidate was killed.
+
+        delay(3000); // more than the timeout on the zeekeeper server.
+        assertTrue("The second candidate should have been elected.", electionCandidate2.isMaster());
+    }
+
+    @Test
+    public void testDualMaster() throws Exception {
+        final DefaultCamelContext candidateOneContext = createNewContext();
+        final DefaultCamelContext candidateTwoContext = createNewContext();
+
+        ZooKeeperElection electionCandidate1 = createElectionCandidate(candidateOneContext,
2);
+        assertTrue("The first candidate was not elected.", electionCandidate1.isMaster());
+        ZooKeeperElection electionCandidate2 = createElectionCandidate(candidateTwoContext,
2);
+        assertTrue("The second candidate should also be a master.", electionCandidate2.isMaster());
+    }
+
+    @Test
+    public void testWatchersAreNotified() throws Exception {
+        final DefaultCamelContext candidateOneContext = createNewContext();
+        final DefaultCamelContext candidateTwoContext = createNewContext();
+
+        final AtomicBoolean notified = new AtomicBoolean(false);
+        ElectionWatcher watcher = new ElectionWatcher() {
+            @Override public void electionResultChanged() { notified.set(true); }
+        };
+
+        ZooKeeperElection electionCandidate1 = createElectionCandidate(candidateOneContext,
2);
+        assertTrue("The first candidate was not elected.", electionCandidate1.isMaster());
+        electionCandidate1.addElectionWatcher(watcher);
+        ZooKeeperElection electionCandidate2 = createElectionCandidate(candidateTwoContext,
2);
+        electionCandidate2.isMaster();
+        assertTrue("The first candidate should have had it's watcher notified", notified.get());
+    }
+
+    private DefaultCamelContext createNewContext() throws Exception {
+        DefaultCamelContext controlledContext = new DefaultCamelContext();
+        controlledContext.start();
+        return controlledContext;
+    }
+
+    private ZooKeeperElection createElectionCandidate(final DefaultCamelContext context,
int masterCount) {
+        return new ZooKeeperElection(context.createProducerTemplate(), context, ELECTION_URI,
masterCount);
+    }
+}



Mime
View raw message