brooklyn-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From henev...@apache.org
Subject [46/64] incubator-brooklyn git commit: brooklyn-software-messaging: add org.apache package prefix
Date Tue, 18 Aug 2015 11:01:01 GMT
http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/messaging/storm/StormSshDriver.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/messaging/storm/StormSshDriver.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/messaging/storm/StormSshDriver.java
new file mode 100644
index 0000000..1f108a0
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/messaging/storm/StormSshDriver.java
@@ -0,0 +1,272 @@
+/*
+ * 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.brooklyn.entity.messaging.storm;
+
+import static java.lang.String.format;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.entity.basic.EntityLocal;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.Entities;
+import brooklyn.entity.basic.SoftwareProcess;
+import brooklyn.entity.java.JavaSoftwareProcessSshDriver;
+import org.apache.brooklyn.entity.zookeeper.ZooKeeperEnsemble;
+import brooklyn.event.basic.DependentConfiguration;
+import org.apache.brooklyn.location.basic.Machines;
+import org.apache.brooklyn.location.basic.SshMachineLocation;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.net.Networking;
+import brooklyn.util.os.Os;
+import brooklyn.util.ssh.BashCommands;
+import brooklyn.util.time.Duration;
+import brooklyn.util.time.Time;
+
+import com.google.common.base.Optional;
+import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+public class StormSshDriver extends JavaSoftwareProcessSshDriver implements StormDriver {
+
+    private static final Logger log = LoggerFactory.getLogger(StormSshDriver.class);
+
+    public StormSshDriver(EntityLocal entity, SshMachineLocation machine) {
+        super(entity, machine);
+    }
+
+    public String getRoleName() {
+        return entity.getConfig(Storm.ROLE).name().toLowerCase();
+    }
+
+    public String getZeromqVersion() {
+        return entity.getConfig(Storm.ZEROMQ_VERSION);
+    }
+
+    public String getLocalDir() {
+        return Optional.fromNullable(entity.getConfig(Storm.LOCAL_DIR)).or(Os.mergePathsUnix(getRunDir(), "storm"));
+    }
+
+    public String getNimbusHostname() {
+        String result = entity.getConfig(Storm.NIMBUS_HOSTNAME);
+        if (result != null) return result;
+
+        Entity nimbus = entity.getConfig(Storm.NIMBUS_ENTITY);
+        if (nimbus == null) {
+            log.warn("No nimbus hostname available; using 'localhost'");
+            return "localhost";
+        }
+        return Entities.submit(entity, DependentConfiguration.attributeWhenReady(nimbus, Attributes.HOSTNAME)).getUnchecked();
+    }
+
+    public Integer getUiPort() {
+        return entity.getAttribute(Storm.UI_PORT);
+    }
+
+    public Map<String, Integer> getPortMap() {
+        return MutableMap.of("uiPort", getUiPort());
+    }
+
+    @Override
+    protected List<String> getCustomJavaConfigOptions() {
+        List<String> result = super.getCustomJavaConfigOptions();
+        if ("nimbus".equals(getRoleName()) || "supervisor".equals(getRoleName())) {
+            result.add("-verbose:gc");
+            result.add("-XX:+PrintGCTimeStamps");
+            result.add("-XX:+PrintGCDetails");
+        }
+
+        if ("ui".equals(getRoleName())) {
+            result.add("-Xmx768m");
+        }
+
+        return result;
+    }
+
+    public String getJvmOptsLine() {
+        return Optional.fromNullable(getShellEnvironment().get("JAVA_OPTS")).or("");
+    }
+    
+    public List<String> getZookeeperServers() {
+        ZooKeeperEnsemble zooKeeperEnsemble = entity.getConfig(Storm.ZOOKEEPER_ENSEMBLE);
+        Supplier<List<String>> supplier = Entities.attributeSupplierWhenReady(zooKeeperEnsemble, ZooKeeperEnsemble.ZOOKEEPER_SERVERS);
+        return supplier.get();
+    }
+
+    public String getStormConfigTemplateUrl() {
+        return entity.getConfig(Storm.STORM_CONFIG_TEMPLATE_URL);
+    }
+
+    @Override
+    public void preInstall() {
+        resolver = Entities.newDownloader(this);
+        setExpandedInstallDir(Os.mergePaths(getInstallDir(), resolver.getUnpackedDirectoryName(format("storm-%s", getVersion()))));
+    }
+
+    @Override
+    public void install() {
+        List<String> urls = resolver.getTargets();
+        String saveAs = resolver.getFilename();
+        
+        ImmutableList.Builder<String> commands= ImmutableList.<String> builder();
+        if (!getLocation().getOsDetails().isMac()) {
+            commands.add(BashCommands.installPackage(ImmutableMap.of(
+                        "yum", "libuuid-devel",
+                        "apt", "build-essential uuid-dev pkg-config libtool automake"), 
+                    "libuuid-devel"));
+            commands.add(BashCommands.ifExecutableElse0("yum", BashCommands.sudo("yum -y groupinstall 'Development Tools'")));
+        }
+        commands.add(BashCommands.installPackage(ImmutableMap.of("yum", "git"), "git"))
+                .add(BashCommands.INSTALL_UNZIP)
+                .addAll(installNativeDependencies())
+                .addAll(BashCommands.commandsToDownloadUrlsAs(urls, saveAs))
+                .add("unzip " + saveAs)
+                .add("mkdir -p " + getLocalDir())
+                .add("chmod 777 " + getLocalDir()); // FIXME
+        newScript(INSTALLING)
+                .body.append(commands.build())
+                .gatherOutput()
+                .execute();
+    }
+
+    public String getPidFile() {
+        return Os.mergePathsUnix(getRunDir(), format("%s.pid", getRoleName()));
+    }
+
+    @Override
+    protected String getLogFileLocation() {
+        return Os.mergePathsUnix(getRunDir(), "logs", format("%s.log", getRoleName()));
+    }
+
+    @Override
+    public void launch() {
+        boolean needsSleep = false;
+        if (getRoleName().equals("supervisor")) {
+            Entity nimbus = entity.getConfig(Storm.NIMBUS_ENTITY);
+            if (nimbus == null) {
+                log.warn("No nimbus entity available; not blocking before starting supervisors");
+            } else {
+                Entities.waitForServiceUp(nimbus, entity.getConfig(SoftwareProcess.START_TIMEOUT));
+                needsSleep = true;
+            }
+        }
+
+        String subnetHostname = Machines.findSubnetOrPublicHostname(entity).get();
+        log.info("Launching " + entity + " with role " + getRoleName() + " and " + "hostname (public) " 
+                + getEntity().getAttribute(Attributes.HOSTNAME) + ", " + "hostname (subnet) " + subnetHostname + ")");
+
+        // ensure only one node at a time tries to start
+        // attempting to eliminate the causes of:
+        // 2013-12-12 09:21:45 supervisor [ERROR] Error on initialization of server mk-supervisor
+        // org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /assignments
+        // TODO use SoftwareProcess#START_LATCH instead here?
+
+        Object startMutex = Optional.fromNullable(entity.getConfig(Storm.START_MUTEX)).or(new Object());
+        synchronized (startMutex) {
+            if (needsSleep) {
+                // give 10s extra to make sure nimbus is ready; we see weird zookeeper no /assignments node error otherwise
+                // (this could be optimized by recording nimbus service_up time)
+                Time.sleep(Duration.TEN_SECONDS);
+            }
+            newScript(MutableMap.of(USE_PID_FILE, getPidFile()), LAUNCHING)
+                    .body.append(format("nohup ./bin/storm %s > %s 2>&1 &", getRoleName(), getLogFileLocation()))
+                    .execute();
+        }
+    }
+
+    @Override
+    public boolean isRunning() {
+        return newScript(MutableMap.of(USE_PID_FILE, getPidFile()), CHECK_RUNNING).execute() == 0;
+    }
+
+    @Override
+    public void stop() {
+        newScript(MutableMap.of(USE_PID_FILE, getPidFile()), STOPPING).execute();
+    }
+
+    @Override
+    public void customize() {
+        log.debug("Customizing {}", entity);
+        Networking.checkPortsValid(getPortMap());
+
+        newScript(CUSTOMIZING)
+                .body.append(format("cp -R %s/* .", getExpandedInstallDir()))
+                .execute();
+
+        String destinationConfigFile = Os.mergePathsUnix(getRunDir(), "conf/storm.yaml");
+        copyTemplate(getStormConfigTemplateUrl(), destinationConfigFile);
+    }
+
+    protected List<String> installNativeDependencies() {
+        String zeromqUrl = format("http://download.zeromq.org/zeromq-%s.tar.gz", getZeromqVersion());
+        String targz = format("zeromq-%s.tar.gz", getZeromqVersion());
+        String jzmq = "https://github.com/nathanmarz/jzmq.git";
+
+        ImmutableList.Builder<String> commands = ImmutableList.<String>builder();
+        if (getLocation().getOsDetails().isMac()) {
+            commands.add("export PATH=$PATH:/usr/local/bin")
+                   .add("export JAVA_HOME=$(/usr/libexec/java_home)")
+                   .add("cd " + getInstallDir())
+                   .add(BashCommands.installPackage(ImmutableMap.of("brew", "automake"), "make"))
+                   .add(BashCommands.installPackage(ImmutableMap.of("brew", "libtool"), "libtool"))
+                   .add(BashCommands.installPackage(ImmutableMap.of("brew", "pkg-config"), "pkg-config"))
+                   .add(BashCommands.installPackage(ImmutableMap.of("brew", "zeromq"), "zeromq"))
+                   .add("git clone https://github.com/asmaier/jzmq")
+                   .add("cd jzmq")
+                   .add("./autogen.sh")
+                   .add("./configure")
+                   .add("make")
+                   .add((BashCommands.sudo("make install")))
+                   .add("cd " + getInstallDir());
+        } else {
+            commands.add("export JAVA_HOME=$(dirname $(readlink -m `which java`))/../../ || export JAVA_HOME=/usr/lib/jvm/java")
+                   .add("cd " + getInstallDir())
+                   .add(BashCommands.commandToDownloadUrlAs(zeromqUrl, targz))
+                   .add("tar xzf " + targz)
+                   .add(format("cd zeromq-%s", getZeromqVersion()))
+                   .add("./configure")
+                   .add("make")
+                   .add((BashCommands.sudo("make install")))
+                   // install jzmq
+                   .add("cd " + getInstallDir())
+                   .add("git clone " + jzmq)
+                   .add("cd jzmq")
+                   .add("./autogen.sh")
+                   .add("./configure")
+                           
+                   // hack needed on ubuntu 12.04; ignore if it fails
+                   // see https://github.com/zeromq/jzmq/issues/114
+                   .add(BashCommands.ok(
+                       "pushd src ; touch classdist_noinst.stamp ; CLASSPATH=.:./.:$CLASSPATH "
+                       + "javac -d . org/zeromq/ZMQ.java org/zeromq/App.java org/zeromq/ZMQForwarder.java org/zeromq/EmbeddedLibraryTools.java org/zeromq/ZMQQueue.java org/zeromq/ZMQStreamer.java org/zeromq/ZMQException.java"))
+                   .add(BashCommands.ok("popd"))
+
+                   .add("make")
+                   .add((BashCommands.sudo("make install")))
+                   .add("cd " + getInstallDir());
+        }
+        return commands.build();
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/AbstractZooKeeperImpl.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/AbstractZooKeeperImpl.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/AbstractZooKeeperImpl.java
new file mode 100644
index 0000000..6cb5ab0
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/AbstractZooKeeperImpl.java
@@ -0,0 +1,109 @@
+/*
+ * 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.brooklyn.entity.zookeeper;
+
+import java.util.concurrent.TimeUnit;
+
+import javax.management.ObjectName;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import brooklyn.entity.basic.SoftwareProcessImpl;
+import brooklyn.entity.java.JavaSoftwareProcessDriver;
+import brooklyn.event.feed.jmx.JmxAttributePollConfig;
+import brooklyn.event.feed.jmx.JmxFeed;
+import brooklyn.event.feed.jmx.JmxHelper;
+
+import com.google.common.base.Functions;
+import com.google.common.base.Objects.ToStringHelper;
+
+/**
+ * An {@link org.apache.brooklyn.api.entity.Entity} that represents a single Apache ZooKeeper instance.
+ */
+public abstract class AbstractZooKeeperImpl extends SoftwareProcessImpl implements ZooKeeperNode {
+
+    @SuppressWarnings("unused")
+    private static final Logger log = LoggerFactory.getLogger(AbstractZooKeeperImpl.class);
+    private static final ObjectName ZOOKEEPER_MBEAN = JmxHelper.createObjectName("org.apache.ZooKeeperService:name0=StandaloneServer_port-1");
+
+    private volatile JmxFeed jmxFeed;
+
+    public AbstractZooKeeperImpl() {
+    }
+
+    @Override
+    public Integer getZookeeperPort() { return getAttribute(ZOOKEEPER_PORT); }
+
+    @Override
+    public String getHostname() { return getAttribute(HOSTNAME); }
+
+    @Override
+    public void waitForServiceUp(long duration, TimeUnit units) {
+        super.waitForServiceUp(duration, units);
+
+        if (((JavaSoftwareProcessDriver)getDriver()).isJmxEnabled()) {
+            // Wait for the MBean to exist
+            JmxHelper helper = new JmxHelper(this);
+            try {
+                helper.assertMBeanExistsEventually(ZOOKEEPER_MBEAN, units.toMillis(duration));
+            } finally {
+                helper.terminate();
+            }
+        }
+    }
+
+    @Override
+    protected void connectSensors() {
+        connectServiceUpIsRunning();
+
+        if (((JavaSoftwareProcessDriver)getDriver()).isJmxEnabled()) {
+            jmxFeed = JmxFeed.builder()
+                .entity(this)
+                .period(500, TimeUnit.MILLISECONDS)
+                .pollAttribute(new JmxAttributePollConfig<Long>(OUTSTANDING_REQUESTS)
+                        .objectName(ZOOKEEPER_MBEAN)
+                        .attributeName("OutstandingRequests")
+                        .onFailureOrException(Functions.constant(-1l)))
+                .pollAttribute(new JmxAttributePollConfig<Long>(PACKETS_RECEIVED)
+                        .objectName(ZOOKEEPER_MBEAN)
+                        .attributeName("PacketsReceived")
+                        .onFailureOrException(Functions.constant(-1l)))
+                .pollAttribute(new JmxAttributePollConfig<Long>(PACKETS_SENT)
+                        .objectName(ZOOKEEPER_MBEAN)
+                        .attributeName("PacketsSent")
+                        .onFailureOrException(Functions.constant(-1l)))
+                .build();
+        }
+    }
+
+    @Override
+    public void disconnectSensors() {
+        super.disconnectSensors();
+        disconnectServiceUpIsRunning();
+        if (jmxFeed != null) jmxFeed.stop();
+    }
+
+    @Override
+    protected ToStringHelper toStringHelper() {
+        return super.toStringHelper()
+                .add("zookeeperPort", getZookeeperPort());
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperDriver.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperDriver.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperDriver.java
new file mode 100644
index 0000000..36388f0
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperDriver.java
@@ -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.brooklyn.entity.zookeeper;
+
+import brooklyn.entity.java.JavaSoftwareProcessDriver;
+
+public interface ZooKeeperDriver extends JavaSoftwareProcessDriver {
+
+    Integer getZooKeeperPort();
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsemble.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsemble.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsemble.java
new file mode 100644
index 0000000..d29bc91
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsemble.java
@@ -0,0 +1,53 @@
+/*
+ * 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.brooklyn.entity.zookeeper;
+
+import java.util.List;
+
+import org.apache.brooklyn.api.catalog.Catalog;
+import org.apache.brooklyn.api.entity.proxying.ImplementedBy;
+import org.apache.brooklyn.api.event.AttributeSensor;
+import org.apache.brooklyn.core.util.flags.SetFromFlag;
+
+import brooklyn.config.ConfigKey;
+import brooklyn.entity.basic.ConfigKeys;
+import brooklyn.entity.group.DynamicCluster;
+import brooklyn.event.basic.BasicAttributeSensorAndConfigKey;
+import brooklyn.event.basic.Sensors;
+
+import com.google.common.reflect.TypeToken;
+
+@Catalog(name="ZooKeeper ensemble", description="A cluster of ZooKeeper servers. "
+        + "Apache ZooKeeper enables highly reliable distributed coordination.")
+@ImplementedBy(ZooKeeperEnsembleImpl.class)
+public interface ZooKeeperEnsemble extends DynamicCluster {
+
+    @SetFromFlag("clusterName")
+    BasicAttributeSensorAndConfigKey<String> CLUSTER_NAME = new BasicAttributeSensorAndConfigKey<String>(String
+            .class, "zookeeper.cluster.name", "Name of the Zookeeper cluster", "BrooklynZookeeperCluster");
+
+    @SetFromFlag("initialSize")
+    public static final ConfigKey<Integer> INITIAL_SIZE = ConfigKeys.newConfigKeyWithDefault(DynamicCluster.INITIAL_SIZE, 3);
+
+    @SuppressWarnings("serial")
+    AttributeSensor<List<String>> ZOOKEEPER_SERVERS = Sensors.newSensor(new TypeToken<List<String>>() { },
+            "zookeeper.servers", "Hostnames to connect to cluster with");
+
+    String getClusterName();
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsembleImpl.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsembleImpl.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsembleImpl.java
new file mode 100644
index 0000000..f9ce930
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperEnsembleImpl.java
@@ -0,0 +1,105 @@
+/*
+ * 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.brooklyn.entity.zookeeper;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.entity.proxying.EntitySpec;
+import org.apache.brooklyn.api.location.Location;
+import org.apache.brooklyn.api.policy.PolicySpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import brooklyn.entity.basic.Attributes;
+import brooklyn.entity.basic.EntityInternal;
+import brooklyn.entity.group.AbstractMembershipTrackingPolicy;
+import brooklyn.entity.group.DynamicClusterImpl;
+
+import com.google.common.collect.Lists;
+
+public class ZooKeeperEnsembleImpl extends DynamicClusterImpl implements ZooKeeperEnsemble {
+
+    private static final Logger log = LoggerFactory.getLogger(ZooKeeperEnsembleImpl.class);
+    private static final AtomicInteger myId = new AtomicInteger();
+    
+    private MemberTrackingPolicy policy;
+
+    public ZooKeeperEnsembleImpl() {}
+
+    /**
+     * Sets the default {@link #MEMBER_SPEC} to describe the ZooKeeper nodes.
+     */
+    @Override
+    protected EntitySpec<?> getMemberSpec() {
+        return getConfig(MEMBER_SPEC, EntitySpec.create(ZooKeeperNode.class));
+    }
+
+    @Override
+    public String getClusterName() {
+        return getAttribute(CLUSTER_NAME);
+    }
+
+    @Override
+    public void init() {
+        log.info("Initializing the ZooKeeper Ensemble");
+        super.init();
+
+        policy = addPolicy(PolicySpec.create(MemberTrackingPolicy.class)
+                .displayName("Members tracker")
+                .configure("group", this));
+    }
+
+    public static class MemberTrackingPolicy extends AbstractMembershipTrackingPolicy {
+        @Override
+        protected void onEntityChange(Entity member) {
+        }
+
+        @Override
+        protected void onEntityAdded(Entity member) {
+            if (member.getAttribute(ZooKeeperNode.MY_ID) == null) {
+                ((EntityInternal) member).setAttribute(ZooKeeperNode.MY_ID, myId.incrementAndGet());
+            }
+        }
+
+        @Override
+        protected void onEntityRemoved(Entity member) {
+        }
+    };
+
+    @Override
+    protected void initEnrichers() {
+        super.initEnrichers();
+        
+    }
+    
+    @Override
+    public void start(Collection<? extends Location> locations) {
+        super.start(locations);
+        
+        List<String> zookeeperServers = Lists.newArrayList();
+        for (Entity zookeeper : getMembers()) {
+            zookeeperServers.add(zookeeper.getAttribute(Attributes.HOSTNAME));
+        }
+        setAttribute(ZOOKEEPER_SERVERS, zookeeperServers);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNode.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNode.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNode.java
new file mode 100644
index 0000000..6a67394
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNode.java
@@ -0,0 +1,67 @@
+/*
+ * 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.brooklyn.entity.zookeeper;
+
+import org.apache.brooklyn.api.catalog.Catalog;
+import org.apache.brooklyn.api.entity.proxying.ImplementedBy;
+import org.apache.brooklyn.api.event.AttributeSensor;
+import org.apache.brooklyn.core.util.flags.SetFromFlag;
+
+import brooklyn.config.ConfigKey;
+import brooklyn.entity.basic.ConfigKeys;
+import brooklyn.entity.basic.SoftwareProcess;
+import brooklyn.event.basic.BasicAttributeSensor;
+import brooklyn.event.basic.BasicAttributeSensorAndConfigKey;
+import brooklyn.event.basic.PortAttributeSensorAndConfigKey;
+
+/**
+ * An {@link org.apache.brooklyn.api.entity.Entity} that represents a single Apache ZooKeeper instance.
+ */
+@Catalog(name="ZooKeeper Node", description="Apache ZooKeeper is a server which enables "
+        + "highly reliable distributed coordination.")
+@ImplementedBy(ZooKeeperNodeImpl.class)
+public interface ZooKeeperNode extends SoftwareProcess {
+
+    @SetFromFlag("version")
+    ConfigKey<String> SUGGESTED_VERSION = ConfigKeys.newConfigKeyWithDefault(SoftwareProcess.SUGGESTED_VERSION, "3.4.5");
+    @SetFromFlag("zookeeperPort")
+    PortAttributeSensorAndConfigKey ZOOKEEPER_PORT = new PortAttributeSensorAndConfigKey("zookeeper.port", "Zookeeper port", "2181+");
+    @SetFromFlag("zookeeperLeaderPort")
+    PortAttributeSensorAndConfigKey ZOOKEEPER_LEADER_PORT = new PortAttributeSensorAndConfigKey("zookeeper.leader.port", "Zookeeper leader ports", "2888+");
+    @SetFromFlag("zookeeperElectionPort")
+    PortAttributeSensorAndConfigKey ZOOKEEPER_ELECTION_PORT = new PortAttributeSensorAndConfigKey("zookeeper.election.port", "Zookeeper election ports", "3888+");
+    @SetFromFlag("downloadUrl")
+    BasicAttributeSensorAndConfigKey<String> DOWNLOAD_URL = new BasicAttributeSensorAndConfigKey<String>(
+            SoftwareProcess.DOWNLOAD_URL, "http://apache.fastbull.org/zookeeper/zookeeper-${version}/zookeeper-${version}.tar.gz");
+    /**
+     * Location of the ZK configuration file template to be copied to the server.
+     */
+    @SetFromFlag("zookeeperConfig")
+    ConfigKey<String> ZOOKEEPER_CONFIG_TEMPLATE = ConfigKeys.newStringConfigKey(
+            "zookeeper.configTemplate", "Zookeeper configuration template (in freemarker format)",
+            "classpath://org/apache/brooklyn/entity/messaging/zookeeper/zoo.cfg");
+    AttributeSensor<Long> OUTSTANDING_REQUESTS = new BasicAttributeSensor<Long>(Long.class, "zookeeper.outstandingRequests", "Outstanding request count");
+    AttributeSensor<Long> PACKETS_RECEIVED = new BasicAttributeSensor<Long>(Long.class, "zookeeper.packets.received", "Total packets received");
+    AttributeSensor<Long> PACKETS_SENT = new BasicAttributeSensor<Long>(Long.class, "zookeeper.packets.sent", "Total packets sent");
+    AttributeSensor<Integer> MY_ID = new BasicAttributeSensor<Integer>(Integer.class, "zookeeper.myid", "ZooKeeper node's myId");
+
+    Integer getZookeeperPort();
+
+    String getHostname();
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNodeImpl.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNodeImpl.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNodeImpl.java
new file mode 100644
index 0000000..275e101
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperNodeImpl.java
@@ -0,0 +1,33 @@
+/*
+ * 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.brooklyn.entity.zookeeper;
+
+/**
+ * An {@link org.apache.brooklyn.api.entity.Entity} that represents a single standalone zookeeper instance.
+ */
+public class ZooKeeperNodeImpl extends AbstractZooKeeperImpl implements ZooKeeperNode {
+
+    public ZooKeeperNodeImpl() {}
+
+    @Override
+    public Class<?> getDriverInterface() {
+        return ZooKeeperDriver.class;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperSshDriver.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperSshDriver.java b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperSshDriver.java
new file mode 100644
index 0000000..709e44c
--- /dev/null
+++ b/software/messaging/src/main/java/org/apache/brooklyn/entity/zookeeper/ZooKeeperSshDriver.java
@@ -0,0 +1,163 @@
+/*
+ * 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.brooklyn.entity.zookeeper;
+
+import static java.lang.String.format;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.brooklyn.api.entity.Entity;
+
+import brooklyn.entity.basic.Entities;
+import brooklyn.entity.java.JavaSoftwareProcessSshDriver;
+import org.apache.brooklyn.location.basic.SshMachineLocation;
+import brooklyn.util.collections.MutableMap;
+import brooklyn.util.net.Networking;
+import brooklyn.util.os.Os;
+import brooklyn.util.ssh.BashCommands;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+
+public class ZooKeeperSshDriver extends JavaSoftwareProcessSshDriver implements ZooKeeperDriver {
+
+    public ZooKeeperSshDriver(ZooKeeperNodeImpl entity, SshMachineLocation machine) {
+        super(entity, machine);
+    }
+
+    @Override
+    protected String getLogFileLocation() { return Os.mergePathsUnix(getRunDir(), "console.out"); }
+
+    protected Map<String, Integer> getPortMap() {
+        return MutableMap.of("zookeeperPort", getZooKeeperPort());
+    }
+
+    protected String getConfigFileName() {
+        return entity.getConfig(ZooKeeperNode.ZOOKEEPER_CONFIG_TEMPLATE);
+    }
+
+    protected int getMyId() {
+        return entity.getAttribute(ZooKeeperNode.MY_ID);
+    }
+
+    // FIXME All for one, and one for all! If any node fails then we're stuck waiting for its hostname/port forever.
+    // Need a way to terminate the wait based on the entity going on-fire etc.
+    // FIXME Race in getMemebers. Should we change DynamicCluster.grow to create the members and only then call start on them all?
+    public List<ZooKeeperServerConfig> getZookeeperServers() throws ExecutionException, InterruptedException {
+        ZooKeeperEnsemble ensemble = (ZooKeeperEnsemble) entity.getParent();
+        List<ZooKeeperServerConfig> result = Lists.newArrayList();
+
+        for (Entity member : ensemble.getMembers()) {
+            Integer myid = Entities.attributeSupplierWhenReady(member, ZooKeeperNode.MY_ID).get();
+            String hostname = Entities.attributeSupplierWhenReady(member, ZooKeeperNode.HOSTNAME).get();
+            Integer port = Entities.attributeSupplierWhenReady(member, ZooKeeperNode.ZOOKEEPER_PORT).get();
+            Integer leaderPort = Entities.attributeSupplierWhenReady(member, ZooKeeperNode.ZOOKEEPER_LEADER_PORT).get();
+            Integer electionPort = Entities.attributeSupplierWhenReady(member, ZooKeeperNode.ZOOKEEPER_ELECTION_PORT).get();
+            result.add(new ZooKeeperServerConfig(myid, hostname, port, leaderPort, electionPort));
+        }
+        return result;
+    }
+
+    @Override
+    public Integer getZooKeeperPort() {
+        return getEntity().getAttribute(ZooKeeperNode.ZOOKEEPER_PORT);
+    }
+
+    @Override
+    public boolean isRunning() {
+        return newScript(MutableMap.of(USE_PID_FILE, getPidFile()), CHECK_RUNNING).execute() == 0;
+    }
+
+    @Override
+    public void stop() {
+        newScript(ImmutableMap.of(USE_PID_FILE, getPidFile()), STOPPING).execute();     
+    }
+
+    @Override
+    public void preInstall() {
+        resolver = Entities.newDownloader(this);
+        setExpandedInstallDir(Os.mergePaths(getInstallDir(), resolver.getUnpackedDirectoryName(format("zookeeper-%s", getVersion()))));
+    }
+
+    @Override
+    public void install() {
+        List<String> urls = resolver.getTargets();
+        String saveAs = resolver.getFilename();
+
+        List<String> commands = ImmutableList.<String> builder()
+                .addAll(BashCommands.commandsToDownloadUrlsAs(urls, saveAs))
+                .add(BashCommands.INSTALL_TAR)
+                .add("tar xzfv " + saveAs)
+                .build();
+
+        newScript(INSTALLING)
+                .body.append(commands)
+                .execute();
+    }
+
+    @Override
+    public void customize() {
+        log.debug("Customizing {}", entity);
+        Networking.checkPortsValid(getPortMap());
+        newScript(CUSTOMIZING)
+                .body.append(
+                        format("cp -R %s/* .", getExpandedInstallDir()),
+                        format("mkdir %s/zookeeper", getRunDir()),
+                        format("echo %d > %s/zookeeper/myid", getMyId(), getRunDir())
+                    )
+                .execute();
+
+        String destinationConfigFile = Os.mergePathsUnix(getRunDir(), "conf/zoo.cfg");
+        copyTemplate(getConfigFileName(), destinationConfigFile);
+    }
+
+    public String getPidFile() { return Os.mergePathsUnix(getRunDir(), "zookeeper.pid"); }
+
+    @Override
+    public void launch() {
+        newScript(MutableMap.of(USE_PID_FILE, getPidFile()), LAUNCHING)
+                .body.append(format("nohup java $JAVA_OPTS -cp zookeeper-%s.jar:lib/*:conf org.apache.zookeeper.server.quorum.QuorumPeerMain conf/zoo.cfg > %s 2>&1 &", getVersion(), getLogFileLocation()))
+                .execute();
+    }
+
+    public static class ZooKeeperServerConfig {
+        private final Integer myid;
+        private final String hostname;
+        private final Integer port;
+        private final Integer leaderPort;
+        private final Integer electionPort;
+
+        public ZooKeeperServerConfig(Integer myid, String hostname, Integer port, Integer leaderPort, Integer electionPort) {
+            this.myid = myid;
+            this.hostname = hostname;
+            this.port = port;
+            this.leaderPort = leaderPort;
+            this.electionPort = electionPort;
+        }
+
+        public Integer getMyid() { return myid; }
+        public String getHostname() { return hostname; }
+        public Integer getPort() { return port; }
+        public Integer getLeaderPort() { return leaderPort; }
+        public Integer getElectionPort() { return electionPort; }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/activemq/activemq.xml
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/activemq/activemq.xml b/software/messaging/src/main/resources/brooklyn/entity/messaging/activemq/activemq.xml
deleted file mode 100644
index 52114d1..0000000
--- a/software/messaging/src/main/resources/brooklyn/entity/messaging/activemq/activemq.xml
+++ /dev/null
@@ -1,154 +0,0 @@
-[#ftl]
-<?xml version="1.0" encoding="UTF-8"?>
-<!-- Based on standard file from ActiveMQ Version 5.7.0 -->
-<!-- START SNIPPET: example -->
-<beans
-  xmlns="http://www.springframework.org/schema/beans"
-  xmlns:amq="http://activemq.apache.org/schema/core"
-  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
-
-    <!-- Allows us to use system properties as variables in this configuration file -->
-    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
-        <property name="locations">
-            <value>file:[#noparse]${activemq.conf}[/#noparse]/credentials.properties</value>
-        </property>
-    </bean>
-
-    <!--
-        The <broker> element is used to configure the ActiveMQ broker.
-    -->
-    <broker useJmx="true" xmlns="http://activemq.apache.org/schema/core" brokerName="${entity.brokerName}" dataDirectory="[#noparse]${activemq.data}[/#noparse]">
-
-        <!--
-            For better performances use VM cursor and small memory limit.
-            For more information, see:
-
-            http://activemq.apache.org/message-cursors.html
-
-            Also, if your producer is "hanging", it's probably due to producer flow control.
-            For more information, see:
-            http://activemq.apache.org/producer-flow-control.html
-        -->
-
-        <destinationPolicy>
-            <policyMap>
-              <policyEntries>
-                <policyEntry topic=">" producerFlowControl="true">
-                    <!-- The constantPendingMessageLimitStrategy is used to prevent
-                         slow topic consumers to block producers and affect other consumers
-                         by limiting the number of messages that are retained
-                         For more information, see:
-
-                         http://activemq.apache.org/slow-consumer-handling.html
-
-                    -->
-                  <pendingMessageLimitStrategy>
-                    <constantPendingMessageLimitStrategy limit="1000"/>
-                  </pendingMessageLimitStrategy>
-                </policyEntry>
-                <policyEntry queue=">" producerFlowControl="true" memoryLimit="1mb">
-                  <!-- Use VM cursor for better latency
-                       For more information, see:
-
-                       http://activemq.apache.org/message-cursors.html
-
-                  <pendingQueuePolicy>
-                    <vmQueueCursor/>
-                  </pendingQueuePolicy>
-                  -->
-                </policyEntry>
-              </policyEntries>
-            </policyMap>
-        </destinationPolicy>
-
-
-        <!--
-            The managementContext is used to configure how ActiveMQ is exposed in
-            JMX. By default, ActiveMQ uses the MBean server that is started by
-            the JVM. For more information, see:
-
-            http://activemq.apache.org/jmx.html
-        -->
-        <managementContext>
-            [#if entity.jmxPort > 0]
-            <managementContext connectorPort="${entity.jmxPort?c}"/>
-            [#else]
-            <managementContext createConnector="false"/>
-            [/#if]
-        </managementContext>
-
-        <!--
-            Configure message persistence for the broker. The default persistence
-            mechanism is the KahaDB store (identified by the kahaDB tag).
-            For more information, see:
-
-            http://activemq.apache.org/persistence.html
-        -->
-        <persistenceAdapter>
-            <kahaDB directory="[#noparse]${activemq.data}[/#noparse]/kahadb"/>
-        </persistenceAdapter>
-
-
-          <!--
-            The systemUsage controls the maximum amount of space the broker will
-            use before slowing down producers. For more information, see:
-            http://activemq.apache.org/producer-flow-control.html
-            If using ActiveMQ embedded - the following limits could safely be used:
-
-        <systemUsage>
-            <systemUsage>
-                <memoryUsage>
-                    <memoryUsage limit="20 mb"/>
-                </memoryUsage>
-                <storeUsage>
-                    <storeUsage limit="1 gb"/>
-                </storeUsage>
-                <tempUsage>
-                    <tempUsage limit="100 mb"/>
-                </tempUsage>
-            </systemUsage>
-        </systemUsage>
-        -->
-          <systemUsage>
-            <systemUsage>
-                <memoryUsage>
-                    <memoryUsage limit="64 mb"/>
-                </memoryUsage>
-                <storeUsage>
-                    <storeUsage limit="100 gb"/>
-                </storeUsage>
-                <tempUsage>
-                    <tempUsage limit="50 gb"/>
-                </tempUsage>
-            </systemUsage>
-        </systemUsage>
-
-        <!--
-            The transport connectors expose ActiveMQ over a given protocol to
-            clients and other brokers. For more information, see:
-
-            http://activemq.apache.org/configuring-transports.html
-        -->
-        <transportConnectors>
-            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
-            <transportConnector name="openwire" uri="tcp://0.0.0.0:${entity.openWirePort?c}?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600"/>
-        </transportConnectors>
-
-        <!-- destroy the spring context on shutdown to stop jetty -->
-        <shutdownHooks>
-            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
-        </shutdownHooks>
-
-    </broker>
-
-    <!--
-        Enable web consoles, REST and Ajax APIs and demos
-
-        Take a look at [#noparse]${ACTIVEMQ_HOME}[/#noparse]/conf/jetty.xml for more details
-    -->
-    <import resource="jetty.xml"/>
-
-</beans>
-<!-- END SNIPPET: example -->

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg b/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg
deleted file mode 100644
index d600ef5..0000000
Binary files a/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg and /dev/null differ

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/server.properties
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/server.properties b/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/server.properties
deleted file mode 100644
index feb871f..0000000
--- a/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/server.properties
+++ /dev/null
@@ -1,112 +0,0 @@
-[#ftl]
-#
-##
-# KafkaBroker configuration template for Brooklyn
-#
-# see kafka.server.KafkaConfig for additional details and defaults
-##
-
-############################# Server Basics #############################
-# The id of the broker. This must be set to a unique integer for each broker.
-broker.id=${entity.brokerId?c}
-
-############################# Socket Server Settings #############################
-
-# The port the socket server listens on
-port=${entity.kafkaPort?c}
-
-# Hostname the broker will bind to. If not set, the server will bind to all interfaces
-host.name=${driver.hostname}
-
-# Hostname the broker will advertise to producers and consumers. If not set, it uses the
-# value for "host.name" if configured.  Otherwise, it will use the value returned from
-# java.net.InetAddress.getCanonicalHostName().
-#advertised.host.name=<hostname routable by clients>
-
-# The port to publish to ZooKeeper for clients to use. If this is not set,
-# it will publish the same port that the broker binds to.
-#advertised.port=<port accessible by clients>
-
-# The number of threads handling network requests
-num.network.threads=3
- 
-# The number of threads doing disk I/O
-num.io.threads=8
-
-# The send buffer (SO_SNDBUF) used by the socket server
-socket.send.buffer.bytes=102400
-
-# The receive buffer (SO_RCVBUF) used by the socket server
-socket.receive.buffer.bytes=102400
-
-# The maximum size of a request that the socket server will accept (protection against OOM)
-max.socket.request.bytes=104857600
-
-
-############################# Log Basics #############################
-
-# The directory under which to store log files
-log.dir=${driver.runDir}/kafka-logs
-
-# The default number of log partitions per topic. More partitions allow greater
-# parallelism for consumption, but this will also result in more files across
-# the brokers.
-num.partitions=1
-
-# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
-# This value is recommended to be increased for installations with data dirs located in RAID array.
-num.recovery.threads.per.data.dir=1
-
-############################# Log Flush Policy #############################
-
-# Messages are immediately written to the filesystem but by default we only fsync() to sync
-# the OS cache lazily. The following configurations control the flush of data to disk. 
-# There are a few important trade-offs here:
-#    1. Durability: Unflushed data may be lost if you are not using replication.
-#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
-#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. 
-# The settings below allow one to configure the flush policy to flush data after a period of time or
-# every N messages (or both). This can be done globally and overridden on a per-topic basis.
-
-# The number of messages to accept before forcing a flush of data to disk
-log.flush.interval.messages=10000
-
-# The maximum amount of time a message can sit in a log before we force a flush
-log.flush.interval.ms=1000
-
-############################# Log Retention Policy #############################
-
-# The following configurations control the disposal of log segments. The policy can
-# be set to delete segments after a period of time, or after a given size has accumulated.
-# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
-# from the end of the log.
-
-# The minimum age of a log file to be eligible for deletion
-log.retention.hours=168
-
-# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
-# segments don't drop below log.retention.bytes.
-#log.retention.bytes=1073741824
-
-# The maximum size of a log segment file. When this size is reached a new log segment will be created.
-log.segment.bytes=1073741824
-
-# The interval at which log segments are checked to see if they can be deleted according 
-# to the retention policies
-log.retention.check.interval.ms=300000
-
-# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
-# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
-log.cleaner.enable=false
-
-############################# Zookeeper #############################
-
-# Zookeeper connection string (see zookeeper docs for details).
-# This is a comma separated host:port pairs, each corresponding to a zk
-# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
-# You can also append an optional chroot string to the urls to specify the
-# root directory for all kafka znodes.
-zookeeper.connect=${entity.zookeeper.hostname}:${entity.zookeeper.zookeeperPort?c}
-
-# Timeout in ms for connecting to zookeeper
-zookeeper.connection.timeout.ms=1000000
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/zookeeper.properties
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/zookeeper.properties b/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/zookeeper.properties
deleted file mode 100644
index 646d2f1..0000000
--- a/software/messaging/src/main/resources/brooklyn/entity/messaging/kafka/zookeeper.properties
+++ /dev/null
@@ -1,13 +0,0 @@
-[#ftl]
-#
-
-##
-# KafkaZookeeper configuration template for Brooklyn
-##
-
-# the directory where the snapshot is stored.
-dataDir=${driver.runDir}/zookeeper
-# the port at which the clients will connect
-clientPort=${entity.zookeeperPort?c}
-# disable the per-ip limit on the number of connections since this is a non-production config
-maxClientCnxns=0

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/rabbit/rabbitmq.config
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/rabbit/rabbitmq.config b/software/messaging/src/main/resources/brooklyn/entity/messaging/rabbit/rabbitmq.config
deleted file mode 100644
index b4428f0..0000000
--- a/software/messaging/src/main/resources/brooklyn/entity/messaging/rabbit/rabbitmq.config
+++ /dev/null
@@ -1,5 +0,0 @@
-[
-<#if entity.enableManagementPlugin>
-    {rabbitmq_mochiweb, [{listeners, [{mgmt, [{port, ${entity.managementPort?c}}]}]}]}
-</#if>
-].
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/storm/storm.yaml
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/storm/storm.yaml b/software/messaging/src/main/resources/brooklyn/entity/messaging/storm/storm.yaml
deleted file mode 100644
index 99f0c71..0000000
--- a/software/messaging/src/main/resources/brooklyn/entity/messaging/storm/storm.yaml
+++ /dev/null
@@ -1,39 +0,0 @@
-[#ftl]
-#
-# Storm Configuration
-[#if driver.zookeeperServers?has_content]
- storm.zookeeper.servers:
-[#list driver.zookeeperServers as zkServer]
-   - "${zkServer}"
-[/#list]
-[/#if]
-
- storm.local.dir: "${driver.localDir}"
-
-### ui.* configs are for the master
- ui.port: ${driver.uiPort?c}
- ui.childopts: "-Xmx768m"
-
-[#if driver.roleName == "ui"]
- nimbus.host: "${driver.nimbusHostname}"
-[/#if]
-
- nimbus.childopts: " ${driver.jvmOptsLine}"
- worker.childopts: " ${driver.jvmOptsLine}"
- supervisor.childopts: " ${driver.jvmOptsLine}" 
-  
-# ##### These may optionally be filled in:
-#    
-## List of custom serializations
-# topology.kryo.register:
-#     - org.mycompany.MyType
-#     - org.mycompany.MyType2: org.mycompany.MyType2Serializer
-#
-## List of custom kryo decorators
-# topology.kryo.decorators:
-#     - org.mycompany.MyDecorator
-#
-## Locations of the drpc servers
-# drpc.servers:
-#     - "server1"
-#     - "server2"

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/brooklyn/entity/messaging/zookeeper/zoo.cfg
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/brooklyn/entity/messaging/zookeeper/zoo.cfg b/software/messaging/src/main/resources/brooklyn/entity/messaging/zookeeper/zoo.cfg
deleted file mode 100644
index 79721a6..0000000
--- a/software/messaging/src/main/resources/brooklyn/entity/messaging/zookeeper/zoo.cfg
+++ /dev/null
@@ -1,42 +0,0 @@
-[#ftl]
-#
-# 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.
-#
-
-##
-# ZooKeeper configuration template for Brooklyn
-##
-
-# The number of milliseconds of each tick
-tickTime=2000
-# The number of ticks that the initial
-# synchronization phase can take
-initLimit=10
-# The number of ticks that can pass between
-# sending a request and getting an acknowledgement
-syncLimit=5
-# the directory where the snapshot is stored.
-dataDir=${driver.runDir}/zookeeper
-# the port at which the clients will connect
-clientPort=${entity.zookeeperPort?c}
-# disable the per-ip limit on the number of connections since this is a non-production config
-maxClientCnxns=0
-
-[#list driver.zookeeperServers as zkServer]
-server.${zkServer.myid?c}=${zkServer.hostname}:${zkServer.leaderPort?c}:${zkServer.electionPort?c}
-[/#list]

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/activemq/activemq.xml
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/activemq/activemq.xml b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/activemq/activemq.xml
new file mode 100644
index 0000000..52114d1
--- /dev/null
+++ b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/activemq/activemq.xml
@@ -0,0 +1,154 @@
+[#ftl]
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- Based on standard file from ActiveMQ Version 5.7.0 -->
+<!-- START SNIPPET: example -->
+<beans
+  xmlns="http://www.springframework.org/schema/beans"
+  xmlns:amq="http://activemq.apache.org/schema/core"
+  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+  http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+    <!-- Allows us to use system properties as variables in this configuration file -->
+    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
+        <property name="locations">
+            <value>file:[#noparse]${activemq.conf}[/#noparse]/credentials.properties</value>
+        </property>
+    </bean>
+
+    <!--
+        The <broker> element is used to configure the ActiveMQ broker.
+    -->
+    <broker useJmx="true" xmlns="http://activemq.apache.org/schema/core" brokerName="${entity.brokerName}" dataDirectory="[#noparse]${activemq.data}[/#noparse]">
+
+        <!--
+            For better performances use VM cursor and small memory limit.
+            For more information, see:
+
+            http://activemq.apache.org/message-cursors.html
+
+            Also, if your producer is "hanging", it's probably due to producer flow control.
+            For more information, see:
+            http://activemq.apache.org/producer-flow-control.html
+        -->
+
+        <destinationPolicy>
+            <policyMap>
+              <policyEntries>
+                <policyEntry topic=">" producerFlowControl="true">
+                    <!-- The constantPendingMessageLimitStrategy is used to prevent
+                         slow topic consumers to block producers and affect other consumers
+                         by limiting the number of messages that are retained
+                         For more information, see:
+
+                         http://activemq.apache.org/slow-consumer-handling.html
+
+                    -->
+                  <pendingMessageLimitStrategy>
+                    <constantPendingMessageLimitStrategy limit="1000"/>
+                  </pendingMessageLimitStrategy>
+                </policyEntry>
+                <policyEntry queue=">" producerFlowControl="true" memoryLimit="1mb">
+                  <!-- Use VM cursor for better latency
+                       For more information, see:
+
+                       http://activemq.apache.org/message-cursors.html
+
+                  <pendingQueuePolicy>
+                    <vmQueueCursor/>
+                  </pendingQueuePolicy>
+                  -->
+                </policyEntry>
+              </policyEntries>
+            </policyMap>
+        </destinationPolicy>
+
+
+        <!--
+            The managementContext is used to configure how ActiveMQ is exposed in
+            JMX. By default, ActiveMQ uses the MBean server that is started by
+            the JVM. For more information, see:
+
+            http://activemq.apache.org/jmx.html
+        -->
+        <managementContext>
+            [#if entity.jmxPort > 0]
+            <managementContext connectorPort="${entity.jmxPort?c}"/>
+            [#else]
+            <managementContext createConnector="false"/>
+            [/#if]
+        </managementContext>
+
+        <!--
+            Configure message persistence for the broker. The default persistence
+            mechanism is the KahaDB store (identified by the kahaDB tag).
+            For more information, see:
+
+            http://activemq.apache.org/persistence.html
+        -->
+        <persistenceAdapter>
+            <kahaDB directory="[#noparse]${activemq.data}[/#noparse]/kahadb"/>
+        </persistenceAdapter>
+
+
+          <!--
+            The systemUsage controls the maximum amount of space the broker will
+            use before slowing down producers. For more information, see:
+            http://activemq.apache.org/producer-flow-control.html
+            If using ActiveMQ embedded - the following limits could safely be used:
+
+        <systemUsage>
+            <systemUsage>
+                <memoryUsage>
+                    <memoryUsage limit="20 mb"/>
+                </memoryUsage>
+                <storeUsage>
+                    <storeUsage limit="1 gb"/>
+                </storeUsage>
+                <tempUsage>
+                    <tempUsage limit="100 mb"/>
+                </tempUsage>
+            </systemUsage>
+        </systemUsage>
+        -->
+          <systemUsage>
+            <systemUsage>
+                <memoryUsage>
+                    <memoryUsage limit="64 mb"/>
+                </memoryUsage>
+                <storeUsage>
+                    <storeUsage limit="100 gb"/>
+                </storeUsage>
+                <tempUsage>
+                    <tempUsage limit="50 gb"/>
+                </tempUsage>
+            </systemUsage>
+        </systemUsage>
+
+        <!--
+            The transport connectors expose ActiveMQ over a given protocol to
+            clients and other brokers. For more information, see:
+
+            http://activemq.apache.org/configuring-transports.html
+        -->
+        <transportConnectors>
+            <!-- DOS protection, limit concurrent connections to 1000 and frame size to 100MB -->
+            <transportConnector name="openwire" uri="tcp://0.0.0.0:${entity.openWirePort?c}?maximumConnections=1000&amp;wireformat.maxFrameSize=104857600"/>
+        </transportConnectors>
+
+        <!-- destroy the spring context on shutdown to stop jetty -->
+        <shutdownHooks>
+            <bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
+        </shutdownHooks>
+
+    </broker>
+
+    <!--
+        Enable web consoles, REST and Ajax APIs and demos
+
+        Take a look at [#noparse]${ACTIVEMQ_HOME}[/#noparse]/conf/jetty.xml for more details
+    -->
+    <import resource="jetty.xml"/>
+
+</beans>
+<!-- END SNIPPET: example -->

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg
new file mode 100644
index 0000000..d600ef5
Binary files /dev/null and b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/kafka-google-doorway.jpg differ

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/server.properties
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/server.properties b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/server.properties
new file mode 100644
index 0000000..feb871f
--- /dev/null
+++ b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/server.properties
@@ -0,0 +1,112 @@
+[#ftl]
+#
+##
+# KafkaBroker configuration template for Brooklyn
+#
+# see kafka.server.KafkaConfig for additional details and defaults
+##
+
+############################# Server Basics #############################
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id=${entity.brokerId?c}
+
+############################# Socket Server Settings #############################
+
+# The port the socket server listens on
+port=${entity.kafkaPort?c}
+
+# Hostname the broker will bind to. If not set, the server will bind to all interfaces
+host.name=${driver.hostname}
+
+# Hostname the broker will advertise to producers and consumers. If not set, it uses the
+# value for "host.name" if configured.  Otherwise, it will use the value returned from
+# java.net.InetAddress.getCanonicalHostName().
+#advertised.host.name=<hostname routable by clients>
+
+# The port to publish to ZooKeeper for clients to use. If this is not set,
+# it will publish the same port that the broker binds to.
+#advertised.port=<port accessible by clients>
+
+# The number of threads handling network requests
+num.network.threads=3
+ 
+# The number of threads doing disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=102400
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=102400
+
+# The maximum size of a request that the socket server will accept (protection against OOM)
+max.socket.request.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# The directory under which to store log files
+log.dir=${driver.runDir}/kafka-logs
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions=1
+
+# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
+# This value is recommended to be increased for installations with data dirs located in RAID array.
+num.recovery.threads.per.data.dir=1
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data to disk. 
+# There are a few important trade-offs here:
+#    1. Durability: Unflushed data may be lost if you are not using replication.
+#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
+#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. 
+# The settings below allow one to configure the flush policy to flush data after a period of time or
+# every N messages (or both). This can be done globally and overridden on a per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+log.flush.interval.ms=1000
+
+############################# Log Retention Policy #############################
+
+# The following configurations control the disposal of log segments. The policy can
+# be set to delete segments after a period of time, or after a given size has accumulated.
+# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
+# segments don't drop below log.retention.bytes.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log segment will be created.
+log.segment.bytes=1073741824
+
+# The interval at which log segments are checked to see if they can be deleted according 
+# to the retention policies
+log.retention.check.interval.ms=300000
+
+# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
+# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
+log.cleaner.enable=false
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect=${entity.zookeeper.hostname}:${entity.zookeeper.zookeeperPort?c}
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=1000000
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/zookeeper.properties
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/zookeeper.properties b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/zookeeper.properties
new file mode 100644
index 0000000..646d2f1
--- /dev/null
+++ b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/kafka/zookeeper.properties
@@ -0,0 +1,13 @@
+[#ftl]
+#
+
+##
+# KafkaZookeeper configuration template for Brooklyn
+##
+
+# the directory where the snapshot is stored.
+dataDir=${driver.runDir}/zookeeper
+# the port at which the clients will connect
+clientPort=${entity.zookeeperPort?c}
+# disable the per-ip limit on the number of connections since this is a non-production config
+maxClientCnxns=0

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/rabbit/rabbitmq.config
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/rabbit/rabbitmq.config b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/rabbit/rabbitmq.config
new file mode 100644
index 0000000..b4428f0
--- /dev/null
+++ b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/rabbit/rabbitmq.config
@@ -0,0 +1,5 @@
+[
+<#if entity.enableManagementPlugin>
+    {rabbitmq_mochiweb, [{listeners, [{mgmt, [{port, ${entity.managementPort?c}}]}]}]}
+</#if>
+].
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/storm/storm.yaml
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/storm/storm.yaml b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/storm/storm.yaml
new file mode 100644
index 0000000..99f0c71
--- /dev/null
+++ b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/storm/storm.yaml
@@ -0,0 +1,39 @@
+[#ftl]
+#
+# Storm Configuration
+[#if driver.zookeeperServers?has_content]
+ storm.zookeeper.servers:
+[#list driver.zookeeperServers as zkServer]
+   - "${zkServer}"
+[/#list]
+[/#if]
+
+ storm.local.dir: "${driver.localDir}"
+
+### ui.* configs are for the master
+ ui.port: ${driver.uiPort?c}
+ ui.childopts: "-Xmx768m"
+
+[#if driver.roleName == "ui"]
+ nimbus.host: "${driver.nimbusHostname}"
+[/#if]
+
+ nimbus.childopts: " ${driver.jvmOptsLine}"
+ worker.childopts: " ${driver.jvmOptsLine}"
+ supervisor.childopts: " ${driver.jvmOptsLine}" 
+  
+# ##### These may optionally be filled in:
+#    
+## List of custom serializations
+# topology.kryo.register:
+#     - org.mycompany.MyType
+#     - org.mycompany.MyType2: org.mycompany.MyType2Serializer
+#
+## List of custom kryo decorators
+# topology.kryo.decorators:
+#     - org.mycompany.MyDecorator
+#
+## Locations of the drpc servers
+# drpc.servers:
+#     - "server1"
+#     - "server2"

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/zookeeper/zoo.cfg
----------------------------------------------------------------------
diff --git a/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/zookeeper/zoo.cfg b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/zookeeper/zoo.cfg
new file mode 100644
index 0000000..79721a6
--- /dev/null
+++ b/software/messaging/src/main/resources/org/apache/brooklyn/entity/messaging/zookeeper/zoo.cfg
@@ -0,0 +1,42 @@
+[#ftl]
+#
+# 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.
+#
+
+##
+# ZooKeeper configuration template for Brooklyn
+##
+
+# The number of milliseconds of each tick
+tickTime=2000
+# The number of ticks that the initial
+# synchronization phase can take
+initLimit=10
+# The number of ticks that can pass between
+# sending a request and getting an acknowledgement
+syncLimit=5
+# the directory where the snapshot is stored.
+dataDir=${driver.runDir}/zookeeper
+# the port at which the clients will connect
+clientPort=${entity.zookeeperPort?c}
+# disable the per-ip limit on the number of connections since this is a non-production config
+maxClientCnxns=0
+
+[#list driver.zookeeperServers as zkServer]
+server.${zkServer.myid?c}=${zkServer.hostname}:${zkServer.leaderPort?c}:${zkServer.electionPort?c}
+[/#list]

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQEc2LiveTest.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQEc2LiveTest.java b/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQEc2LiveTest.java
deleted file mode 100644
index aaffda8..0000000
--- a/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQEc2LiveTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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 brooklyn.entity.messaging.activemq;
-
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotNull;
-
-import javax.jms.Connection;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.brooklyn.api.entity.proxying.EntitySpec;
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.test.EntityTestUtils;
-import org.testng.annotations.Test;
-
-import brooklyn.entity.AbstractEc2LiveTest;
-import brooklyn.entity.trait.Startable;
-
-import com.google.common.collect.ImmutableList;
-
-public class ActiveMQEc2LiveTest extends AbstractEc2LiveTest {
-
-    /**
-     * Test that can install+start, and use, ActiveMQ.
-     */
-    @Override
-    protected void doTest(Location loc) throws Exception {
-        String queueName = "testQueue";
-        int number = 10;
-        String content = "01234567890123456789012345678901";
-
-        // Start broker with a configured queue
-        ActiveMQBroker activeMQ = app.createAndManageChild(EntitySpec.create(ActiveMQBroker.class).configure("queue", queueName));
-        
-        app.start(ImmutableList.of(loc));
-        
-        EntityTestUtils.assertAttributeEqualsEventually(activeMQ, Startable.SERVICE_UP, true);
-
-        // Check queue created
-        assertEquals(ImmutableList.copyOf(activeMQ.getQueueNames()), ImmutableList.of(queueName));
-        assertEquals(activeMQ.getChildren().size(), 1);
-        assertEquals(activeMQ.getQueues().size(), 1);
-
-        // Get the named queue entity
-        ActiveMQQueue queue = activeMQ.getQueues().get(queueName);
-        assertNotNull(queue);
-
-        // Connect to broker using JMS and send messages
-        Connection connection = getActiveMQConnection(activeMQ);
-        clearQueue(connection, queueName);
-        EntityTestUtils.assertAttributeEqualsEventually(queue, ActiveMQQueue.QUEUE_DEPTH_MESSAGES, 0);
-        sendMessages(connection, number, queueName, content);
-
-        // Check messages arrived
-        EntityTestUtils.assertAttributeEqualsEventually(queue, ActiveMQQueue.QUEUE_DEPTH_MESSAGES, number);
-
-        connection.close();
-    }
-
-    private Connection getActiveMQConnection(ActiveMQBroker activeMQ) throws Exception {
-        int port = activeMQ.getAttribute(ActiveMQBroker.OPEN_WIRE_PORT);
-        String address = activeMQ.getAttribute(ActiveMQBroker.ADDRESS);
-        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(String.format("tcp://%s:%s", address, port));
-        Connection connection = factory.createConnection("admin", "activemq");
-        connection.start();
-        return connection;
-    }
-
-    private void sendMessages(Connection connection, int count, String queueName, String content) throws Exception {
-        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        org.apache.activemq.command.ActiveMQQueue destination = (org.apache.activemq.command.ActiveMQQueue) session.createQueue(queueName);
-        MessageProducer messageProducer = session.createProducer(destination);
-
-        for (int i = 0; i < count; i++) {
-            TextMessage message = session.createTextMessage(content);
-            messageProducer.send(message);
-        }
-
-        session.close();
-    }
-
-    private int clearQueue(Connection connection, String queueName) throws Exception {
-        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        org.apache.activemq.command.ActiveMQQueue destination = (org.apache.activemq.command.ActiveMQQueue) session.createQueue(queueName);
-        MessageConsumer messageConsumer = session.createConsumer(destination);
-
-        int received = 0;
-        while (messageConsumer.receive(500) != null) received++;
-
-        session.close();
-
-        return received;
-    }
-    
-    @Test(enabled=false)
-    public void testDummy() {} // Convince testng IDE integration that this really does have test methods  
-}

http://git-wip-us.apache.org/repos/asf/incubator-brooklyn/blob/c14fef53/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQGoogleComputeLiveTest.java
----------------------------------------------------------------------
diff --git a/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQGoogleComputeLiveTest.java b/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQGoogleComputeLiveTest.java
deleted file mode 100644
index e26dc2d..0000000
--- a/software/messaging/src/test/java/brooklyn/entity/messaging/activemq/ActiveMQGoogleComputeLiveTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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 brooklyn.entity.messaging.activemq;
-
-import brooklyn.entity.AbstractGoogleComputeLiveTest;
-import brooklyn.entity.trait.Startable;
-
-import com.google.common.collect.ImmutableList;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.brooklyn.api.entity.proxying.EntitySpec;
-import org.apache.brooklyn.api.location.Location;
-import org.apache.brooklyn.test.EntityTestUtils;
-import org.testng.annotations.Test;
-
-import javax.jms.Connection;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotNull;
-
-public class ActiveMQGoogleComputeLiveTest extends AbstractGoogleComputeLiveTest {
-
-    /**
-     * Test that can install+start, and use, ActiveMQ.
-     */
-    @Override
-    protected void doTest(Location loc) throws Exception {
-        String queueName = "testQueue";
-        int number = 10;
-        String content = "01234567890123456789012345678901";
-
-        // Start broker with a configured queue
-        ActiveMQBroker activeMQ = app.createAndManageChild(EntitySpec.create(ActiveMQBroker.class).configure("queue", queueName));
-        
-        app.start(ImmutableList.of(loc));
-        
-        EntityTestUtils.assertAttributeEqualsEventually(activeMQ, Startable.SERVICE_UP, true);
-
-        // Check queue created
-        assertEquals(ImmutableList.copyOf(activeMQ.getQueueNames()), ImmutableList.of(queueName));
-        assertEquals(activeMQ.getChildren().size(), 1);
-        assertEquals(activeMQ.getQueues().size(), 1);
-
-        // Get the named queue entity
-        ActiveMQQueue queue = activeMQ.getQueues().get(queueName);
-        assertNotNull(queue);
-
-        // Connect to broker using JMS and send messages
-        Connection connection = getActiveMQConnection(activeMQ);
-        clearQueue(connection, queueName);
-        EntityTestUtils.assertAttributeEqualsEventually(queue, ActiveMQQueue.QUEUE_DEPTH_MESSAGES, 0);
-        sendMessages(connection, number, queueName, content);
-
-        // Check messages arrived
-        EntityTestUtils.assertAttributeEqualsEventually(queue, ActiveMQQueue.QUEUE_DEPTH_MESSAGES, number);
-
-        connection.close();
-    }
-
-    private Connection getActiveMQConnection(ActiveMQBroker activeMQ) throws Exception {
-        int port = activeMQ.getAttribute(ActiveMQBroker.OPEN_WIRE_PORT);
-        String address = activeMQ.getAttribute(ActiveMQBroker.ADDRESS);
-        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(String.format("tcp://%s:%s", address, port));
-        Connection connection = factory.createConnection("admin", "activemq");
-        connection.start();
-        return connection;
-    }
-
-    private void sendMessages(Connection connection, int count, String queueName, String content) throws Exception {
-        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        org.apache.activemq.command.ActiveMQQueue destination = (org.apache.activemq.command.ActiveMQQueue) session.createQueue(queueName);
-        MessageProducer messageProducer = session.createProducer(destination);
-
-        for (int i = 0; i < count; i++) {
-            TextMessage message = session.createTextMessage(content);
-            messageProducer.send(message);
-        }
-
-        session.close();
-    }
-
-    private int clearQueue(Connection connection, String queueName) throws Exception {
-        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        org.apache.activemq.command.ActiveMQQueue destination = (org.apache.activemq.command.ActiveMQQueue) session.createQueue(queueName);
-        MessageConsumer messageConsumer = session.createConsumer(destination);
-
-        int received = 0;
-        while (messageConsumer.receive(500) != null) received++;
-
-        session.close();
-
-        return received;
-    }
-    
-    @Test(enabled=false)
-    public void testDummy() {} // Convince testng IDE integration that this really does have test methods  
-}


Mime
View raw message