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 2013/12/11 07:45:39 UTC
[2/2] git commit: Fix compile errors,
fill out ContainerProvisioningStage
Fix compile errors, fill out ContainerProvisioningStage
Project: http://git-wip-us.apache.org/repos/asf/incubator-helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-helix/commit/1022d073
Tree: http://git-wip-us.apache.org/repos/asf/incubator-helix/tree/1022d073
Diff: http://git-wip-us.apache.org/repos/asf/incubator-helix/diff/1022d073
Branch: refs/heads/helix-provisioning
Commit: 1022d0733465e3f2410b9077dd6afc5edb4aafba
Parents: c3aca96
Author: Kanak Biscuitwala <ka...@apache.org>
Authored: Tue Dec 10 22:45:34 2013 -0800
Committer: Kanak Biscuitwala <ka...@apache.org>
Committed: Tue Dec 10 22:45:34 2013 -0800
----------------------------------------------------------------------
.../java/org/apache/helix/api/Controller.java | 4 +-
.../context/ControllerContextProvider.java | 6 +-
.../provisioner/ContainerProvider.java | 38 +++--
.../controller/provisioner/ContainerSpec.java | 31 ++++
.../controller/provisioner/ContainerState.java | 27 +++-
.../provisioner/ParticipantContainer.java | 5 +-
.../controller/provisioner/Provisioner.java | 23 ++-
.../provisioner/ProvisionerConfig.java | 21 ++-
.../controller/provisioner/ProvisionerRef.java | 100 ++++++++++++
.../controller/provisioner/TargetProvider.java | 34 +++-
.../provisioner/TargetProviderResponse.java | 5 +-
.../controller/stages/ClusterDataCache.java | 3 +-
.../stages/ContainerProvisioningStage.java | 155 ++++++++++++++-----
.../stages/MessageSelectionStage.java | 3 +-
.../helix/healthcheck/DecayAggregationType.java | 3 +-
.../DefaultControllerMessageHandlerFactory.java | 7 +-
...ltParticipantErrorMessageHandlerFactory.java | 7 +-
.../DefaultSchedulerMessageHandlerFactory.java | 7 +-
.../manager/zk/HelixConnectionAdaptor.java | 3 +-
.../apache/helix/manager/zk/ZKHelixAdmin.java | 3 +-
.../helix/manager/zk/ZkBaseDataAccessor.java | 7 +-
.../manager/zk/ZkCacheBaseDataAccessor.java | 5 +-
.../helix/manager/zk/ZkCallbackHandler.java | 31 ++--
.../helix/manager/zk/ZkHelixConnection.java | 3 +-
.../helix/manager/zk/ZkHelixController.java | 9 +-
.../helix/manager/zk/ZkHelixLeaderElection.java | 1 -
.../helix/manager/zk/ZkHelixParticipant.java | 3 +-
.../handling/AsyncCallbackService.java | 10 +-
.../org/apache/helix/model/InstanceConfig.java | 34 +++-
.../helix/participant/StateMachineEngine.java | 3 +-
.../java/org/apache/helix/task/TaskDriver.java | 6 +-
.../java/org/apache/helix/task/TaskRunner.java | 3 +-
.../org/apache/helix/task/TaskStateModel.java | 3 +-
.../java/org/apache/helix/task/TaskUtil.java | 15 +-
.../java/org/apache/helix/task/Workflow.java | 2 +-
.../tools/ClusterExternalViewVerifier.java | 3 +-
.../org/apache/helix/tools/ClusterSetup.java | 4 +-
.../org/apache/helix/util/StatusUpdateUtil.java | 5 +-
.../java/org/apache/helix/ZkTestHelper.java | 6 +-
.../stages/TestRebalancePipeline.java | 1 -
.../helix/healthcheck/TestAlertFireHistory.java | 1 -
.../helix/healthcheck/TestExpandAlert.java | 3 +-
.../helix/healthcheck/TestSimpleAlert.java | 3 +-
.../healthcheck/TestSimpleWildcardAlert.java | 3 +-
.../helix/healthcheck/TestStalenessAlert.java | 3 +-
.../helix/healthcheck/TestWildcardAlert.java | 3 +-
.../TestAddNodeAfterControllerStart.java | 4 +-
.../TestAddStateModelFactoryAfterConnect.java | 4 +-
.../helix/integration/TestAutoRebalance.java | 3 +-
.../integration/TestCleanupExternalView.java | 8 +-
.../TestMessagePartitionStateMismatch.java | 3 +-
.../TestParticipantErrorMessage.java | 9 +-
.../integration/TestResetPartitionState.java | 7 +-
.../helix/integration/TestSchedulerMessage.java | 6 +-
.../helix/integration/TestStandAloneCMMain.java | 3 +-
.../integration/TestStateTransitionTimeout.java | 3 +-
...dAloneCMTestBaseWithPropertyServerCheck.java | 30 ++--
.../manager/ClusterDistributedController.java | 3 +-
.../zk/TestZKPropertyTransferServer.java | 6 +-
.../manager/zk/TestZkHelixParticipant.java | 5 +-
.../zk/TestZkManagerFlappingDetection.java | 1 -
.../handling/TestConfigThreadpoolSize.java | 7 +-
.../handling/TestHelixTaskExecutor.java | 9 +-
.../apache/helix/tools/TestHelixAdminCli.java | 3 +-
64 files changed, 539 insertions(+), 230 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/api/Controller.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/api/Controller.java b/helix-core/src/main/java/org/apache/helix/api/Controller.java
index 1218287..c94a6e1 100644
--- a/helix-core/src/main/java/org/apache/helix/api/Controller.java
+++ b/helix-core/src/main/java/org/apache/helix/api/Controller.java
@@ -39,8 +39,8 @@ public class Controller {
if (liveInstance != null) {
_runningInstance =
- new RunningInstance(liveInstance.getTypedSessionId(), liveInstance.getTypedHelixVersion(),
- liveInstance.getProcessId());
+ new RunningInstance(liveInstance.getTypedSessionId(),
+ liveInstance.getTypedHelixVersion(), liveInstance.getProcessId());
} else {
_runningInstance = null;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/context/ControllerContextProvider.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/context/ControllerContextProvider.java b/helix-core/src/main/java/org/apache/helix/controller/context/ControllerContextProvider.java
index 93ed78c..a7a780a 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/context/ControllerContextProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/context/ControllerContextProvider.java
@@ -68,8 +68,7 @@ public class ControllerContextProvider {
* @return a typed ControllerContext, or null if no context with given id is available for this
* type
*/
- public <T extends ControllerContext> T getContext(ContextId contextId,
- Class<T> contextClass) {
+ public <T extends ControllerContext> T getContext(ContextId contextId, Class<T> contextClass) {
try {
if (_pendingContexts.containsKey(contextId)) {
return contextClass.cast(_pendingContexts.get(contextId));
@@ -98,8 +97,7 @@ public class ControllerContextProvider {
* @param overwriteAllowed true if existing objects can be overwritten, false otherwise
* @return true if saved, false if an object with that id exists and overwrite is not allowed
*/
- public boolean putContext(ContextId contextId, ControllerContext context,
- boolean overwriteAllowed) {
+ public boolean putContext(ContextId contextId, ControllerContext context, boolean overwriteAllowed) {
// avoid persisting null contexts
if (context == null) {
LOG.error("Cannot save a null context, id: " + contextId);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerProvider.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerProvider.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerProvider.java
index 2bc5aab..2dee697 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerProvider.java
@@ -1,15 +1,33 @@
package org.apache.helix.controller.provisioner;
+/*
+ * 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.
+ */
public interface ContainerProvider {
-
- ContainerId allocateContainer(ContainerSpec spec);
-
- boolean deallocateContainer(ContainerId containerId);
-
- boolean startContainer(ContainerId containerId);
-
- boolean stopContainer(ContainerId containerId);
-
- ContainerState getContainerState(ContainerId containerId);
+
+ ContainerId allocateContainer(ContainerSpec spec);
+
+ boolean deallocateContainer(ContainerId containerId);
+
+ boolean startContainer(ContainerId containerId);
+
+ boolean stopContainer(ContainerId containerId);
+
+ ContainerState getContainerState(ContainerId containerId);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerSpec.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerSpec.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerSpec.java
index 6ca3fa0..e814059 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerSpec.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerSpec.java
@@ -1,13 +1,44 @@
package org.apache.helix.controller.provisioner;
+/*
+ * 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.
+ */
+
public class ContainerSpec {
/**
* Some unique id representing the container.
*/
ContainerId containerId;
+ public ContainerSpec(ContainerId containerId) {
+ this.containerId = containerId;
+ }
+
public ContainerId getContainerId() {
return containerId;
}
+ @Override
+ public String toString() {
+ return containerId.toString();
+ }
+
+ public static ContainerSpec from(String serialized) {
+ return new ContainerSpec(ContainerId.from(serialized));
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerState.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerState.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerState.java
index 421ff3d..64c8ad5 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerState.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ContainerState.java
@@ -1,5 +1,30 @@
package org.apache.helix.controller.provisioner;
+/*
+ * 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.
+ */
+
public enum ContainerState {
- ACQUIRE, CONNECTING, ACTIVE, TEARDOWN, FAILED, HALTED, FINALIZE
+ ACQUIRING,
+ CONNECTING,
+ ACTIVE,
+ TEARDOWN,
+ FAILED,
+ HALTED,
+ FINALIZING
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/ParticipantContainer.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ParticipantContainer.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ParticipantContainer.java
index 84b69e5..7b39aca 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ParticipantContainer.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ParticipantContainer.java
@@ -6,8 +6,7 @@ public class ParticipantContainer {
* Id request by the target provider
*/
String requestId;
-
+
String allocatedId;
-
-
+
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/Provisioner.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/Provisioner.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/Provisioner.java
index 98c4f9b..0a7a175 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/Provisioner.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/Provisioner.java
@@ -1,9 +1,24 @@
package org.apache.helix.controller.provisioner;
+/*
+ * 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.
+ */
-public interface Provisioner extends ContainerProvider, TargetProvider{
+public interface Provisioner extends ContainerProvider, TargetProvider {
-
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerConfig.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerConfig.java
index 58c099f..54fd492 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerConfig.java
@@ -1,5 +1,24 @@
package org.apache.helix.controller.provisioner;
-public interface ProvisionerConfig {
+/*
+ * 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.
+ */
+public interface ProvisionerConfig {
+ ProvisionerRef getProvisionerRef();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerRef.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerRef.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerRef.java
new file mode 100644
index 0000000..98ecfc4
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/ProvisionerRef.java
@@ -0,0 +1,100 @@
+package org.apache.helix.controller.provisioner;
+
+/*
+ * 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.helix.util.HelixUtil;
+import org.apache.log4j.Logger;
+import org.codehaus.jackson.annotate.JsonCreator;
+import org.codehaus.jackson.annotate.JsonIgnore;
+import org.codehaus.jackson.annotate.JsonProperty;
+
+/**
+ * Reference to a class that extends {@link Provisioner}. It loads the class automatically.
+ */
+public class ProvisionerRef {
+ private static final Logger LOG = Logger.getLogger(ProvisionerRef.class);
+
+ @JsonProperty("provisionerClassName")
+ private final String _provisionerClassName;
+ @JsonIgnore
+ private Provisioner _provisioner;
+
+ @JsonCreator
+ private ProvisionerRef(@JsonProperty("provisionerClassName") String provisionerClassName) {
+ _provisionerClassName = provisionerClassName;
+ _provisioner = null;
+ }
+
+ /**
+ * Get an instantiated Provisioner
+ * @return Provisioner or null if instantiation failed
+ */
+ @JsonIgnore
+ public Provisioner getProvisioner() {
+ if (_provisioner == null) {
+ try {
+ _provisioner =
+ (Provisioner) (HelixUtil.loadClass(getClass(), _provisionerClassName).newInstance());
+ } catch (Exception e) {
+ LOG.warn("Exception while invoking custom provisioner class:" + _provisionerClassName, e);
+ }
+ }
+ return _provisioner;
+ }
+
+ @Override
+ public String toString() {
+ return _provisionerClassName;
+ }
+
+ @Override
+ public boolean equals(Object that) {
+ if (that instanceof ProvisionerRef) {
+ return this.toString().equals(((ProvisionerRef) that).toString());
+ } else if (that instanceof String) {
+ return this.toString().equals(that);
+ }
+ return false;
+ }
+
+ /**
+ * Get a provisioner class reference
+ * @param provisionerClassName name of the class
+ * @return ProvisionerRef or null if name is null
+ */
+ public static ProvisionerRef from(String provisionerClassName) {
+ if (provisionerClassName == null) {
+ return null;
+ }
+ return new ProvisionerRef(provisionerClassName);
+ }
+
+ /**
+ * Get a ProvisionerRef from a class object
+ * @param provisionerClass class that implements Provisioner
+ * @return ProvisionerRef
+ */
+ public static ProvisionerRef from(Class<? extends Provisioner> provisionerClass) {
+ if (provisionerClass == null) {
+ return null;
+ }
+ return ProvisionerRef.from(provisionerClass.getName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProvider.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProvider.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProvider.java
index 82c7d3b..4204da4 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProvider.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProvider.java
@@ -1,7 +1,25 @@
package org.apache.helix.controller.provisioner;
+/*
+ * 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.Collection;
-import java.util.List;
import org.apache.helix.HelixManager;
import org.apache.helix.api.Cluster;
@@ -12,13 +30,13 @@ public interface TargetProvider {
public void init(HelixManager helixManager);
- /**
- * @param cluster
- * @param resourceId ResourceId name of the resource
- * @param participants
- * @return
- */
- TargetProviderResponse evaluateExistingContainers(Cluster cluster,ResourceId resourceId,
+ /**
+ * @param cluster
+ * @param resourceId ResourceId name of the resource
+ * @param participants
+ * @return
+ */
+ TargetProviderResponse evaluateExistingContainers(Cluster cluster, ResourceId resourceId,
Collection<Participant> participants);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProviderResponse.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProviderResponse.java b/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProviderResponse.java
index ecd8a16..1e27525 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProviderResponse.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/provisioner/TargetProviderResponse.java
@@ -5,7 +5,7 @@ import java.util.List;
import org.apache.helix.api.Participant;
public class TargetProviderResponse {
-
+
List<ContainerSpec> containersToAcquire;
List<Participant> containersToRelease;
@@ -46,7 +46,4 @@ public class TargetProviderResponse {
this.containersToStart = containersToStart;
}
-
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
index ac1cef4..d9b563e 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/ClusterDataCache.java
@@ -73,7 +73,8 @@ public class ClusterDataCache {
_liveInstanceMap = accessor.getChildValuesMap(keyBuilder.liveInstances());
for (LiveInstance instance : _liveInstanceMap.values()) {
- LOG.trace("live instance: " + instance.getParticipantId() + " " + instance.getTypedSessionId());
+ LOG.trace("live instance: " + instance.getParticipantId() + " "
+ + instance.getTypedSessionId());
}
_stateModelDefMap = accessor.getChildValuesMap(keyBuilder.stateModelDefs());
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/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 a7ffc99..5b0671c 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
@@ -1,14 +1,33 @@
package org.apache.helix.controller.stages;
-import java.util.ArrayList;
+/*
+ * 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.Collection;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+import java.util.UUID;
import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixDataAccessor;
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.ResourceConfig;
@@ -17,11 +36,13 @@ import org.apache.helix.api.id.ResourceId;
import org.apache.helix.controller.pipeline.AbstractBaseStage;
import org.apache.helix.controller.provisioner.ContainerId;
import org.apache.helix.controller.provisioner.ContainerSpec;
+import org.apache.helix.controller.provisioner.ContainerState;
import org.apache.helix.controller.provisioner.Provisioner;
import org.apache.helix.controller.provisioner.ProvisionerConfig;
-import org.apache.helix.controller.provisioner.TargetProvider;
+import org.apache.helix.controller.provisioner.ProvisionerRef;
import org.apache.helix.controller.provisioner.TargetProviderResponse;
-import org.apache.helix.util.HelixUtil;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.log4j.Logger;
/**
* This stage will manager the container allocation/deallocation needed for a
@@ -34,6 +55,7 @@ import org.apache.helix.util.HelixUtil;
* stop container <br/>
*/
public class ContainerProvisioningStage extends AbstractBaseStage {
+ private static final Logger LOG = Logger.getLogger(ContainerProvisioningStage.class);
Map<ResourceId, Provisioner> _provisionerMap = new HashMap<ResourceId, Provisioner>();
@@ -43,6 +65,8 @@ public class ContainerProvisioningStage extends AbstractBaseStage {
Map<ResourceId, ResourceConfig> resourceMap =
event.getAttribute(AttributeName.RESOURCES.toString());
HelixAdmin helixAdmin = helixManager.getClusterManagmentTool();
+ HelixDataAccessor accessor = helixManager.getHelixDataAccessor();
+ PropertyKey.Builder keyBuilder = accessor.keyBuilder();
for (ResourceId resourceId : resourceMap.keySet()) {
ResourceConfig resourceConfig = resourceMap.get(resourceId);
ProvisionerConfig provisionerConfig = resourceConfig.getProvisionerConfig();
@@ -50,59 +74,108 @@ public class ContainerProvisioningStage extends AbstractBaseStage {
Provisioner provisioner;
provisioner = _provisionerMap.get(resourceId);
+ // instantiate and cache a provisioner if there isn't one already cached
if (provisioner == null) {
- String provisionerClass = resourceConfig.getProvisionerClass();
- provisioner =
- (Provisioner) (HelixUtil.loadClass(getClass(), provisionerClass).newInstance());
- _provisionerMap.put(resourceId, provisioner);
+ ProvisionerRef provisionerRef = provisionerConfig.getProvisionerRef();
+ if (provisionerRef != null) {
+ provisioner = provisionerRef.getProvisioner();
+ }
+ if (provisioner != null) {
+ provisioner.init(helixManager);
+ _provisionerMap.put(resourceId, provisioner);
+ } else {
+ LOG.error("Resource " + resourceId + " does not have a valid provisioner class!");
+ break;
+ }
}
Cluster cluster = event.getAttribute("clusterDataCache");
Collection<Participant> participants = cluster.getParticipantMap().values();
-
- //Participants registered in helix
- //Give those participants to targetprovider
- //Provide the response that contains, new containerspecs, containers to be released, containers to be stopped
- //call the respective provisioner to allocate and start the container.
- //Each container is then started its state is changed from any place.
- //The target provider is given the state of container and asked for its new state. For each state there is a corresponding handler function.
-
- //TargetProvider should be stateless, given the state of cluster and existing participants it should return the same result
- TargetProviderResponse response = provisioner.evaluateExistingContainers(cluster,resourceId, participants);
-
- //start new containers
- for(Participant participant: response.getContainersToStart()){
+
+ // Participants registered in helix
+ // Give those participants to targetprovider
+ // Provide the response that contains, new containerspecs, containers to be released,
+ // containers to be stopped
+ // call the respective provisioner to allocate and start the container.
+ // Each container is then started its state is changed from any place.
+ // The target provider is given the state of container and asked for its new state. For each
+ // state there is a corresponding handler function.
+
+ // TargetProvider should be stateless, given the state of cluster and existing participants
+ // it should return the same result
+ TargetProviderResponse response =
+ provisioner.evaluateExistingContainers(cluster, resourceId, participants);
+
+ // random participant id
+ ParticipantId participantId = ParticipantId.from(UUID.randomUUID().toString());
+
+ // allocate new containers
+ for (ContainerSpec spec : response.getContainersToAcquire()) {
+ // create a new Participant, attach the container spec
+ InstanceConfig instanceConfig = new InstanceConfig(participantId);
+ instanceConfig.setContainerSpec(spec);
+ // create a helix_participant in ACQUIRING state
+ instanceConfig.setContainerState(ContainerState.ACQUIRING);
+ // create the helix participant and add it to cluster
+ helixAdmin.addInstance(cluster.getId().toString(), instanceConfig);
+
+ ContainerId containerId = provisioner.allocateContainer(spec);
+ InstanceConfig existingInstance =
+ helixAdmin.getInstanceConfig(cluster.getId().toString(), participantId.toString());
+ existingInstance.setContainerId(containerId);
+ accessor.setProperty(keyBuilder.instanceConfig(participantId.toString()),
+ existingInstance);
+ }
+
+ // start new containers
+ for (Participant participant : response.getContainersToStart()) {
String containerIdStr = participant.getUserConfig().getSimpleField("ContainerId");
ContainerId containerId = ContainerId.from(containerIdStr);
- //create the helix participant and add it to cluster
+ InstanceConfig existingInstance =
+ helixAdmin.getInstanceConfig(cluster.getId().toString(), participantId.toString());
+ existingInstance.setContainerState(ContainerState.CONNECTING);
+ accessor.setProperty(keyBuilder.instanceConfig(participantId.toString()),
+ existingInstance);
+ // create the helix participant and add it to cluster
provisioner.startContainer(containerId);
+ existingInstance =
+ helixAdmin.getInstanceConfig(cluster.getId().toString(), participantId.toString());
+ existingInstance.setContainerState(ContainerState.ACTIVE);
+ accessor.setProperty(keyBuilder.instanceConfig(participantId.toString()),
+ existingInstance);
}
-
- //allocate new containers
- for(ContainerSpec spec: response.getContainersToAcquire()){
- //create a new Participant, attach the container spec
- //create a helix_participant in ACQUIRING STATE
- ContainerId containerId = provisioner.allocateContainer(spec);
- //create the helix participant and add it to cluster
- provisioner.startContainer(containerId);
- }
-
- //release containers
- for(Participant participant: response.getContainersToRelease()){
+
+ // release containers
+ for (Participant participant : response.getContainersToRelease()) {
String containerIdStr = participant.getUserConfig().getSimpleField("ContainerId");
ContainerId containerId = ContainerId.from(containerIdStr);
- //disable the node first
- provisioner.stopContainer(containerId);
- //this will change the container state
+ // this will change the container state
provisioner.deallocateContainer(containerId);
- //remove the participant
+ // remove the participant
+ InstanceConfig existingInstance =
+ helixAdmin.getInstanceConfig(cluster.getId().toString(), participantId.toString());
+ helixAdmin.dropInstance(cluster.getId().toString(), existingInstance);
}
- //stop but dont remove
- for(Participant participant: participants){
+ // stop but don't remove
+ for (Participant participant : participants) {
String containerIdStr = participant.getUserConfig().getSimpleField("ContainerId");
+ // disable the node first
+ // TODO: get the participant id from the container id
+ InstanceConfig existingInstance =
+ helixAdmin.getInstanceConfig(cluster.getId().toString(), participantId.toString());
+ existingInstance.setInstanceEnabled(false);
+ existingInstance.setContainerState(ContainerState.TEARDOWN);
+ accessor.setProperty(keyBuilder.instanceConfig(participantId.toString()),
+ existingInstance);
+ // stop the container
ContainerId containerId = ContainerId.from(containerIdStr);
provisioner.stopContainer(containerId);
+ existingInstance =
+ helixAdmin.getInstanceConfig(cluster.getId().toString(), participantId.toString());
+ existingInstance.setContainerState(ContainerState.HALTED);
+ accessor.setProperty(keyBuilder.instanceConfig(participantId.toString()),
+ existingInstance);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
index 966160c..9ac5012 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/MessageSelectionStage.java
@@ -112,7 +112,8 @@ public class MessageSelectionStage extends AbstractBaseStage {
stateModelDefMap.get(resource.getRebalancerConfig().getStateModelDefId());
if (stateModelDef == null) {
- LOG.info("resource: " + resourceId
+ LOG.info("resource: "
+ + resourceId
+ " doesn't have state-model-def; e.g. we add a resource config but not add the resource in ideal-states");
continue;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/healthcheck/DecayAggregationType.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/healthcheck/DecayAggregationType.java b/helix-core/src/main/java/org/apache/helix/healthcheck/DecayAggregationType.java
index c3252e9..2409b84 100644
--- a/helix-core/src/main/java/org/apache/helix/healthcheck/DecayAggregationType.java
+++ b/helix-core/src/main/java/org/apache/helix/healthcheck/DecayAggregationType.java
@@ -52,7 +52,6 @@ public class DecayAggregationType implements AggregationType {
// come up with decay coeff for old value. More time passed, the more it
// decays
double oldDecayCoeff = Math.pow((1 - _decayFactor), minutesOld);
- return String
- .valueOf((oldDecayCoeff * existingVal + (1 - oldDecayCoeff) * incomingVal));
+ return String.valueOf((oldDecayCoeff * existingVal + (1 - oldDecayCoeff) * incomingVal));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
index 33d271b..a848296 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultControllerMessageHandlerFactory.java
@@ -36,8 +36,8 @@ public class DefaultControllerMessageHandlerFactory implements MessageHandlerFac
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMessageId() + " type:"
- + message.getMsgType());
+ throw new HelixException("Unexpected msg type for message " + message.getMessageId()
+ + " type:" + message.getMsgType());
}
return new DefaultControllerMessageHandler(message, context);
@@ -74,7 +74,8 @@ public class DefaultControllerMessageHandlerFactory implements MessageHandlerFac
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(),
+ e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
index 5e3a7ea..ef1d974 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultParticipantErrorMessageHandlerFactory.java
@@ -99,7 +99,8 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(),
+ e);
}
}
@@ -109,8 +110,8 @@ public class DefaultParticipantErrorMessageHandlerFactory implements MessageHand
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMessageId() + " type:"
- + message.getMsgType());
+ throw new HelixException("Unexpected msg type for message " + message.getMessageId()
+ + " type:" + message.getMsgType());
}
return new DefaultParticipantErrorMessageHandler(message, context, _manager);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
index 4fe9164..4c30808 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/DefaultSchedulerMessageHandlerFactory.java
@@ -134,8 +134,8 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
String type = message.getMsgType();
if (!type.equals(getMessageType())) {
- throw new HelixException("Unexpected msg type for message " + message.getMessageId() + " type:"
- + message.getMsgType());
+ throw new HelixException("Unexpected msg type for message " + message.getMessageId()
+ + " type:" + message.getMsgType());
}
return new DefaultSchedulerMessageHandler(message, context, _manager);
@@ -348,7 +348,8 @@ public class DefaultSchedulerMessageHandlerFactory implements MessageHandlerFact
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(),
+ e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/HelixConnectionAdaptor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/HelixConnectionAdaptor.java b/helix-core/src/main/java/org/apache/helix/manager/zk/HelixConnectionAdaptor.java
index e52427c..e13c127 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/HelixConnectionAdaptor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/HelixConnectionAdaptor.java
@@ -235,8 +235,7 @@ public class HelixConnectionAdaptor implements HelixManager {
engine = autoController.getStateMachineEngine();
break;
default:
- LOG.info("helix manager type: " + _role.getType()
- + " does NOT have state-machine-engine");
+ LOG.info("helix manager type: " + _role.getType() + " does NOT have state-machine-engine");
break;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
index 6e30074..39ea1c5 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZKHelixAdmin.java
@@ -917,7 +917,8 @@ public class ZKHelixAdmin implements HelixAdmin {
new ZKHelixDataAccessor(grandCluster, new ZkBaseDataAccessor<ZNRecord>(_zkClient));
Builder keyBuilder = accessor.keyBuilder();
- accessor.setProperty(keyBuilder.idealStates(idealState.getResourceId().stringify()), idealState);
+ accessor
+ .setProperty(keyBuilder.idealStates(idealState.getResourceId().stringify()), idealState);
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
index 4c7798f..724e299 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkBaseDataAccessor.java
@@ -641,7 +641,8 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T> {
if (failOnNoParentNode) {
List<AccessResult> createParentResults =
- doCreate(parentPaths, null, Arrays.copyOf(needCreate, needCreate.length), AccessOption.PERSISTENT);
+ doCreate(parentPaths, null, Arrays.copyOf(needCreate, needCreate.length),
+ AccessOption.PERSISTENT);
for (int i = 0; i < createParentResults.size(); i++) {
if (!needCreate[i]) {
continue;
@@ -730,8 +731,8 @@ public class ZkBaseDataAccessor<T> implements BaseDataAccessor<T> {
final int size = paths.size();
if (records != null && records.size() != size) {
throw new IllegalArgumentException(
- "paths and records should be of same size, but paths size: " + size
- + ", records size: " + records.size());
+ "paths and records should be of same size, but paths size: " + size + ", records size: "
+ + records.size());
}
CreateMode mode = AccessOption.getMode(options);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
index a36c49a..57bdc21 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCacheBaseDataAccessor.java
@@ -483,12 +483,13 @@ public class ZkCacheBaseDataAccessor<T> implements HelixPropertyStore<T> {
List<List<String>> pathsCreatedList =
new ArrayList<List<String>>(Collections.<List<String>> nCopies(size, null));
- List<ZkBaseDataAccessor<T>.AccessResult> results = _baseAccessor.doUpdate(serverPaths, updaters, options);
+ List<ZkBaseDataAccessor<T>.AccessResult> results =
+ _baseAccessor.doUpdate(serverPaths, updaters, options);
for (int i = 0; i < size; i++) {
ZkBaseDataAccessor<T>.AccessResult result = results.get(i);
success[i] = (result._retCode == RetCode.OK);
- updateCache(cache, pathsCreatedList.get(i), success[i], serverPaths.get(i),
+ updateCache(cache, pathsCreatedList.get(i), success[i], serverPaths.get(i),
result._resultValue, results.get(i)._stat);
}
return success;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
index e0a0b33..8d39268 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkCallbackHandler.java
@@ -116,8 +116,7 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
*/
private List<NotificationContext.Type> _expectTypes = nextNotificationType.get(Type.FINALIZE);
- public ZkCallbackHandler(HelixRole role, ZkClient client,
- PropertyKey propertyKey,
+ public ZkCallbackHandler(HelixRole role, ZkClient client, PropertyKey propertyKey,
Object listener, EventType[] eventTypes, ChangeType changeType) {
if (listener == null) {
throw new HelixException("listener could not be null");
@@ -253,12 +252,12 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
private void subscribeChildChange(String path, NotificationContext context) {
NotificationContext.Type type = context.getType();
if (type == NotificationContext.Type.INIT || type == NotificationContext.Type.CALLBACK) {
- logger.info(_instanceName + " subscribes child-change. path: " + path
- + ", listener: " + _listener);
+ logger.info(_instanceName + " subscribes child-change. path: " + path + ", listener: "
+ + _listener);
_zkClient.subscribeChildChanges(path, this);
} else if (type == NotificationContext.Type.FINALIZE) {
- logger.info(_instanceName + " unsubscribe child-change. path: " + path
- + ", listener: " + _listener);
+ logger.info(_instanceName + " unsubscribe child-change. path: " + path + ", listener: "
+ + _listener);
_zkClient.unsubscribeChildChanges(path, this);
}
@@ -268,14 +267,14 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
NotificationContext.Type type = context.getType();
if (type == NotificationContext.Type.INIT || type == NotificationContext.Type.CALLBACK) {
if (logger.isDebugEnabled()) {
- logger.debug(_instanceName + " subscribe data-change. path: " + path
- + ", listener: " + _listener);
+ logger.debug(_instanceName + " subscribe data-change. path: " + path + ", listener: "
+ + _listener);
}
_zkClient.subscribeDataChanges(path, this);
} else if (type == NotificationContext.Type.FINALIZE) {
- logger.info(_instanceName + " unsubscribe data-change. path: " + path
- + ", listener: " + _listener);
+ logger.info(_instanceName + " unsubscribe data-change. path: " + path + ", listener: "
+ + _listener);
_zkClient.unsubscribeDataChanges(path, this);
}
@@ -382,14 +381,14 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
try {
updateNotificationTime(System.nanoTime());
if (dataPath != null && dataPath.startsWith(_path)) {
- logger.info(_instanceName + " unsubscribe data-change. path: " + dataPath
- + ", listener: " + _listener);
+ logger.info(_instanceName + " unsubscribe data-change. path: " + dataPath + ", listener: "
+ + _listener);
_zkClient.unsubscribeDataChanges(dataPath, this);
// only needed for bucketized parent, but OK if we don't have child-change
// watch on the bucketized parent path
- logger.info(_instanceName + " unsubscribe child-change. path: " + dataPath
- + ", listener: " + _listener);
+ logger.info(_instanceName + " unsubscribe child-change. path: " + dataPath + ", listener: "
+ + _listener);
_zkClient.unsubscribeChildChanges(dataPath, this);
// No need to invoke() since this event will handled by child-change on parent-node
// NotificationContext changeContext = new NotificationContext(_manager);
@@ -425,8 +424,8 @@ public class ZkCallbackHandler implements IZkChildListener, IZkDataListener
}
} catch (Exception e) {
String msg =
- "exception in handling child-change. instance: " + _instanceName
- + ", parentPath: " + parentPath + ", listener: " + _listener;
+ "exception in handling child-change. instance: " + _instanceName + ", parentPath: "
+ + parentPath + ", listener: " + _listener;
ZKExceptionHandler.getInstance().handle(msg, e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
index 1bdc54c..0a9349f 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixConnection.java
@@ -380,8 +380,7 @@ public class ZkHelixConnection implements HelixConnection, IZkStateListener {
ClusterId clusterId, ParticipantId participantId) {
addListener(role, listener,
new PropertyKey.Builder(clusterId.stringify()).healthReports(participantId.stringify()),
- ChangeType.HEALTH,
- new EventType[] {
+ ChangeType.HEALTH, new EventType[] {
EventType.NodeChildrenChanged, EventType.NodeDeleted, EventType.NodeCreated
});
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
index f116c76..3091a90 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixController.java
@@ -59,7 +59,8 @@ public class ZkHelixController implements HelixController {
final HelixManager _manager;
final ZkHelixLeaderElection _leaderElection;
- public ZkHelixController(ZkHelixConnection connection, ClusterId clusterId, ControllerId controllerId) {
+ public ZkHelixController(ZkHelixConnection connection, ClusterId clusterId,
+ ControllerId controllerId) {
_connection = connection;
_clusterId = clusterId;
_controllerId = controllerId;
@@ -200,8 +201,7 @@ public class ZkHelixController implements HelixController {
/**
* setup controller message listener and register message handlers
*/
- _connection.addControllerMessageListener(this, _messagingService.getExecutor(),
- _clusterId);
+ _connection.addControllerMessageListener(this, _messagingService.getExecutor(), _clusterId);
MessageHandlerFactory defaultControllerMsgHandlerFactory =
new DefaultControllerMessageHandlerFactory();
_messagingService.getExecutor().registerMessageHandlerFactory(
@@ -224,8 +224,7 @@ public class ZkHelixController implements HelixController {
_connection.addIdealStateChangeListener(this, pipeline, _clusterId);
_connection.addControllerListener(this, pipeline, _clusterId);
} catch (ZkInterruptedException e) {
- LOG.warn("zk connection is interrupted during addListenersToController()"
- + e);
+ LOG.warn("zk connection is interrupted during addListenersToController()" + e);
} catch (Exception e) {
LOG.error("Error addListenersToController", e);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixLeaderElection.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixLeaderElection.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixLeaderElection.java
index 77da158..036571c 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixLeaderElection.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixLeaderElection.java
@@ -84,7 +84,6 @@ public class ZkHelixLeaderElection implements ControllerChangeListener {
|| changeContext.getType().equals(NotificationContext.Type.CALLBACK)) {
LOG.info(_controllerId + " is trying to acquire leadership for cluster: " + _clusterId);
-
while (accessor.getProperty(keyBuilder.controllerLeader()) == null) {
boolean success = tryUpdateController(_manager);
if (success) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
index 56e4be8..f8f7a46 100644
--- a/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
+++ b/helix-core/src/main/java/org/apache/helix/manager/zk/ZkHelixParticipant.java
@@ -280,8 +280,7 @@ public class ZkHelixParticipant implements HelixParticipant {
String curStatePath =
_keyBuilder.currentState(participantName, sessionId, lastCurState.getResourceName())
.getPath();
- _accessor.getBaseDataAccessor().update(
- curStatePath,
+ _accessor.getBaseDataAccessor().update(curStatePath,
new CurStateCarryOverUpdater(sessionId, stateModel.getInitialState(), lastCurState),
AccessOption.PERSISTENT);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java b/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
index 17fc67d..ac7a862 100644
--- a/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
+++ b/helix-core/src/main/java/org/apache/helix/messaging/handling/AsyncCallbackService.java
@@ -47,8 +47,8 @@ public class AsyncCallbackService implements MessageHandlerFactory {
void verifyMessage(Message message) {
if (!message.getMsgType().toString().equalsIgnoreCase(MessageType.TASK_REPLY.toString())) {
String errorMsg =
- "Unexpected msg type for message " + message.getMessageId() + " type:" + message.getMsgType()
- + " Expected : " + MessageType.TASK_REPLY;
+ "Unexpected msg type for message " + message.getMessageId() + " type:"
+ + message.getMsgType() + " Expected : " + MessageType.TASK_REPLY;
_logger.error(errorMsg);
throw new HelixException(errorMsg);
}
@@ -68,7 +68,8 @@ public class AsyncCallbackService implements MessageHandlerFactory {
_logger.error(errorMsg);
throw new HelixException(errorMsg);
}
- _logger.info("Verified reply message " + message.getMessageId() + " correlation:" + correlationId);
+ _logger.info("Verified reply message " + message.getMessageId() + " correlation:"
+ + correlationId);
}
@Override
@@ -118,7 +119,8 @@ public class AsyncCallbackService implements MessageHandlerFactory {
@Override
public void onError(Exception e, ErrorCode code, ErrorType type) {
- _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(), e);
+ _logger.error("Message handling pipeline get an exception. MsgId:" + _message.getMessageId(),
+ e);
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java b/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
index 35b4bd4..c547fa2 100644
--- a/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
+++ b/helix-core/src/main/java/org/apache/helix/model/InstanceConfig.java
@@ -31,6 +31,9 @@ import org.apache.helix.api.config.NamespacedConfig;
import org.apache.helix.api.config.UserConfig;
import org.apache.helix.api.id.ParticipantId;
import org.apache.helix.api.id.PartitionId;
+import org.apache.helix.controller.provisioner.ContainerId;
+import org.apache.helix.controller.provisioner.ContainerSpec;
+import org.apache.helix.controller.provisioner.ContainerState;
import com.google.common.base.Enums;
import com.google.common.base.Optional;
@@ -47,7 +50,10 @@ public class InstanceConfig extends HelixProperty {
HELIX_PORT,
HELIX_ENABLED,
HELIX_DISABLED_PARTITION,
- TAG_LIST
+ TAG_LIST,
+ CONTAINER_SPEC,
+ CONTAINER_STATE,
+ CONTAINER_ID
}
/**
@@ -297,6 +303,32 @@ public class InstanceConfig extends HelixProperty {
return userConfig;
}
+ public void setContainerSpec(ContainerSpec spec) {
+ _record.setSimpleField(InstanceConfigProperty.CONTAINER_SPEC.toString(), spec.toString());
+ }
+
+ public ContainerSpec getContainerSpec() {
+ return ContainerSpec.from(_record.getSimpleField(InstanceConfigProperty.CONTAINER_SPEC
+ .toString()));
+ }
+
+ public void setContainerState(ContainerState state) {
+ _record.setEnumField(InstanceConfigProperty.CONTAINER_STATE.toString(), state);
+ }
+
+ public ContainerState getContainerState() {
+ return _record.getEnumField(InstanceConfigProperty.CONTAINER_STATE.toString(),
+ ContainerState.class, null);
+ }
+
+ public void setContainerId(ContainerId containerId) {
+ _record.setSimpleField(InstanceConfigProperty.CONTAINER_ID.toString(), containerId.toString());
+ }
+
+ public ContainerId getContainerId() {
+ return ContainerId.from(_record.getSimpleField(InstanceConfigProperty.CONTAINER_ID.toString()));
+ }
+
@Override
public boolean isValid() {
// HELIX-65: remove check for hostname/port existence
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/participant/StateMachineEngine.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/participant/StateMachineEngine.java b/helix-core/src/main/java/org/apache/helix/participant/StateMachineEngine.java
index 80c9545..abb7d81 100644
--- a/helix-core/src/main/java/org/apache/helix/participant/StateMachineEngine.java
+++ b/helix-core/src/main/java/org/apache/helix/participant/StateMachineEngine.java
@@ -33,6 +33,7 @@ public interface StateMachineEngine extends MessageHandlerFactory {
/**
* Replaced by {@link #registerStateModelFactory(StateModelDefId, HelixStateModelFactory)
+
*/
@Deprecated
public boolean registerStateModelFactory(String stateModelDef,
@@ -82,7 +83,7 @@ public interface StateMachineEngine extends MessageHandlerFactory {
public boolean registerStateModelFactory(StateModelDefId stateModelDefId, String factoryName,
HelixStateModelFactory<? extends StateModel> factory);
- /**
+/**
* Remove the default state model factory for a state model definition
* Replacing {@link #removeStateModelFactory(String, StateModelFactory)
* @param stateModelDefId
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java b/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
index d017134..5189fb7 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskDriver.java
@@ -283,13 +283,11 @@ public class TaskDriver {
Option clusterNameOption =
OptionBuilder.isRequired().hasArgs(1).withArgName("clusterName")
- .withLongOpt(CLUSTER_NAME_OPTION)
- .withDescription("Target cluster name").create();
+ .withLongOpt(CLUSTER_NAME_OPTION).withDescription("Target cluster name").create();
Option taskResourceOption =
OptionBuilder.isRequired().hasArgs(1).withArgName("resourceName")
- .withLongOpt(RESOURCE_OPTION)
- .withDescription("Target workflow or task").create();
+ .withLongOpt(RESOURCE_OPTION).withDescription("Target workflow or task").create();
OptionGroup group = new OptionGroup();
group.addOption(zkAddressOption);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java b/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
index e7a9abb..92976b0 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskRunner.java
@@ -51,8 +51,7 @@ public class TaskRunner implements Runnable {
private volatile boolean _done = false;
public TaskRunner(StateModel taskStateModel, Task task, String taskName, String taskPartition,
- String instance,
- HelixManager manager, String sessionId) {
+ String instance, HelixManager manager, String sessionId) {
_taskStateModel = taskStateModel;
_task = task;
_taskName = taskName;
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java b/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
index 2a6d003..cecf2e8 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskStateModel.java
@@ -222,8 +222,7 @@ public class TaskStateModel extends StateModel {
_taskRunner =
new TaskRunner(this, task, msg.getResourceName(), taskPartition, msg.getTgtName(),
- _manager,
- msg.getTgtSessionId());
+ _manager, msg.getTgtSessionId());
_taskExecutor.submit(_taskRunner);
_taskRunner.waitTillStarted();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java b/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
index a9428c6..c81be5d 100644
--- a/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/task/TaskUtil.java
@@ -107,8 +107,7 @@ public class TaskUtil {
ZNRecord r =
manager.getHelixPropertyStore().get(
Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName,
- TaskUtilEnum.PREV_RA_NODE.value()),
- null, AccessOption.PERSISTENT);
+ TaskUtilEnum.PREV_RA_NODE.value()), null, AccessOption.PERSISTENT);
return r != null ? new ResourceAssignment(r) : null;
}
@@ -116,24 +115,21 @@ public class TaskUtil {
ResourceAssignment ra) {
manager.getHelixPropertyStore().set(
Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName,
- TaskUtilEnum.PREV_RA_NODE.value()),
- ra.getRecord(), AccessOption.PERSISTENT);
+ TaskUtilEnum.PREV_RA_NODE.value()), ra.getRecord(), AccessOption.PERSISTENT);
}
public static TaskContext getTaskContext(HelixManager manager, String taskResource) {
ZNRecord r =
manager.getHelixPropertyStore().get(
Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, taskResource,
- TaskUtilEnum.CONTEXT_NODE.value()),
- null, AccessOption.PERSISTENT);
+ TaskUtilEnum.CONTEXT_NODE.value()), null, AccessOption.PERSISTENT);
return r != null ? new TaskContext(r) : null;
}
public static void setTaskContext(HelixManager manager, String taskResource, TaskContext ctx) {
manager.getHelixPropertyStore().set(
Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, taskResource,
- TaskUtilEnum.CONTEXT_NODE.value()),
- ctx.getRecord(), AccessOption.PERSISTENT);
+ TaskUtilEnum.CONTEXT_NODE.value()), ctx.getRecord(), AccessOption.PERSISTENT);
}
public static WorkflowContext getWorkflowContext(HelixManager manager, String workflowResource) {
@@ -148,8 +144,7 @@ public class TaskUtil {
WorkflowContext ctx) {
manager.getHelixPropertyStore().set(
Joiner.on("/").join(TaskConstants.REBALANCER_CONTEXT_ROOT, workflowResource,
- TaskUtilEnum.CONTEXT_NODE.value()),
- ctx.getRecord(), AccessOption.PERSISTENT);
+ TaskUtilEnum.CONTEXT_NODE.value()), ctx.getRecord(), AccessOption.PERSISTENT);
}
public static String getNamespacedTaskName(String singleTaskWorkflow) {
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/task/Workflow.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/task/Workflow.java b/helix-core/src/main/java/org/apache/helix/task/Workflow.java
index c5c005b..902f616 100644
--- a/helix-core/src/main/java/org/apache/helix/task/Workflow.java
+++ b/helix-core/src/main/java/org/apache/helix/task/Workflow.java
@@ -94,7 +94,7 @@ public class Workflow {
* the following
* form:
* <p/>
- *
+ *
* <pre>
* name: MyFlow
* tasks:
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/tools/ClusterExternalViewVerifier.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterExternalViewVerifier.java b/helix-core/src/main/java/org/apache/helix/tools/ClusterExternalViewVerifier.java
index a0959cc..f090763 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterExternalViewVerifier.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterExternalViewVerifier.java
@@ -68,7 +68,8 @@ public class ClusterExternalViewVerifier extends ClusterVerifier {
boolean verifyLiveNodes(List<ParticipantId> actualLiveNodes) {
Collections.sort(actualLiveNodes);
- List<String> rawActualLiveNodes = Lists.transform(actualLiveNodes, Functions.toStringFunction());
+ List<String> rawActualLiveNodes =
+ Lists.transform(actualLiveNodes, Functions.toStringFunction());
return _expectSortedLiveNodes.equals(rawActualLiveNodes);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java b/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
index 0239312..0247846 100644
--- a/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
+++ b/helix-core/src/main/java/org/apache/helix/tools/ClusterSetup.java
@@ -312,8 +312,8 @@ public class ClusterSetup {
accessor.getChildValues(accessor.keyBuilder().idealStates());
for (IdealState idealState : existingIdealStates) {
swapInstanceInIdealState(idealState, oldInstanceName, newInstanceName);
- accessor.setProperty(
- accessor.keyBuilder().idealStates(idealState.getResourceId().stringify()), idealState);
+ accessor.setProperty(accessor.keyBuilder()
+ .idealStates(idealState.getResourceId().stringify()), idealState);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java b/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
index d0a57f1..6e4dbbc 100644
--- a/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
+++ b/helix-core/src/main/java/org/apache/helix/util/StatusUpdateUtil.java
@@ -305,8 +305,9 @@ public class StatusUpdateUtil {
String getRecordIdForMessage(Message message) {
if (message.getMsgType().equals(MessageType.STATE_TRANSITION)) {
- return message.getPartitionId() + " Trans:" + message.getTypedFromState().toString().charAt(0)
- + "->" + message.getTypedToState().toString().charAt(0) + " " + UUID.randomUUID().toString();
+ return message.getPartitionId() + " Trans:"
+ + message.getTypedFromState().toString().charAt(0) + "->"
+ + message.getTypedToState().toString().charAt(0) + " " + UUID.randomUUID().toString();
} else {
return message.getMsgType() + " " + UUID.randomUUID().toString();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java b/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
index 7aa2351..444c069 100644
--- a/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
+++ b/helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
@@ -174,7 +174,8 @@ public class ZkTestHelper {
while (dupZookeeper.getState() != States.CONNECTED) {
Thread.sleep(10);
}
- Assert.assertEquals(dupZookeeper.getState(), States.CONNECTED, "Fail to connect to zk using current session info");
+ Assert.assertEquals(dupZookeeper.getState(), States.CONNECTED,
+ "Fail to connect to zk using current session info");
dupZookeeper.close();
// make sure session expiry really happens
@@ -186,7 +187,8 @@ public class ZkTestHelper {
String newSessionId = Long.toHexString(curZookeeper.getSessionId());
LOG.info("After session expiry. sessionId: " + newSessionId + ", zk: " + curZookeeper);
- Assert.assertNotSame(newSessionId, oldSessionId, "Fail to expire current session, zk: " + curZookeeper);
+ Assert.assertNotSame(newSessionId, oldSessionId, "Fail to expire current session, zk: "
+ + curZookeeper);
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java b/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
index 2e17ad3..93f496e 100644
--- a/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
+++ b/helix-core/src/test/java/org/apache/helix/controller/stages/TestRebalancePipeline.java
@@ -174,7 +174,6 @@ public class TestRebalancePipeline extends ZkUnitTestBase {
ClusterAccessor clusterAccessor = new ClusterAccessor(ClusterId.from(clusterName), accessor);
clusterAccessor.initClusterStructure();
-
ClusterControllerManager controller =
new ClusterControllerManager(ZK_ADDR, clusterName, "controller_0");
controller.syncStart();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/healthcheck/TestAlertFireHistory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestAlertFireHistory.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestAlertFireHistory.java
index 8618b1c..798ce92 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestAlertFireHistory.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestAlertFireHistory.java
@@ -420,4 +420,3 @@ public class TestAlertFireHistory extends ZkStandAloneCMTestBaseWithPropertyServ
}
}
-
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
index 23741c3..a902158 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestExpandAlert.java
@@ -138,8 +138,7 @@ public class TestExpandAlert extends ZkIntegrationTestBase {
{
String instanceName = "localhost_" + (12918 + i);
- participants[i] =
- new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
+ participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
participants[i].setTransition(new ExpandAlertTransition());
participants[i].syncStart();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
index dbbd7aa..44937fe 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleAlert.java
@@ -143,8 +143,7 @@ public class TestSimpleAlert extends ZkIntegrationTestBase {
{
String instanceName = "localhost_" + (12918 + i);
- participants[i] =
- new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
+ participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
participants[i].setTransition(new SimpleAlertTransition(15));
participants[i].syncStart();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleWildcardAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleWildcardAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleWildcardAlert.java
index 90223ad..99c0608 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleWildcardAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestSimpleWildcardAlert.java
@@ -149,8 +149,7 @@ public class TestSimpleWildcardAlert extends ZkIntegrationTestBase {
{
String instanceName = "localhost_" + (12944 + i);
- participants[i] =
- new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
+ participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
participants[i].setTransition(new SimpleAlertTransition(i * 5));
participants[i].syncStart();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/healthcheck/TestStalenessAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestStalenessAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestStalenessAlert.java
index 76784d3..3227a65 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestStalenessAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestStalenessAlert.java
@@ -136,8 +136,7 @@ public class TestStalenessAlert extends ZkIntegrationTestBase {
{
String instanceName = "localhost_" + (12918 + i);
- participants[i] =
- new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
+ participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
participants[i].setTransition(new StalenessAlertTransition());
participants[i].syncStart();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/healthcheck/TestWildcardAlert.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/healthcheck/TestWildcardAlert.java b/helix-core/src/test/java/org/apache/helix/healthcheck/TestWildcardAlert.java
index 79c85ca..9677fcf 100644
--- a/helix-core/src/test/java/org/apache/helix/healthcheck/TestWildcardAlert.java
+++ b/helix-core/src/test/java/org/apache/helix/healthcheck/TestWildcardAlert.java
@@ -229,8 +229,7 @@ public class TestWildcardAlert extends ZkIntegrationTestBase {
{
String instanceName = "localhost_" + (12918 + i);
- participants[i] =
- new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
+ participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
participants[i].setTransition(new WildcardAlertTransition());
participants[i].syncStart();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/integration/TestAddNodeAfterControllerStart.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestAddNodeAfterControllerStart.java b/helix-core/src/test/java/org/apache/helix/integration/TestAddNodeAfterControllerStart.java
index 79d8b89..c7d83d0 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/TestAddNodeAfterControllerStart.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestAddNodeAfterControllerStart.java
@@ -99,8 +99,8 @@ public class TestAddNodeAfterControllerStart extends ZkIntegrationTestBase {
// setup grand cluster
final String grandClusterName = "GRAND_" + clusterName;
- TestHelper.setupCluster(grandClusterName, ZK_ADDR, 0, "controller", null, 0, 0, 1, 0,
- null, true);
+ TestHelper.setupCluster(grandClusterName, ZK_ADDR, 0, "controller", null, 0, 0, 1, 0, null,
+ true);
ClusterDistributedController distController =
new ClusterDistributedController(ZK_ADDR, grandClusterName, "controller_0");
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/integration/TestAddStateModelFactoryAfterConnect.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestAddStateModelFactoryAfterConnect.java b/helix-core/src/test/java/org/apache/helix/integration/TestAddStateModelFactoryAfterConnect.java
index 123ce6e..c48b6c9 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/TestAddStateModelFactoryAfterConnect.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestAddStateModelFactoryAfterConnect.java
@@ -115,8 +115,8 @@ public class TestAddStateModelFactoryAfterConnect extends ZkIntegrationTestBase
// register "TestDB1_Factory" state model factory
// Logger.getRootLogger().setLevel(Level.INFO);
for (int i = 0; i < n; i++) {
- participants[i].getStateMachineEngine()
- .registerStateModelFactory("MasterSlave", new MockMSModelFactory(), "TestDB1_Factory");
+ participants[i].getStateMachineEngine().registerStateModelFactory("MasterSlave",
+ new MockMSModelFactory(), "TestDB1_Factory");
}
result =
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/integration/TestAutoRebalance.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestAutoRebalance.java b/helix-core/src/test/java/org/apache/helix/integration/TestAutoRebalance.java
index 2aec114..cc43ecb 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/TestAutoRebalance.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestAutoRebalance.java
@@ -102,8 +102,7 @@ public class TestAutoRebalance extends ZkStandAloneCMTestBaseWithPropertyServerC
// start controller
String controllerName = CONTROLLER_PREFIX + "_0";
- _controller =
- new ClusterControllerManager(ZK_ADDR, CLUSTER_NAME, controllerName);
+ _controller = new ClusterControllerManager(ZK_ADDR, CLUSTER_NAME, controllerName);
_controller.syncStart();
boolean result =
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/1022d073/helix-core/src/test/java/org/apache/helix/integration/TestCleanupExternalView.java
----------------------------------------------------------------------
diff --git a/helix-core/src/test/java/org/apache/helix/integration/TestCleanupExternalView.java b/helix-core/src/test/java/org/apache/helix/integration/TestCleanupExternalView.java
index 93c58eb..071fd13 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/TestCleanupExternalView.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/TestCleanupExternalView.java
@@ -99,11 +99,11 @@ public class TestCleanupExternalView extends ZkUnitTestBase {
// System.out.println("remove current-state");
LiveInstance liveInstance = accessor.getProperty(keyBuilder.liveInstance("localhost_12918"));
- accessor.removeProperty(keyBuilder.currentState("localhost_12918", liveInstance.getTypedSessionId()
- .stringify(), "TestDB0"));
+ accessor.removeProperty(keyBuilder.currentState("localhost_12918", liveInstance
+ .getTypedSessionId().stringify(), "TestDB0"));
liveInstance = accessor.getProperty(keyBuilder.liveInstance("localhost_12919"));
- accessor.removeProperty(keyBuilder.currentState("localhost_12919", liveInstance.getTypedSessionId()
- .stringify(), "TestDB0"));
+ accessor.removeProperty(keyBuilder.currentState("localhost_12919", liveInstance
+ .getTypedSessionId().stringify(), "TestDB0"));
// re-enable controller shall remove orphan external-view
// System.out.println("re-enabling controller");