helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kisho...@apache.org
Subject [04/15] Adding Helix-task-framework and Yarn integration modules
Date Fri, 20 Sep 2013 18:30:13 GMT
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProcess.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProcess.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProcess.java
new file mode 100644
index 0000000..9ea713c
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProcess.java
@@ -0,0 +1,82 @@
+package org.apache.helix.metamanager.provider;
+
+import java.util.Properties;
+
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.manager.zk.ZKHelixAdmin;
+import org.apache.helix.metamanager.ClusterAdmin;
+import org.apache.helix.metamanager.ContainerProvider;
+import org.apache.helix.metamanager.HelixClusterAdmin;
+import org.apache.helix.metamanager.Service;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Helix participant for ContainerProvider. Configurable via ProviderProperties
+ * and runnable service.
+ * 
+ */
+public class ProviderProcess implements Service {
+    static final Logger log = Logger.getLogger(ProviderProcess.class);
+
+    ClusterAdmin        admin;
+
+    ProviderProperties  properties;
+    ContainerProvider   provider;
+    HelixAdmin          helixAdmin;
+    HelixManager        participantManager;
+
+    @Override
+    public void configure(Properties properties) throws Exception {
+        Preconditions.checkNotNull(properties);
+        ProviderProperties providerProperties = new ProviderProperties();
+        providerProperties.putAll(properties);
+        Preconditions.checkArgument(providerProperties.isValid());
+
+        this.properties = providerProperties;
+
+    }
+
+    public void setConteinerProvider(ContainerProvider provider) {
+        this.provider = provider;
+    }
+
+    @Override
+    public void start() throws Exception {
+        Preconditions.checkNotNull(provider);
+
+        log.info(String.format("Registering provider '%s' at '%s/%s'", properties.getName(), properties.getMetaAddress(), properties.getMetaCluster()));
+        HelixAdmin metaHelixAdmin = new ZKHelixAdmin(properties.getMetaAddress());
+        metaHelixAdmin.addInstance(properties.getMetaCluster(), new InstanceConfig(properties.getName()));
+        metaHelixAdmin.close();
+
+        log.info(String.format("Starting provider '%s'", properties.getName()));
+        helixAdmin = new ZKHelixAdmin(properties.getAddress());
+        admin = new HelixClusterAdmin(properties.getCluster(), helixAdmin);
+
+        participantManager = HelixManagerFactory.getZKHelixManager(properties.getMetaCluster(), properties.getName(), InstanceType.PARTICIPANT,
+                properties.getMetaAddress());
+        participantManager.getStateMachineEngine().registerStateModelFactory("OnlineOffline", new ProviderStateModelFactory(provider, admin));
+        participantManager.connect();
+
+        log.info(String.format("Successfully started provider '%s'", properties.getName()));
+    }
+
+    @Override
+    public void stop() {
+        log.info(String.format("Stopping provider '%s'", properties.getName()));
+        if (participantManager != null) {
+            participantManager.disconnect();
+            participantManager = null;
+        }
+        if (helixAdmin != null) {
+            helixAdmin.close();
+            helixAdmin = null;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProperties.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProperties.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProperties.java
new file mode 100644
index 0000000..098592a
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderProperties.java
@@ -0,0 +1,97 @@
+package org.apache.helix.metamanager.provider;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.helix.metamanager.bootstrapper.BootUtils;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Base configuration for {@link ProviderProcess}. 
+ *
+ */
+public class ProviderProperties extends Properties {
+	/**
+	 * 
+	 */
+	private static final long serialVersionUID = -2209509977839674160L;
+	
+	public final static String ADDRESS = "address";
+	public final static String CLUSTER = "cluster";
+    public final static String METAADDRESS = "metaaddress";
+    public final static String METACLUSTER = "metacluster";
+	public final static String NAME = "name";
+	
+	public final static String CONTAINER_NAMESPACE = "containers";
+	
+	public boolean isValid() {
+		return(containsKey(ADDRESS) &&
+		       containsKey(CLUSTER) &&
+		       containsKey(METAADDRESS) &&
+               containsKey(METACLUSTER) &&
+               containsKey(NAME));
+	}
+	
+	public String getAddress() {
+		return getProperty(ADDRESS);
+	}
+	
+	public String getCluster() {
+	    return getProperty(CLUSTER);
+	}
+	
+    public String getMetaAddress() {
+        return getProperty(METAADDRESS);
+    }
+    
+    public String getMetaCluster() {
+        return getProperty(METACLUSTER);
+    }
+    
+	public String getName() {
+	    return getProperty(NAME);
+	}
+	
+	public Set<String> getContainers() {
+        if(!BootUtils.hasNamespace(this, CONTAINER_NAMESPACE))
+            return Collections.emptySet();
+	    return BootUtils.getNamespaces(BootUtils.getNamespace(this, CONTAINER_NAMESPACE));
+	}
+	
+	public boolean hasContainer(String id) {
+	    if(!BootUtils.hasNamespace(this, CONTAINER_NAMESPACE)) return false;
+	    if(!BootUtils.hasNamespace(BootUtils.getNamespace(this, CONTAINER_NAMESPACE), id)) return false;
+	    return true;
+	}
+	
+	public Properties getContainer(String id) {
+	    Preconditions.checkArgument(BootUtils.hasNamespace(this, CONTAINER_NAMESPACE), "no container namespace");
+        Preconditions.checkArgument(BootUtils.hasNamespace(BootUtils.getNamespace(this, CONTAINER_NAMESPACE), id), "container %s not configured", id);
+	    return BootUtils.getNamespace(BootUtils.getNamespace(this, CONTAINER_NAMESPACE), id);
+	}
+	
+	public void addContainer(String id, Properties properties) {
+	    Preconditions.checkArgument(!getContainers().contains(id), "Already contains container type %s", id);
+	    
+	    // add container config
+        for(Map.Entry<Object, Object> entry : properties.entrySet()) {
+            this.put(CONTAINER_NAMESPACE + "." + id + "." + entry.getKey(), entry.getValue());
+        }
+	}
+
+    @Override
+    public Object get(Object key) {
+        Preconditions.checkState(containsKey(key));
+        return super.get(key);
+    }
+    
+    @Override
+    public String getProperty(String key) {
+        Preconditions.checkState(containsKey(key));
+        return super.getProperty(key);
+    }
+    
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancer.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancer.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancer.java
new file mode 100644
index 0000000..4be1a05
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancer.java
@@ -0,0 +1,352 @@
+package org.apache.helix.metamanager.provider;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.controller.rebalancer.Rebalancer;
+import org.apache.helix.controller.stages.ClusterDataCache;
+import org.apache.helix.controller.stages.CurrentStateOutput;
+import org.apache.helix.metamanager.StatusProvider;
+import org.apache.helix.metamanager.TargetProvider;
+import org.apache.helix.model.IdealState;
+import org.apache.helix.model.LiveInstance;
+import org.apache.helix.model.Partition;
+import org.apache.helix.model.Resource;
+import org.apache.helix.model.ResourceAssignment;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
+
+/**
+ * Rebalancer for meta cluster. Polls {@link TargetProvider} and
+ * {@link StatusProvider} and reads and sets IdealState of meta cluster participants (
+ * {@link ProviderProcess}). The number of active container is set to the target
+ * count. Failed containers are shut down and restarted on any available
+ * provider. Also, container counts are balanced across multiple providers.<br/>
+ * <b>NOTE:</b> status and target provider are injected via
+ * {@link ProviderRebalancerSingleton}<br/>
+ * <br/>
+ * <b>IdealState mapping:</b><br/>
+ * resource = container type<br/>
+ * partition = logical container instance<br/>
+ * instance = container provider<br/>
+ * status = physical container instance presence<br/>
+ */
+public class ProviderRebalancer implements Rebalancer {
+
+    static final Logger log                 = Logger.getLogger(ProviderRebalancer.class);
+
+    static final long   UPDATE_INTERVAL_MIN = 1500;
+
+    static final Object lock                = new Object();
+    static long         nextUpdate          = 0;
+
+    TargetProvider      targetProvider;
+    StatusProvider      statusProvider;
+    HelixManager        manager;
+
+    @Override
+    public void init(HelixManager manager) {
+        this.targetProvider = ProviderRebalancerSingleton.getTargetProvider();
+        this.statusProvider = ProviderRebalancerSingleton.getStatusProvider();
+        this.manager = manager;
+    }
+
+    @Override
+    public ResourceAssignment computeResourceMapping(Resource resource, IdealState idealState, CurrentStateOutput currentStateOutput,
+            ClusterDataCache clusterData) {
+
+        final String resourceName = resource.getResourceName();
+        final String containerType = resourceName;
+
+        final SortedSet<String> allContainers = Sets.newTreeSet(new IndexedNameComparator());
+        allContainers.addAll(idealState.getPartitionSet());
+
+        final SortedSet<String> allProviders = Sets.newTreeSet(new IndexedNameComparator());
+        for (LiveInstance instance : clusterData.getLiveInstances().values()) {
+            allProviders.add(instance.getId());
+        }
+
+        final ResourceState currentState = new ResourceState(resourceName, currentStateOutput);
+
+        // target container count
+        log.debug(String.format("Retrieving target container count for type '%s'", containerType));
+        int targetCount = -1;
+        try {
+            targetCount = targetProvider.getTargetContainerCount(containerType);
+        } catch (Exception e) {
+            log.error(String.format("Could not retrieve target count for '%s'", containerType), e);
+            return new ResourceAssignment(resourceName);
+        }
+
+        // provider sanity check
+        if (allProviders.isEmpty()) {
+            log.warn(String.format("Could not find any providers"));
+            return new ResourceAssignment(resourceName);
+        }
+
+        // all containers
+        SortedSet<String> assignedContainers = getAssignedContainers(currentState, allContainers);
+        SortedSet<String> failedContainers = getFailedContainers(currentState, allContainers);
+
+        log.info(String.format("Rebalancing '%s' (target=%d, active=%d, failures=%d)", resourceName, targetCount, assignedContainers.size(),
+                failedContainers.size()));
+
+        if (log.isDebugEnabled()) {
+            log.debug(String.format("%s: assigned containers %s", resourceName, assignedContainers));
+            log.debug(String.format("%s: failed containers %s", resourceName, failedContainers));
+        }
+
+        // assignment
+        int maxCountPerProvider = (int) Math.ceil(targetCount / (float) allProviders.size());
+
+        ResourceAssignment assignment = new ResourceAssignment(resourceName);
+        CountMap counts = new CountMap(allProviders);
+        int assignmentCount = 0;
+
+        // currently assigned
+        for (String containerName : assignedContainers) {
+            String providerName = getProvider(currentState, containerName);
+            Partition partition = new Partition(containerName);
+
+            if (failedContainers.contains(containerName)) {
+                log.warn(String.format("Container '%s:%s' failed, going offline", providerName, containerName));
+                assignment.addReplicaMap(partition, Collections.singletonMap(providerName, "OFFLINE"));
+
+            } else if (counts.get(providerName) >= maxCountPerProvider) {
+                log.warn(String.format("Container '%s:%s' misassigned, going offline", providerName, containerName));
+                assignment.addReplicaMap(partition, Collections.singletonMap(providerName, "OFFLINE"));
+
+            } else {
+                assignment.addReplicaMap(partition, Collections.singletonMap(providerName, "ONLINE"));
+            }
+
+            counts.increment(providerName);
+            assignmentCount++;
+        }
+
+        // currently unassigned
+        SortedSet<String> unassignedContainers = Sets.newTreeSet(new IndexedNameComparator());
+        unassignedContainers.addAll(allContainers);
+        unassignedContainers.removeAll(assignedContainers);
+
+        for (String containerName : unassignedContainers) {
+            if (assignmentCount >= targetCount)
+                break;
+
+            String providerName = counts.getMinKey();
+            Partition partition = new Partition(containerName);
+
+            if (failedContainers.contains(containerName)) {
+                log.warn(String.format("Container '%s:%s' failed and unassigned, going offline", providerName, containerName));
+                assignment.addReplicaMap(partition, Collections.singletonMap(providerName, "OFFLINE"));
+
+            } else {
+                assignment.addReplicaMap(partition, Collections.singletonMap(providerName, "ONLINE"));
+            }
+
+            counts.increment(providerName);
+            assignmentCount++;
+        }
+
+        if (log.isDebugEnabled()) {
+            log.debug(String.format("assignment counts: %s", counts));
+            log.debug(String.format("assignment: %s", assignment));
+        }
+
+        return assignment;
+    }
+
+    boolean hasProvider(ResourceState state, String containerName) {
+        Map<String, String> currentStateMap = state.getCurrentStateMap(containerName);
+        Map<String, String> pendingStateMap = state.getPendingStateMap(containerName);
+        return hasInstance(currentStateMap, "ONLINE") || hasInstance(pendingStateMap, "ONLINE");
+    }
+
+    String getProvider(ResourceState state, String containerName) {
+        Map<String, String> currentStateMap = state.getCurrentStateMap(containerName);
+        if (hasInstance(currentStateMap, "ONLINE"))
+            return getInstance(currentStateMap, "ONLINE");
+
+        Map<String, String> pendingStateMap = state.getPendingStateMap(containerName);
+        return getInstance(pendingStateMap, "ONLINE");
+    }
+
+    SortedSet<String> getFailedContainers(ResourceState state, Collection<String> containers) {
+        SortedSet<String> failedContainers = Sets.newTreeSet(new IndexedNameComparator());
+        for (String containerName : containers) {
+            Map<String, String> currentStateMap = state.getCurrentStateMap(containerName);
+            Map<String, String> pendingStateMap = state.getPendingStateMap(containerName);
+
+            if (hasInstance(currentStateMap, "ERROR")) {
+                failedContainers.add(containerName);
+                continue;
+            }
+
+            if (!hasInstance(currentStateMap, "ONLINE") || hasInstance(pendingStateMap, "OFFLINE"))
+                continue;
+
+            // container listed online and not in transition, but not active
+            if (!statusProvider.isHealthy(containerName)) {
+                log.warn(String.format("Container '%s' designated ONLINE, but is not active", containerName));
+                failedContainers.add(containerName);
+            }
+        }
+        return failedContainers;
+    }
+
+    SortedSet<String> getAssignedContainers(ResourceState state, Collection<String> containers) {
+        SortedSet<String> assignedContainers = Sets.newTreeSet(new IndexedNameComparator());
+        for (String containerName : containers) {
+
+            if (!hasProvider(state, containerName))
+                continue;
+
+            assignedContainers.add(containerName);
+        }
+        return assignedContainers;
+    }
+
+    boolean hasInstance(Map<String, String> stateMap, String state) {
+        if (!stateMap.isEmpty()) {
+            for (Map.Entry<String, String> entry : stateMap.entrySet()) {
+                if (entry.getValue().equals(state)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
+    String getInstance(Map<String, String> stateMap, String state) {
+        if (!stateMap.isEmpty()) {
+            for (Map.Entry<String, String> entry : stateMap.entrySet()) {
+                if (entry.getValue().equals(state)) {
+                    return entry.getKey();
+                }
+            }
+        }
+        throw new IllegalArgumentException(String.format("Could not find instance with state '%s'", state));
+    }
+
+    class IndexedNameComparator implements Comparator<String> {
+        Pattern pattern = Pattern.compile("^(.*)([0-9]+)$");
+
+        @Override
+        public int compare(String o1, String o2) {
+            Matcher m1 = pattern.matcher(o1);
+            Matcher m2 = pattern.matcher(o2);
+
+            boolean find1 = m1.find();
+            boolean find2 = m2.find();
+
+            if (!find1 && !find2)
+                return o1.compareTo(o2);
+
+            if (!find1 && find2)
+                return -1;
+
+            if (find1 && !find2)
+                return 1;
+
+            String name1 = m1.group(1);
+            String name2 = m2.group(1);
+
+            int name_comp = name1.compareTo(name2);
+            if (name_comp != 0)
+                return name_comp;
+
+            int index1 = Integer.valueOf(m1.group(2));
+            int index2 = Integer.valueOf(m2.group(2));
+
+            return (int) Math.signum(index1 - index2);
+        }
+    }
+
+    class CountMap extends HashMap<String, Integer> {
+        /**
+         * 
+         */
+        private static final long serialVersionUID = 3954138748385337978L;
+
+        public CountMap(Collection<String> keys) {
+            super();
+            for (String key : keys) {
+                put(key, 0);
+            }
+        }
+
+        @Override
+        public Integer get(Object key) {
+            Preconditions.checkArgument(containsKey(key), "Key %s not found", key);
+            return super.get(key);
+        }
+
+        public int increment(String key) {
+            int newValue = get(key) + 1;
+            Preconditions.checkArgument(containsKey(key), "Key %s not found", key);
+            put(key, newValue);
+            return newValue;
+        }
+
+        public String getMinKey() {
+            Preconditions.checkState(size() > 0, "Must contain at least one item");
+
+            String minKey = null;
+            int minValue = Integer.MAX_VALUE;
+
+            for (String key : keySet()) {
+                int value = get(key);
+                if (value < minValue) {
+                    minValue = value;
+                    minKey = key;
+                }
+            }
+
+            return minKey;
+        }
+
+        public String getMaxKey() {
+            Preconditions.checkState(size() > 0, "Must contain at least one item");
+
+            String maxKey = null;
+            int maxValue = Integer.MIN_VALUE;
+
+            for (String key : keySet()) {
+                int value = get(key);
+                if (value > maxValue) {
+                    maxValue = value;
+                    maxKey = key;
+                }
+            }
+
+            return maxKey;
+        }
+    }
+
+    class ResourceState {
+        final String             resourceName;
+        final CurrentStateOutput state;
+
+        public ResourceState(String resourceName, CurrentStateOutput state) {
+            this.resourceName = resourceName;
+            this.state = state;
+        }
+
+        Map<String, String> getCurrentStateMap(String partitionName) {
+            return state.getCurrentStateMap(resourceName, new Partition(partitionName));
+        }
+
+        Map<String, String> getPendingStateMap(String partitionName) {
+            return state.getPendingStateMap(resourceName, new Partition(partitionName));
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancerSingleton.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancerSingleton.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancerSingleton.java
new file mode 100644
index 0000000..c46f5f5
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderRebalancerSingleton.java
@@ -0,0 +1,38 @@
+package org.apache.helix.metamanager.provider;
+
+import org.apache.helix.metamanager.StatusProvider;
+import org.apache.helix.metamanager.TargetProvider;
+import org.apache.log4j.Logger;
+
+/**
+ * Utility for dependency injection into ProviderRebalancer.
+ * 
+ */
+public class ProviderRebalancerSingleton {
+
+    static final Logger   log = Logger.getLogger(ProviderRebalancerSingleton.class);
+
+    static TargetProvider targetProvider;
+    static StatusProvider statusProvider;
+
+    private ProviderRebalancerSingleton() {
+        // left blank
+    }
+
+    public static TargetProvider getTargetProvider() {
+        return targetProvider;
+    }
+
+    public static void setTargetProvider(TargetProvider targetProvider) {
+        ProviderRebalancerSingleton.targetProvider = targetProvider;
+    }
+
+    public static StatusProvider getStatusProvider() {
+        return statusProvider;
+    }
+
+    public static void setStatusProvider(StatusProvider statusProvider) {
+        ProviderRebalancerSingleton.statusProvider = statusProvider;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModel.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModel.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModel.java
new file mode 100644
index 0000000..090f807
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModel.java
@@ -0,0 +1,114 @@
+package org.apache.helix.metamanager.provider;
+
+import org.apache.helix.NotificationContext;
+import org.apache.helix.metamanager.ClusterAdmin;
+import org.apache.helix.metamanager.ContainerProvider;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.apache.log4j.Logger;
+
+/**
+ * Helix state model implementation for {@link ContainerProvider}s. Updates
+ * configuration of managed Helix cluster and spawns and destroys container
+ * instances.
+ * 
+ */
+@StateModelInfo(initialState = "OFFLINE", states = { "OFFLINE", "ONLINE" })
+public class ProviderStateModel extends StateModel {
+
+    static final Logger log = Logger.getLogger(ProviderStateModel.class);
+
+    ContainerProvider   provider;
+    ClusterAdmin        admin;
+
+    public ProviderStateModel(ContainerProvider provider, ClusterAdmin admin) {
+        this.provider = provider;
+        this.admin = admin;
+    }
+
+    @Transition(from = "OFFLINE", to = "ONLINE")
+    public void acquire(Message m, NotificationContext context) throws Exception {
+        String containerType = m.getResourceName();
+        String containerId = m.getPartitionName();
+        String instanceId = context.getManager().getInstanceName();
+
+        log.trace(String.format("%s:%s transitioning from OFFLINE to ONLINE", containerId, instanceId));
+
+        bestEffortRemove(containerId);
+
+        // add instance to cluster
+        admin.addInstance(containerId, containerType);
+
+        // create container
+        provider.create(containerId, containerType);
+
+        try {
+            admin.rebalance();
+        } catch (Exception e) {
+            // ignore
+            log.warn(String.format("rebalancing cluster failed (error='%s')", e.getMessage()));
+        }
+
+        log.info(String.format("%s acquired container '%s' (type='%s')", instanceId, containerId, containerType));
+    }
+
+    @Transition(from = "ONLINE", to = "OFFLINE")
+    public void release(Message m, NotificationContext context) {
+        String containerId = m.getPartitionName();
+        String instanceId = context.getManager().getInstanceName();
+
+        log.trace(String.format("%s:%s transitioning from ONLINE to OFFLINE", containerId, instanceId));
+
+        bestEffortRemove(containerId);
+
+        try {
+            admin.rebalance();
+        } catch (Exception e) {
+            // ignore
+            log.warn(String.format("rebalancing cluster failed (error='%s')", e.getMessage()));
+        }
+
+        log.info(String.format("%s destroyed container '%s'", instanceId, containerId));
+
+    }
+
+    @Transition(from = "ERROR", to = "OFFLINE")
+    public void recover(Message m, NotificationContext context) {
+        String containerId = m.getPartitionName();
+        String instanceId = context.getManager().getInstanceName();
+
+        log.trace(String.format("%s:%s transitioning from ERROR to OFFLINE", containerId, instanceId));
+
+        release(m, context);
+    }
+
+    @Transition(from = "OFFLINE", to = "DROPPED")
+    public void drop(Message m, NotificationContext context) {
+        String containerId = m.getPartitionName();
+        String instanceId = context.getManager().getInstanceName();
+
+        log.trace(String.format("%s:%s transitioning from OFFLINE to DROPPED", containerId, instanceId));
+    }
+
+    private void bestEffortRemove(String containerId) {
+        log.debug(String.format("Best effort removal of container '%s'", containerId));
+
+        try {
+            provider.destroy(containerId);
+            log.debug(String.format("Container '%s' destroyed", containerId));
+        } catch (Exception e) {
+            log.debug(String.format("Container '%s' does not exist", containerId));
+        }
+
+        try {
+            admin.removeInstance(containerId);
+            log.debug(String.format("Instance '%s' removed", containerId));
+        } catch (Exception e) {
+            log.debug(String.format("Instance '%s' does not exist", containerId));
+        }
+
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModelFactory.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModelFactory.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModelFactory.java
new file mode 100644
index 0000000..36a071a
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/ProviderStateModelFactory.java
@@ -0,0 +1,27 @@
+package org.apache.helix.metamanager.provider;
+
+import org.apache.helix.metamanager.ClusterAdmin;
+import org.apache.helix.metamanager.ContainerProvider;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+
+/**
+ * Factory for {@link ProviderStateModel}. Injects {@link ClusterAdmin} for
+ * managed cluster and {@link ContainerProvider}.
+ * 
+ */
+class ProviderStateModelFactory extends StateModelFactory<ProviderStateModel> {
+
+    final ContainerProvider provider;
+    final ClusterAdmin      admin;
+
+    public ProviderStateModelFactory(ContainerProvider provider, ClusterAdmin admin) {
+        super();
+        this.provider = provider;
+        this.admin = admin;
+    }
+
+    @Override
+    public ProviderStateModel createNewStateModel(String partitionName) {
+        return new ProviderStateModel(provider, admin);
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerProvider.java
new file mode 100644
index 0000000..b63d760
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerProvider.java
@@ -0,0 +1,75 @@
+package org.apache.helix.metamanager.provider.local;
+
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.helix.metamanager.managed.ContainerProcess;
+import org.apache.helix.metamanager.provider.local.LocalContainerSingleton.LocalProcess;
+import org.apache.log4j.Logger;
+
+public class LocalContainerProvider implements ClusterContainerProvider {
+
+	static final Logger log = Logger.getLogger(LocalContainerProvider.class);
+	
+	static final String REQUIRED_TYPE = "container";
+	
+	final String zkAddress;
+	final String clusterName;
+	final String providerName;
+	
+	public LocalContainerProvider(String zkAddress, String clusterName, String providerName) {
+		this.zkAddress = zkAddress;
+		this.clusterName = clusterName;
+		this.providerName = providerName;
+	}
+
+	@Override
+	public void create(String id, String type) throws Exception {
+		Map<String, LocalProcess> processes = LocalContainerSingleton.getProcesses();
+		
+		synchronized (processes) {	
+			if(processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' already exists", id));
+			
+			if(!type.equals(REQUIRED_TYPE))
+				throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+			
+			log.info(String.format("Running container '%s' (zkAddress='%s', clusterName='%s')", id, zkAddress, clusterName));
+			
+			ContainerProcess process = new ContainerProcess(clusterName, zkAddress, id);
+			process.start();
+		
+			processes.put(id, new LocalProcess(id, providerName, process));
+			
+		}
+	}
+	
+	@Override
+	public void destroy(String id) throws Exception {
+		Map<String, LocalProcess> processes = LocalContainerSingleton.getProcesses();
+		
+		synchronized (processes) {	
+			if(!processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+			
+			log.info(String.format("Destroying container '%s'", id));
+			
+			LocalProcess local = processes.remove(id);
+			
+			local.process.stop();
+		}
+	}
+	
+	@Override
+	public void destroyAll() {
+		Map<String, LocalProcess> processes = LocalContainerSingleton.getProcesses();
+		
+		synchronized (processes) {	
+			log.info("Destroying all processes");
+			for(String id : new HashSet<String>(processes.keySet())) {
+				try { destroy(id); } catch (Exception ignore) {}
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerSingleton.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerSingleton.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerSingleton.java
new file mode 100644
index 0000000..d25d3ba
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerSingleton.java
@@ -0,0 +1,40 @@
+package org.apache.helix.metamanager.provider.local;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.helix.metamanager.managed.ContainerProcess;
+
+public class LocalContainerSingleton {
+	final static Map<String, LocalProcess> processes = new HashMap<String, LocalProcess>();
+
+	private LocalContainerSingleton() {
+		// left blank
+	}
+	
+	public static Map<String, LocalProcess> getProcesses() {
+		return processes;
+	}
+	
+	public static void reset() {
+		synchronized (processes) {
+			for(LocalProcess local : processes.values()) {
+				local.process.stop();
+			}
+			processes.clear();
+		}
+	}
+	
+	static class LocalProcess {
+		final String id;
+		final String owner;
+		final ContainerProcess process;
+		
+		public LocalProcess(String id, String owner, ContainerProcess process) {
+			this.id = id;
+			this.owner = owner;
+			this.process = process;
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerStatusProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerStatusProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerStatusProvider.java
new file mode 100644
index 0000000..383a0d7
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/local/LocalContainerStatusProvider.java
@@ -0,0 +1,37 @@
+package org.apache.helix.metamanager.provider.local;
+
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerStatusProvider;
+import org.apache.helix.metamanager.provider.local.LocalContainerSingleton.LocalProcess;
+
+public class LocalContainerStatusProvider implements ClusterContainerStatusProvider {
+
+	@Override
+	public boolean exists(String id) {
+		Map<String, LocalProcess> processes = LocalContainerSingleton.getProcesses();
+		
+		synchronized (processes) {
+			return processes.containsKey(id);
+		}
+	}
+
+	@Override
+	public boolean isActive(String id) {
+		Map<String, LocalProcess> processes = LocalContainerSingleton.getProcesses();
+		
+		synchronized (processes) {
+			return processes.get(id).process != null;
+		}
+	}
+
+	@Override
+	public boolean isFailed(String id) {
+		Map<String, LocalProcess> processes = LocalContainerSingleton.getProcesses();
+		
+		synchronized (processes) {
+			return processes.get(id).process == null;
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerProvider.java
new file mode 100644
index 0000000..eef730a
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerProvider.java
@@ -0,0 +1,81 @@
+package org.apache.helix.metamanager.provider.shell;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.helix.metamanager.provider.shell.ShellContainerSingleton.ShellProcess;
+import org.apache.log4j.Logger;
+
+public class ShellContainerProvider implements ClusterContainerProvider {
+
+	static final Logger log = Logger.getLogger(ShellContainerProvider.class);
+	
+	static final String REQUIRED_TYPE = "container";
+	static final String RUN_COMMAND = "/bin/sh";
+	
+	// global view of processes required
+	static final Object staticLock = new Object();
+	static final Map<String, ShellProcess> processes = new HashMap<String, ShellProcess>();
+
+	final String zkAddress;
+	final String clusterName;
+	final String command;
+	final String providerName;
+	
+	public ShellContainerProvider(String zkAddress, String clusterName, String providerName, String command) {
+		this.zkAddress = zkAddress;
+		this.clusterName = clusterName;
+		this.command = command;
+		this.providerName = providerName;
+	}
+
+	@Override
+	public void create(String id, String type) throws Exception {
+		Map<String, ShellProcess> processes = ShellContainerSingleton.getProcesses();
+		
+		synchronized (processes) {
+			if(processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' already exists", id));
+			
+			if(!type.equals(REQUIRED_TYPE))
+				throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+			
+			log.info(String.format("Running container '%s' (zkAddress='%s', clusterName='%s', command='%s')", id, zkAddress, clusterName, command));
+			
+			ProcessBuilder builder = new ProcessBuilder(RUN_COMMAND, command, zkAddress, clusterName, id);
+			Process process = builder.start();
+			
+			processes.put(id, new ShellProcess(id, providerName, process));
+		}
+	}
+	
+	@Override
+	public void destroy(String id) throws Exception {
+		Map<String, ShellProcess> processes = ShellContainerSingleton.getProcesses();
+		
+		synchronized (processes) {
+			if(!processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+			
+			log.info(String.format("Destroying container '%s'", id));
+			
+			ShellProcess shell = processes.remove(id);
+			shell.process.destroy();
+			shell.process.waitFor();
+		}
+	}
+	
+	@Override
+	public void destroyAll() {
+		Map<String, ShellProcess> processes = ShellContainerSingleton.getProcesses();
+		
+		synchronized (processes) {
+			log.info("Destroying all processes");
+			for(ShellProcess process : new HashSet<ShellProcess>(processes.values())) {
+				try { destroy(process.id); } catch (Exception ignore) {}
+			}
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerSingleton.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerSingleton.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerSingleton.java
new file mode 100644
index 0000000..ae7f3c1
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerSingleton.java
@@ -0,0 +1,38 @@
+package org.apache.helix.metamanager.provider.shell;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class ShellContainerSingleton {
+	static final Map<String, ShellProcess> processes = new HashMap<String, ShellProcess>();
+
+	private ShellContainerSingleton() {
+		// left blank
+	}
+	
+	public static Map<String, ShellProcess> getProcesses() {
+		return processes;
+	}
+	
+	public static void reset() {
+		synchronized (processes) {
+			for(ShellProcess local : processes.values()) {
+				local.process.destroy();
+				try { local.process.waitFor(); } catch(Exception ignore) {}
+			}
+			processes.clear();
+		}
+	}
+	
+	static class ShellProcess {
+		final String id;
+		final String owner;
+		final Process process;
+
+		public ShellProcess(String id, String owner, Process process) {
+			this.id = id;
+			this.owner = owner;
+			this.process = process;
+		}		
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerStatusProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerStatusProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerStatusProvider.java
new file mode 100644
index 0000000..0030c2d
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/shell/ShellContainerStatusProvider.java
@@ -0,0 +1,52 @@
+package org.apache.helix.metamanager.provider.shell;
+
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerStatusProvider;
+import org.apache.helix.metamanager.provider.shell.ShellContainerSingleton.ShellProcess;
+
+public class ShellContainerStatusProvider implements ClusterContainerStatusProvider {
+
+	@Override
+	public boolean exists(String id) {
+		Map<String, ShellProcess> processes = ShellContainerSingleton.getProcesses();
+
+		synchronized (processes) {
+			return processes.containsKey(id);
+		}
+	}
+
+	@Override
+	public boolean isActive(String id) {
+		Map<String, ShellProcess> processes = ShellContainerSingleton.getProcesses();
+
+		synchronized (processes) {
+			ShellProcess shell = processes.get(id);
+			
+			try {
+				shell.process.exitValue();
+				return false;
+			} catch (IllegalThreadStateException e) {
+				// still running
+				return true;
+			}
+		}
+	}
+
+	@Override
+	public boolean isFailed(String id) {
+		Map<String, ShellProcess> processes = ShellContainerSingleton.getProcesses();
+
+		synchronized (processes) {
+			ShellProcess shell = processes.get(id);
+			
+			try {
+				return (shell.process.exitValue() != 0);
+			} catch (IllegalThreadStateException e) {
+				// still running
+				return false;
+			}
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ApplicationConfig.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ApplicationConfig.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ApplicationConfig.java
new file mode 100644
index 0000000..4c8f303
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ApplicationConfig.java
@@ -0,0 +1,32 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+public class ApplicationConfig {
+	final String clusterAddress;
+	final String clusterName;
+	final String metadataAddress;
+	final String providerName;
+
+	public ApplicationConfig(String clusterAddress, String clusterName,
+			String metadataAddress, String providerName) {
+		this.clusterAddress = clusterAddress;
+		this.clusterName = clusterName;
+		this.metadataAddress = metadataAddress;
+		this.providerName = providerName;
+	}
+
+	public String getClusterAddress() {
+		return clusterAddress;
+	}
+
+	public String getClusterName() {
+		return clusterName;
+	}
+
+	public String getMetadataAddress() {
+		return metadataAddress;
+	}
+
+	public String getProviderName() {
+		return providerName;
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ContainerMetadata.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ContainerMetadata.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ContainerMetadata.java
new file mode 100644
index 0000000..73d1a1b
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/ContainerMetadata.java
@@ -0,0 +1,50 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+
+class ContainerMetadata {
+
+	static enum ContainerState {
+		ACQUIRE,
+		CONNECTING,
+		ACTIVE,
+		TEARDOWN,
+		FAILED,
+		HALTED,
+		FINALIZE
+	}
+	
+	String id;
+	ContainerState state;
+	int yarnId;
+	String command;
+	String owner;
+
+	public ContainerMetadata() {
+		// left blank
+	}
+	
+	public ContainerMetadata(String id, String command, String owner) {
+		this.id = id;
+		this.state = ContainerState.ACQUIRE;
+		this.yarnId = -1;
+		this.command = command;
+		this.owner = owner;
+	}
+	
+	public ContainerMetadata(ContainerMetadata node, ContainerState state) {
+		this.id = node.id;
+		this.state = state;
+		this.yarnId = node.yarnId;
+		this.command = node.command;
+		this.owner = node.owner;
+	}
+	
+	public ContainerMetadata(ContainerMetadata node, ContainerState state, int yarnId) {
+		this.id = node.id;
+		this.state = state;
+		this.yarnId = yarnId;
+		this.command = node.command;
+		this.owner = node.owner;
+	}
+}
+

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/MetadataService.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/MetadataService.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/MetadataService.java
new file mode 100644
index 0000000..dc6c060
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/MetadataService.java
@@ -0,0 +1,42 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import java.util.Collection;
+
+public interface MetadataService {
+
+	public boolean exists(String id);
+
+	public void create(ContainerMetadata meta) throws MetadataServiceException;
+
+	public ContainerMetadata read(String id) throws MetadataServiceException;
+
+	public Collection<ContainerMetadata> readAll() throws MetadataServiceException;
+
+	public void update(ContainerMetadata meta) throws MetadataServiceException;
+
+	public void delete(String id) throws MetadataServiceException;
+
+	public static class MetadataServiceException extends Exception {
+
+		/**
+		 * 
+		 */
+		private static final long serialVersionUID = -2846997013918977056L;
+
+		public MetadataServiceException() {
+			super();
+		}
+
+		public MetadataServiceException(String message, Throwable cause) {
+			super(message, cause);
+		}
+
+		public MetadataServiceException(String message) {
+			super(message);
+		}
+
+		public MetadataServiceException(Throwable cause) {
+			super(cause);
+		}	
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/Utils.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/Utils.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/Utils.java
new file mode 100644
index 0000000..82871f1
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/Utils.java
@@ -0,0 +1,94 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.helix.metamanager.provider.yarn.ContainerMetadata.ContainerState;
+import org.apache.log4j.Logger;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapter;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonToken;
+import com.google.gson.stream.JsonWriter;
+
+public class Utils {
+	
+	static final Logger log = Logger.getLogger(Utils.class);
+	
+	static Gson gson;
+	static {
+		GsonBuilder builder = new GsonBuilder();
+		builder.registerTypeAdapter(ContainerState.class, new ContainerStateAdapter());
+		builder.setPrettyPrinting();
+		gson = builder.create();
+	}
+	
+	static Map<String, LocalResource>  dummyResources = createDummyResources();
+	
+	static String toJson(ContainerMetadata meta) {
+		return gson.toJson(meta);
+	}
+	
+	static ContainerMetadata fromJson(String json) {
+		return gson.fromJson(json, ContainerMetadata.class);
+	}
+	
+	static Map<String, LocalResource> getDummyResources() {
+		return dummyResources;
+	}
+
+	private static Map<String, LocalResource> createDummyResources() {
+		File dummy = new File("/tmp/dummy");
+		
+		if(!dummy.exists()) {
+	    	try {
+	    		dummy.createNewFile();
+	    	} catch(Exception e) {
+	    		log.error("could not create dummy file", e);
+	    		System.exit(1);
+	    	}
+		}
+	    
+	    Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
+	    Path path = new Path(dummy.toURI());
+	    LocalResource localResource = Records.newRecord(LocalResource.class);
+	    localResource.setType(LocalResourceType.FILE);
+	    localResource.setVisibility(LocalResourceVisibility.APPLICATION);          
+	    localResource.setResource(ConverterUtils.getYarnUrlFromPath(path)); 
+	    localResource.setTimestamp(dummy.lastModified());
+	    localResource.setSize(dummy.length());
+	    localResources.put("dummy", localResource);
+		return localResources;
+	}
+	
+	static class ContainerStateAdapter extends TypeAdapter<ContainerState> {
+		@Override
+		public ContainerState read(JsonReader reader) throws IOException {
+			if (reader.peek() == JsonToken.NULL) {
+				reader.nextNull();
+				return null;
+			}
+			return ContainerState.valueOf(reader.nextString());
+		}
+
+		@Override
+		public void write(JsonWriter writer, ContainerState value) throws IOException {
+			if (value == null) {
+				writer.nullValue();
+				return;
+			}
+			writer.value(value.name());
+		}
+	}
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnApplication.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnApplication.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnApplication.java
new file mode 100644
index 0000000..c4f3668
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnApplication.java
@@ -0,0 +1,125 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import java.net.InetSocketAddress;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.yarn.api.ClientRMProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Preconditions;
+
+public class YarnApplication {
+
+	static final Logger log = Logger.getLogger(YarnApplication.class);
+	
+	static final String ENV_CLUSTER_ADDRESS = "YA_CLUSTER_ADDRESS";
+	static final String ENV_CLUSTER_NAME = "YA_CLUSTER_NAME";
+	static final String ENV_METADATA_ADDRESS = "YA_METADATA_ADDRESS";
+	static final String ENV_PROVIDER_NAME = "YA_PROVIDER_NAME";
+
+	static final String MASTER_COMMAND = "/bin/sh /home/apucher/incubator-helix/recipes/meta-cluster-manager/target/meta-cluster-manager-pkg/bin/yarn-master-process.sh 1>%s/stdout 2>%s/stderr";
+
+	Configuration conf;
+	YarnRPC rpc;
+	ClientRMProtocol rmClient;
+	ApplicationId appId;
+	
+	final ApplicationConfig appConfig;
+
+	public YarnApplication(ApplicationConfig appConfig) {
+		this.appConfig = appConfig;
+		configure(new YarnConfiguration());
+	}
+
+	public void start() throws Exception {
+		connect();
+		
+		String command = String.format(MASTER_COMMAND, "/tmp/" + appConfig.providerName, "/tmp/" + appConfig.providerName); 
+				//ApplicationConstants.LOG_DIR_EXPANSION_VAR, ApplicationConstants.LOG_DIR_EXPANSION_VAR);
+
+		log.info(String.format("Starting application '%s' provider '%s' (masterCommand='%s')", appConfig.metadataAddress, appConfig.providerName, command));
+
+		// app id
+		GetNewApplicationRequest appRequest = Records.newRecord(GetNewApplicationRequest.class);
+		GetNewApplicationResponse appResponse = rmClient.getNewApplication(appRequest);
+
+		this.appId = appResponse.getApplicationId();
+
+		log.info(String.format("Acquired app id '%s' for '%s'", appId.toString(), appConfig.providerName));
+		
+		// command
+		ContainerLaunchContext launchContext = Records.newRecord(ContainerLaunchContext.class);
+		launchContext.setCommands(Collections.singletonList(command));
+
+		// resource limit
+		Resource resource = Records.newRecord(Resource.class);
+		resource.setMemory(256); // TODO make dynamic
+		launchContext.setResource(resource);
+		
+	    // environment
+	    Map<String, String> env = new HashMap<String, String>();
+	    env.put(ENV_CLUSTER_ADDRESS, appConfig.clusterAddress);
+	    env.put(ENV_CLUSTER_NAME, appConfig.clusterName);
+	    env.put(ENV_METADATA_ADDRESS, appConfig.metadataAddress);
+	    env.put(ENV_PROVIDER_NAME, appConfig.providerName);
+	    launchContext.setEnvironment(env);
+	    
+	    // local resources
+	    // YARN workaround: create dummy resource 
+	    Map<String, LocalResource> localResources = Utils.getDummyResources();
+	    launchContext.setLocalResources(localResources);
+	    
+	    // app submission
+	    ApplicationSubmissionContext subContext = Records.newRecord(ApplicationSubmissionContext.class);
+		subContext.setApplicationId(appId);
+		subContext.setApplicationName(appConfig.providerName);
+		subContext.setAMContainerSpec(launchContext);
+
+		SubmitApplicationRequest subRequest = Records.newRecord(SubmitApplicationRequest.class);
+		subRequest.setApplicationSubmissionContext(subContext);
+		
+		log.info(String.format("Starting app id '%s'", appId.toString()));
+
+		rmClient.submitApplication(subRequest);
+		
+	}
+
+	public void stop() throws YarnRemoteException {
+		log.info(String.format("Stopping app id '%s'", appId.toString()));
+		KillApplicationRequest killRequest = Records.newRecord(KillApplicationRequest.class);
+		killRequest.setApplicationId(appId);
+
+		rmClient.forceKillApplication(killRequest);
+	}
+
+	void configure(Configuration conf) {
+		this.conf = Preconditions.checkNotNull(conf);
+		this.rpc = YarnRPC.create(conf);
+	}
+
+	void connect() {
+		YarnConfiguration yarnConf = new YarnConfiguration(conf);
+		InetSocketAddress rmAddress = NetUtils.createSocketAddr(yarnConf.get(
+				YarnConfiguration.RM_ADDRESS,
+				YarnConfiguration.DEFAULT_RM_ADDRESS));
+		log.info("Connecting to ResourceManager at: " + rmAddress);
+		this.rmClient = ((ClientRMProtocol) rpc.getProxy(ClientRMProtocol.class, rmAddress, conf));
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProcess.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProcess.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProcess.java
new file mode 100644
index 0000000..0d997bf
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProcess.java
@@ -0,0 +1,60 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+/*
+ * 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.
+ */
+
+import org.apache.log4j.Logger;
+
+public class YarnContainerProcess {
+	static final Logger log = Logger.getLogger(YarnContainerProcess.class);
+
+	public static void main(String[] args) throws Exception {
+		log.trace("BEGIN YarnProcess.main()");
+		  
+		final String clusterAddress = args[0];
+		final String clusterName = args[1];
+		final String metadataAddress = args[2];
+		final String providerName = args[3];
+		final String containerId = args[4];
+		
+		final ApplicationConfig appConfig = new ApplicationConfig(clusterAddress, clusterName, metadataAddress, providerName);
+		
+		log.debug("Launching metadata service");
+		final ZookeeperMetadataService metaService = new ZookeeperMetadataService(metadataAddress);
+		metaService.startService();
+		
+		log.debug("Launching yarn container service");
+		final YarnContainerService yarnProcess = new YarnContainerService(appConfig, metaService, containerId);
+		yarnProcess.startService();
+		
+		log.debug("Installing shutdown hooks");
+		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+			@Override
+			public void run() {
+				yarnProcess.stopService();
+				metaService.stopService();
+			}
+		}));
+		
+		System.out.println("Press ENTER to stop container process");
+		System.in.read();
+		
+		log.trace("END YarnProcess.main()");
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProvider.java
new file mode 100644
index 0000000..9f09d46
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerProvider.java
@@ -0,0 +1,108 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import java.util.concurrent.TimeoutException;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.helix.metamanager.provider.yarn.ContainerMetadata.ContainerState;
+import org.apache.helix.metamanager.provider.yarn.MetadataService.MetadataServiceException;
+import org.apache.log4j.Logger;
+
+public class YarnContainerProvider implements ClusterContainerProvider {
+	
+	static final Logger log = Logger.getLogger(YarnContainerProvider.class);
+
+	static final long POLL_INTERVAL = 1000;
+	
+	static final String REQUIRED_TYPE = "container";
+	
+	static final long CONTAINER_TIMEOUT = 10000;
+	
+	/*
+	 * CONTAINERS
+	 *   A (A, READY)
+	 *   B (B, RUNNING)
+	 */
+	
+	final ApplicationConfig appConfig;
+	final String command;
+	
+	final Object notifier = new Object();
+	
+	ZookeeperMetadataService metaService;
+	
+	public YarnContainerProvider(ApplicationConfig appConfig, String command) {
+		this.appConfig = appConfig;
+		this.command = command;
+	}
+
+	@Override
+	public void create(final String id, final String type) throws Exception {
+		if(!REQUIRED_TYPE.equals(type)) {
+			throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+		}
+		
+		metaService.create(new ContainerMetadata(id, command, appConfig.providerName));
+		waitForState(id, ContainerState.ACTIVE);
+	}
+
+	@Override
+	public void destroy(final String id) throws Exception {
+		ContainerMetadata meta = metaService.read(id);
+
+		if(meta.state == ContainerState.ACTIVE) {
+			log.info(String.format("Destroying active container, going to teardown"));
+			metaService.update(new ContainerMetadata(meta, ContainerState.TEARDOWN));
+			
+		} else if(meta.state == ContainerState.FAILED) {
+			log.info(String.format("Destroying failed container, going to halted"));
+			metaService.update(new ContainerMetadata(meta, ContainerState.HALTED));
+			
+		} else if(meta.state == ContainerState.FINALIZE) {
+			log.info(String.format("Destroying finalized container, skipping"));
+			
+		} else {
+			throw new IllegalStateException(String.format("Container '%s' must be active, failed or finalized", id));
+		}
+		
+		waitForState(id, ContainerState.FINALIZE);
+		metaService.delete(id);
+	}
+
+	@Override
+	public void destroyAll() {
+		try {
+			for(ContainerMetadata meta : metaService.readAll()) {
+				try { destroy(meta.id); } catch (Exception ignore) {}
+			}
+		} catch (Exception ignore) {
+			// ignore
+		}
+	}
+
+	public void startService() {
+		log.debug("Starting yarn container provider service");
+		metaService = new ZookeeperMetadataService(appConfig.metadataAddress);
+		metaService.startService();
+	}
+	
+	public void stopService() {
+		log.debug("Stopping yarn container provider service");
+		if(metaService != null) {
+			metaService.stopService();
+			metaService = null;
+		}
+	}
+	
+	void waitForState(String id, ContainerState state) throws MetadataServiceException, InterruptedException, TimeoutException {
+		long limit = System.currentTimeMillis() + CONTAINER_TIMEOUT;
+		ContainerMetadata meta = metaService.read(id);
+		while(meta.state != state) {
+			if(System.currentTimeMillis() >= limit) {
+				throw new TimeoutException(String.format("Container '%s' failed to reach state '%s' (currently is '%s')", id, state, meta.state));
+			}
+			Thread.sleep(POLL_INTERVAL);
+			meta = metaService.read(id);
+		}
+	}
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerService.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerService.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerService.java
new file mode 100644
index 0000000..8abd8df
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerService.java
@@ -0,0 +1,129 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+/*
+ * 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.
+ */
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.metamanager.managed.ManagedFactory;
+import org.apache.helix.metamanager.provider.yarn.ContainerMetadata.ContainerState;
+import org.apache.log4j.Logger;
+
+public class YarnContainerService {
+	static final Logger log = Logger.getLogger(YarnContainerService.class);
+
+	static final long CONTAINERSERVICE_INTERVAL = 1000;
+
+	final ApplicationConfig appConfig;
+	final String containerId;
+	
+	HelixManager participantManager;
+
+	MetadataService metaService;
+	ScheduledExecutorService executor;
+
+	public YarnContainerService(ApplicationConfig appConfig, MetadataService metaService, String containerId) {
+		this.appConfig = appConfig;
+		this.metaService = metaService;
+		this.containerId = containerId;
+	}
+
+	public void startService() {
+		log.debug("starting yarn container service");
+
+		executor = Executors.newSingleThreadScheduledExecutor();
+		executor.scheduleAtFixedRate(new ContainerService(), 0, CONTAINERSERVICE_INTERVAL, TimeUnit.MILLISECONDS);
+	}
+
+	public void stopService() {
+		log.debug("stopping yarn container service");
+		
+		if(executor != null) {
+			executor.shutdown();
+			while(!executor.isTerminated()) {
+				try {
+					Thread.sleep(100);
+				} catch (InterruptedException e) {
+					// ignore
+				}
+			}
+			executor = null;
+		}
+	}
+	
+	public void startParticipant() throws Exception {
+		log.info("STARTING " + containerId);
+		participantManager = HelixManagerFactory.getZKHelixManager(appConfig.clusterName,
+				containerId, InstanceType.PARTICIPANT, appConfig.clusterAddress);
+		participantManager.getStateMachineEngine().registerStateModelFactory(
+				"MasterSlave", new ManagedFactory());
+		participantManager.connect();
+		log.info("STARTED " + containerId);
+	}
+
+	public void stopParticipant() {
+		if (participantManager != null) {
+			participantManager.disconnect();
+			participantManager = null;
+		}
+	}
+	
+	class ContainerService implements Runnable {
+		@Override
+		public void run() {
+			log.info("updating container status");
+			
+			try {
+				ContainerMetadata meta = metaService.read(containerId);
+				
+				if(meta.state == ContainerState.CONNECTING) {
+					log.info("container connecting, going to active");
+					try {
+						startParticipant();
+						metaService.update(new ContainerMetadata(meta, ContainerState.ACTIVE));
+					} catch (Exception e) {
+						log.error("Failed to start participant, going to failed", e);
+						stopParticipant();
+						metaService.update(new ContainerMetadata(meta, ContainerState.FAILED));
+					}
+				}
+				
+				if(meta.state == ContainerState.ACTIVE) {
+					// do something
+					// and go to failed on error
+				}
+				
+				if(meta.state == ContainerState.TEARDOWN) {
+					log.info("container teardown, going to halted");
+					stopParticipant();
+					metaService.update(new ContainerMetadata(meta, ContainerState.HALTED));
+				}
+				
+			} catch(Exception e) {
+				log.error(String.format("Error while updating container '%s' status", containerId), e);
+			}
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerStatusProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerStatusProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerStatusProvider.java
new file mode 100644
index 0000000..54aa3da
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnContainerStatusProvider.java
@@ -0,0 +1,52 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import org.apache.helix.metamanager.ClusterContainerStatusProvider;
+import org.apache.helix.metamanager.provider.yarn.ContainerMetadata.ContainerState;
+import org.apache.helix.metamanager.provider.yarn.MetadataService.MetadataServiceException;
+
+public class YarnContainerStatusProvider implements ClusterContainerStatusProvider {
+
+	final String metadataAddress;
+	
+	ZookeeperMetadataService metaService;
+	
+	public YarnContainerStatusProvider(String metadataAddress) {
+		this.metadataAddress = metadataAddress;
+		this.metaService = new ZookeeperMetadataService(metadataAddress);
+	}
+
+	@Override
+	public boolean exists(String id) {
+		return metaService.exists(id);
+	}
+
+	@Override
+	public boolean isActive(String id) {
+		try {
+			return metaService.read(id).state == ContainerState.ACTIVE;
+		} catch (MetadataServiceException e) {
+			return false;
+		}
+	}
+
+	@Override
+	public boolean isFailed(String id) {
+		try {
+			return metaService.read(id).state == ContainerState.FAILED;
+		} catch (Exception e) {
+			return false;
+		}
+	}
+
+	public void startService() {
+		metaService = new ZookeeperMetadataService(metadataAddress);
+		metaService.startService();
+	}
+	
+	public void stopService() {
+		if(metaService != null) {
+			metaService.stopService();
+			metaService = null;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMaster.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMaster.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMaster.java
new file mode 100644
index 0000000..f43bb67
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMaster.java
@@ -0,0 +1,134 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.yarn.api.AMRMProtocol;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.log4j.Logger;
+
+public class YarnMaster extends Configured implements Tool {
+
+	static final Logger log = Logger.getLogger(YarnMaster.class);
+	
+	AMRMProtocol resourceManager;
+	ApplicationAttemptId appAttemptId;
+	
+	YarnMasterService service;
+	
+	@Override
+	public int run(String[] args) throws Exception {
+		log.trace("BEGIN YarnMaster.run()");
+			
+		Configuration conf = getConf();
+		
+		this.appAttemptId = getApplicationAttemptId();
+		log.info(String.format("Got application attempt id '%s'", appAttemptId.toString()));
+		
+		log.debug("Getting resource manager");
+		this.resourceManager = getResourceManager(conf);
+
+	    // register the AM with the RM
+		log.debug("Registering application master");
+	    RegisterApplicationMasterRequest appMasterRequest = 
+	        Records.newRecord(RegisterApplicationMasterRequest.class);
+	    appMasterRequest.setApplicationAttemptId(appAttemptId);     
+	    appMasterRequest.setHost("");
+	    appMasterRequest.setRpcPort(0);
+	    appMasterRequest.setTrackingUrl("");
+
+	    resourceManager.registerApplicationMaster(appMasterRequest);
+
+	    String clusterAddress = getEnv(YarnApplication.ENV_CLUSTER_ADDRESS);
+	    String clusterName = getEnv(YarnApplication.ENV_CLUSTER_NAME);
+	    String metadataAddress = getEnv(YarnApplication.ENV_METADATA_ADDRESS);
+	    String providerName = getEnv(YarnApplication.ENV_PROVIDER_NAME);
+	    ApplicationConfig appConfig = new ApplicationConfig(clusterAddress, clusterName, metadataAddress, providerName);
+	    
+	    service = new YarnMasterService(resourceManager, conf, appAttemptId, appConfig);
+	    service.startService();
+	    
+	    Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+	    	@Override
+	    	public void run() {
+
+	    		service.stopService();
+	    		
+	    		// finish application
+	    	    log.debug("Sending finish request");
+	    	    FinishApplicationMasterRequest finishReq = 
+	    	    	Records.newRecord(FinishApplicationMasterRequest.class);
+	    	    
+	    	    finishReq.setAppAttemptId(getApplicationAttemptId());
+	    	    finishReq.setFinishApplicationStatus(FinalApplicationStatus.SUCCEEDED);
+	    	    
+	    	    try { resourceManager.finishApplicationMaster(finishReq); } catch(Exception ignore) {}
+	    	}
+	    }));
+	    
+	    try { Thread.currentThread().join(); } catch(Exception ignore) {}
+	    
+		log.trace("END YarnMaster.run()");
+		
+		return 0;
+	}
+
+	private AMRMProtocol getResourceManager(Configuration conf) {
+		// Connect to the Scheduler of the ResourceManager.
+	    YarnConfiguration yarnConf = new YarnConfiguration(conf);
+	    YarnRPC rpc = YarnRPC.create(yarnConf);
+	    InetSocketAddress rmAddress = 
+	        NetUtils.createSocketAddr(yarnConf.get(
+	            YarnConfiguration.RM_SCHEDULER_ADDRESS,
+	            YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS));           
+	    log.info("Connecting to ResourceManager at " + rmAddress);
+	    AMRMProtocol resourceManager = 
+	        (AMRMProtocol) rpc.getProxy(AMRMProtocol.class, rmAddress, conf);
+		return resourceManager;
+	}
+
+	private ApplicationAttemptId getApplicationAttemptId() {
+	    ContainerId containerId = ConverterUtils.toContainerId(getEnv(ApplicationConstants.AM_CONTAINER_ID_ENV));
+	    ApplicationAttemptId appAttemptID = containerId.getApplicationAttemptId();
+		return appAttemptID;
+	}
+	
+	private String getEnv(String key) {
+		Map<String, String> envs = System.getenv();
+	    String clusterName = envs.get(key);
+	    if (clusterName == null) {
+	      // container id should always be set in the env by the framework 
+	      throw new IllegalArgumentException(
+	          String.format("%s not set in the environment", key));
+	    }
+	    return clusterName;
+	}
+
+	public static void main(String[] args) throws Exception {
+		log.trace("BEGIN YarnMaster.main()");
+
+		try {
+			int rc = ToolRunner.run(new Configuration(), new YarnMaster(), args);
+			System.exit(rc);
+		} catch (Exception e) {
+			System.err.println(e);
+			System.exit(1);
+		}
+
+		log.trace("END YarnMaster.main()");
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMasterProcess.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMasterProcess.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMasterProcess.java
new file mode 100644
index 0000000..bd4fb3d
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/provider/yarn/YarnMasterProcess.java
@@ -0,0 +1,119 @@
+package org.apache.helix.metamanager.provider.yarn;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.yarn.api.AMRMProtocol;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.log4j.Logger;
+
+public class YarnMasterProcess {
+
+	static final Logger log = Logger.getLogger(YarnMasterProcess.class);
+	
+	public static void main(String[] args) throws Exception {
+		log.trace("BEGIN YarnMaster.main()");
+
+		final ApplicationAttemptId appAttemptId = getApplicationAttemptId();
+		log.info(String.format("Got application attempt id '%s'", appAttemptId.toString()));
+		
+		log.debug("Connecting to resource manager");
+		Configuration conf = new YarnConfiguration();
+		
+		final AMRMProtocol resourceManager = getResourceManager(conf);
+
+	    // register the AM with the RM
+		log.debug("Registering application master");
+	    RegisterApplicationMasterRequest appMasterRequest = 
+	        Records.newRecord(RegisterApplicationMasterRequest.class);
+	    appMasterRequest.setApplicationAttemptId(appAttemptId);     
+	    appMasterRequest.setHost("");
+	    appMasterRequest.setRpcPort(0);
+	    appMasterRequest.setTrackingUrl("");
+
+	    resourceManager.registerApplicationMaster(appMasterRequest);
+
+	    String clusterAddress = getEnv(YarnApplication.ENV_CLUSTER_ADDRESS);
+	    String clusterName = getEnv(YarnApplication.ENV_CLUSTER_NAME);
+	    String metadataAddress = getEnv(YarnApplication.ENV_METADATA_ADDRESS);
+	    String providerName = getEnv(YarnApplication.ENV_PROVIDER_NAME);
+	    ApplicationConfig appConfig = new ApplicationConfig(clusterAddress, clusterName, metadataAddress, providerName);
+	    
+		log.debug("Launching metadata service");
+	    final ZookeeperMetadataService metaService = new ZookeeperMetadataService(metadataAddress);
+	    metaService.startService();
+	    
+		log.debug("Launching yarn master service");
+	    final YarnMasterService service = new YarnMasterService(resourceManager, conf, appAttemptId, appConfig, metaService);
+	    service.startService();
+	    
+		log.debug("Installing shutdown hooks");
+	    Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+	    	@Override
+	    	public void run() {
+
+	    		service.stopService();
+	    		
+	    		metaService.stopService();
+	    		
+	    		// finish application
+	    	    log.debug("Sending finish request");
+	    	    FinishApplicationMasterRequest finishReq = 
+	    	    	Records.newRecord(FinishApplicationMasterRequest.class);
+	    	    
+	    	    finishReq.setAppAttemptId(getApplicationAttemptId());
+	    	    finishReq.setFinishApplicationStatus(FinalApplicationStatus.SUCCEEDED);
+	    	    
+	    	    try { resourceManager.finishApplicationMaster(finishReq); } catch(Exception ignore) {}
+	    	}
+	    }));
+	    
+	    System.out.println("Press ENTER to stop master service");
+	    System.in.read();
+	    
+		log.trace("END YarnMaster.main()");
+	}
+	
+	static AMRMProtocol getResourceManager(Configuration conf) {
+		// Connect to the Scheduler of the ResourceManager.
+	    YarnConfiguration yarnConf = new YarnConfiguration(conf);
+	    YarnRPC rpc = YarnRPC.create(yarnConf);
+	    InetSocketAddress rmAddress = 
+	        NetUtils.createSocketAddr(yarnConf.get(
+	            YarnConfiguration.RM_SCHEDULER_ADDRESS,
+	            YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS));
+	    log.info("Connecting to ResourceManager at " + rmAddress);
+	    AMRMProtocol resourceManager = 
+	        (AMRMProtocol) rpc.getProxy(AMRMProtocol.class, rmAddress, conf);
+		return resourceManager;
+	}
+
+	static ApplicationAttemptId getApplicationAttemptId() {
+	    ContainerId containerId = ConverterUtils.toContainerId(getEnv(ApplicationConstants.AM_CONTAINER_ID_ENV));
+	    ApplicationAttemptId appAttemptID = containerId.getApplicationAttemptId();
+		return appAttemptID;
+	}
+	
+	static String getEnv(String key) {
+		Map<String, String> envs = System.getenv();
+	    String clusterName = envs.get(key);
+	    if (clusterName == null) {
+	      // container id should always be set in the env by the framework 
+	      throw new IllegalArgumentException(
+	          String.format("%s not set in the environment", key));
+	    }
+	    return clusterName;
+	}
+
+}


Mime
View raw message