You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by ka...@apache.org on 2014/02/21 08:42:41 UTC
git commit: Check live instance to decide when to mark as connected
or disconnected
Repository: helix
Updated Branches:
refs/heads/helix-provisioning 970770acf -> 0037b745a
Check live instance to decide when to mark as connected or disconnected
Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/0037b745
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/0037b745
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/0037b745
Branch: refs/heads/helix-provisioning
Commit: 0037b745aba1ff947d99f33b5683c86b2aacff0f
Parents: 970770a
Author: Kanak Biscuitwala <ka...@apache.org>
Authored: Thu Feb 20 23:41:53 2014 -0800
Committer: Kanak Biscuitwala <ka...@apache.org>
Committed: Thu Feb 20 23:41:53 2014 -0800
----------------------------------------------------------------------
.../stages/ContainerProvisioningStage.java | 29 +++++++++++++++++---
1 file changed, 25 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/helix/blob/0037b745/helix-core/src/main/java/org/apache/helix/controller/stages/ContainerProvisioningStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ContainerProvisioningStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ContainerProvisioningStage.java
index f7105d1..5cccd68 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ContainerProvisioningStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ContainerProvisioningStage.java
@@ -30,6 +30,7 @@ import org.apache.helix.HelixManager;
import org.apache.helix.PropertyKey;
import org.apache.helix.api.Cluster;
import org.apache.helix.api.Participant;
+import org.apache.helix.api.config.ContainerConfig;
import org.apache.helix.api.config.ResourceConfig;
import org.apache.helix.api.id.ParticipantId;
import org.apache.helix.api.id.ResourceId;
@@ -104,7 +105,27 @@ public class ContainerProvisioningStage extends AbstractBaseStage {
final Cluster cluster = event.getAttribute("ClusterDataCache");
final Collection<Participant> participants = cluster.getParticipantMap().values();
- // TODO: if a process died, we need to mark it as stopped or something
+ // If a process died, we need to mark it as DISCONNECTED or if the process is ready, mark as
+ // CONNECTED
+ Map<ParticipantId, Participant> participantMap = cluster.getParticipantMap();
+ for (ParticipantId participantId : participantMap.keySet()) {
+ Participant participant = participantMap.get(participantId);
+ ContainerConfig config = participant.getContainerConfig();
+ if (config != null) {
+ ContainerState containerState = config.getState();
+ if (!participant.isAlive() && ContainerState.CONNECTED.equals(containerState)) {
+ // Need to mark as disconnected if process died
+ LOG.info("Participant " + participantId + " died, marking as DISCONNECTED");
+ updateContainerState(helixAdmin, accessor, keyBuilder, cluster, participantId,
+ ContainerState.DISCONNECTED);
+ } else if (participant.isAlive() && ContainerState.CONNECTING.equals(containerState)) {
+ // Need to mark as connected only when the live instance is visible
+ LOG.info("Participant " + participantId + " is ready, marking as CONNECTED");
+ updateContainerState(helixAdmin, accessor, keyBuilder, cluster, participantId,
+ ContainerState.CONNECTED);
+ }
+ }
+ }
// Participants registered in helix
// Give those participants to targetprovider
@@ -123,7 +144,8 @@ public class ContainerProvisioningStage extends AbstractBaseStage {
// allocate new containers
for (final ContainerSpec spec : response.getContainersToAcquire()) {
final ParticipantId participantId = spec.getParticipantId();
- List<String> instancesInCluster = helixAdmin.getInstancesInCluster(cluster.getId().stringify());
+ List<String> instancesInCluster =
+ helixAdmin.getInstancesInCluster(cluster.getId().stringify());
if (!instancesInCluster.contains(participantId.stringify())) {
// create a new Participant, attach the container spec
InstanceConfig instanceConfig = new InstanceConfig(participantId);
@@ -171,8 +193,7 @@ public class ContainerProvisioningStage extends AbstractBaseStage {
FutureCallback<Boolean> callback = new FutureCallback<Boolean>() {
@Override
public void onSuccess(Boolean result) {
- updateContainerState(helixAdmin, accessor, keyBuilder, cluster, participant.getId(),
- ContainerState.CONNECTED);
+ // Do nothing yet, need to wait for live instance
}
@Override