You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by zz...@apache.org on 2013/08/21 22:44:00 UTC
[47/51] [partial] [HELIX-198] Unify helix code style, rb=13710
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/ClusterMessagingService.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/ClusterMessagingService.java b/helix-core/src/main/java/org/apache/helix/ClusterMessagingService.java
index 6bab243..6ff76e9 100644
--- a/helix-core/src/main/java/org/apache/helix/ClusterMessagingService.java
+++ b/helix-core/src/main/java/org/apache/helix/ClusterMessagingService.java
@@ -28,18 +28,15 @@ import org.apache.helix.model.Message;
/**
* Provides the ability to <br>
- * <li>Send message to a specific component in the cluster[ participant,
- * controller, Spectator(probably not needed) ]</li> <li>Broadcast message to all
- * participants</li> <li>Send message to instances that hold a specific resource</li>
- * <li>Asynchronous request response api. Send message with a co-relation id and
- * invoke a method when there is a response. Can support timeout.</li>
- *
+ * <li>Send message to a specific component in the cluster[ participant, controller,
+ * Spectator(probably not needed) ]</li> <li>Broadcast message to all participants</li> <li>Send
+ * message to instances that hold a specific resource</li> <li>Asynchronous request response api.
+ * Send message with a co-relation id and invoke a method when there is a response. Can support
+ * timeout.</li>
*/
-public interface ClusterMessagingService
-{
+public interface ClusterMessagingService {
/**
* Send message matching the specifications mentioned in recipientCriteria.
- *
* @param receipientCriteria criteria to be met, defined as {@link Criteria}
* @See Criteria
* @param message
@@ -57,7 +54,6 @@ public interface ClusterMessagingService
* This method will return after sending the messages. <br>
* This is useful when message need to be sent and current thread need not
* wait for response since processing will be done in another thread.
- *
* @see #send(Criteria, Message)
* @param receipientCriteria
* @param message
@@ -65,8 +61,7 @@ public interface ClusterMessagingService
* @param timeOut Time to wait before failing the send
* @return the number of messages that were successfully sent
*/
- int send(Criteria receipientCriteria, Message message,
- AsyncCallback callbackOnReply, int timeOut);
+ int send(Criteria receipientCriteria, Message message, AsyncCallback callbackOnReply, int timeOut);
/**
* @see #send(Criteria, Message, AsyncCallback, int)
@@ -77,8 +72,8 @@ public interface ClusterMessagingService
* @param retryCount maximum number of times to retry the send
* @return the number of messages that were successfully sent
*/
- int send(Criteria receipientCriteria, Message message,
- AsyncCallback callbackOnReply, int timeOut, int retryCount);
+ int send(Criteria receipientCriteria, Message message, AsyncCallback callbackOnReply,
+ int timeOut, int retryCount);
/**
* This will send the message to all instances matching the criteria<br>
@@ -90,7 +85,6 @@ public interface ClusterMessagingService
* for response. <br>
* The current thread can use callbackOnReply instance to store application
* specific data.
- *
* @see #send(Criteria, Message, AsyncCallback, int)
* @param receipientCriteria
* @param message
@@ -99,8 +93,8 @@ public interface ClusterMessagingService
* @param retryCount
* @return the number of messages that were successfully sent
*/
- int sendAndWait(Criteria receipientCriteria, Message message,
- AsyncCallback callbackOnReply, int timeOut);
+ int sendAndWait(Criteria receipientCriteria, Message message, AsyncCallback callbackOnReply,
+ int timeOut);
/**
* @see #send(Criteria, Message, AsyncCallback, int, int)
@@ -111,8 +105,8 @@ public interface ClusterMessagingService
* @param retryCount
* @return the number of messages that were successfully sent
*/
- int sendAndWait(Criteria receipientCriteria, Message message,
- AsyncCallback callbackOnReply, int timeOut, int retryCount);
+ int sendAndWait(Criteria receipientCriteria, Message message, AsyncCallback callbackOnReply,
+ int timeOut, int retryCount);
/**
* This will register a message handler factory to create handlers for
@@ -121,24 +115,20 @@ public interface ClusterMessagingService
* Messages are processed in a threadpool which is hosted by cluster manager,
* and cluster manager will call the factory to create handler, and the
* handler is called in the threadpool.
- *
* Note that only one message handler factory can be registered with one
* message type.
- *
* @param type
* The message type that the factory will create handler for
* @param factory
* The per-type message factory
* @param threadpoolSize
- * size of the execution threadpool that handles the message
+ * size of the execution threadpool that handles the message
*/
- public void registerMessageHandlerFactory(String type,
- MessageHandlerFactory factory);
-
+ public void registerMessageHandlerFactory(String type, MessageHandlerFactory factory);
+
/**
* This will generate all messages to be sent given the recipientCriteria and MessageTemplate,
* the messages are not sent.
- *
* @param receipientCriteria criteria to be met, defined as {@link Criteria}
* @param messageTemplate the Message on which to base the messages to send
* @return messages to be sent, grouped by the type of instance to send the message to
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java b/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
index d3de9eb..34aef49 100644
--- a/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/ConfigAccessor.java
@@ -39,59 +39,47 @@ import org.apache.log4j.Logger;
* Provides access to the persistent configuration of the cluster, the instances that live on it,
* and the logical resources assigned to it.
*/
-public class ConfigAccessor
-{
- private static Logger LOG = Logger.getLogger(ConfigAccessor.class);
+public class ConfigAccessor {
+ private static Logger LOG = Logger.getLogger(ConfigAccessor.class);
private static final StringTemplate template = new StringTemplate();
- static
- {
+ static {
// @formatter:off
template.addEntry(ConfigScopeProperty.CLUSTER, 1, "/{clusterName}/CONFIGS/CLUSTER");
- template.addEntry(ConfigScopeProperty.CLUSTER,
- 2,
- "/{clusterName}/CONFIGS/CLUSTER/{clusterName}|SIMPLEKEYS");
- template.addEntry(ConfigScopeProperty.PARTICIPANT,
- 1,
- "/{clusterName}/CONFIGS/PARTICIPANT");
- template.addEntry(ConfigScopeProperty.PARTICIPANT,
- 2,
- "/{clusterName}/CONFIGS/PARTICIPANT/{participantName}|SIMPLEKEYS");
+ template.addEntry(ConfigScopeProperty.CLUSTER, 2,
+ "/{clusterName}/CONFIGS/CLUSTER/{clusterName}|SIMPLEKEYS");
+ template.addEntry(ConfigScopeProperty.PARTICIPANT, 1, "/{clusterName}/CONFIGS/PARTICIPANT");
+ template.addEntry(ConfigScopeProperty.PARTICIPANT, 2,
+ "/{clusterName}/CONFIGS/PARTICIPANT/{participantName}|SIMPLEKEYS");
template.addEntry(ConfigScopeProperty.RESOURCE, 1, "/{clusterName}/CONFIGS/RESOURCE");
- template.addEntry(ConfigScopeProperty.RESOURCE,
- 2,
- "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|SIMPLEKEYS");
- template.addEntry(ConfigScopeProperty.PARTITION,
- 2,
- "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|MAPKEYS");
- template.addEntry(ConfigScopeProperty.PARTITION,
- 3,
- "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|MAPMAPKEYS|{partitionName}");
+ template.addEntry(ConfigScopeProperty.RESOURCE, 2,
+ "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|SIMPLEKEYS");
+ template.addEntry(ConfigScopeProperty.PARTITION, 2,
+ "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|MAPKEYS");
+ template.addEntry(ConfigScopeProperty.PARTITION, 3,
+ "/{clusterName}/CONFIGS/RESOURCE/{resourceName}|MAPMAPKEYS|{partitionName}");
// @formatter:on
}
- private final ZkClient zkClient;
+ private final ZkClient zkClient;
/**
* Initialize an accessor with a Zookeeper client
* @param zkClient
*/
- public ConfigAccessor(ZkClient zkClient)
- {
+ public ConfigAccessor(ZkClient zkClient) {
this.zkClient = zkClient;
}
/**
* get config
* @deprecated replaced by {@link #get(HelixConfigScope, String)}
- *
* @param scope
* @param key
* @return value or null if doesn't exist
*/
@Deprecated
- public String get(ConfigScope scope, String key)
- {
+ public String get(ConfigScope scope, String key) {
Map<String, String> map = get(scope, Arrays.asList(key));
return map.get(key);
}
@@ -99,16 +87,13 @@ public class ConfigAccessor
/**
* get configs
* @deprecated replaced by {@link #get(HelixConfigScope, List<String>)}
- *
* @param scope
* @param keys
* @return
*/
@Deprecated
- public Map<String, String> get(ConfigScope scope, List<String> keys)
- {
- if (scope == null || scope.getScope() == null)
- {
+ public Map<String, String> get(ConfigScope scope, List<String> keys) {
+ if (scope == null || scope.getScope() == null) {
LOG.error("Scope can't be null");
return null;
}
@@ -116,8 +101,7 @@ public class ConfigAccessor
// String value = null;
Map<String, String> map = new HashMap<String, String>();
String clusterName = scope.getClusterName();
- if (!ZKUtil.isClusterSetup(clusterName, zkClient))
- {
+ if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
throw new HelixException("cluster " + clusterName + " is not setup yet");
}
@@ -126,20 +110,15 @@ public class ConfigAccessor
ZNRecord record = zkClient.readData(splits[0], true);
- if (record != null)
- {
- if (splits.length == 1)
- {
+ if (record != null) {
+ if (splits.length == 1) {
for (String key : keys) {
if (record.getSimpleFields().containsKey(key)) {
map.put(key, record.getSimpleField(key));
}
}
- }
- else if (splits.length == 2)
- {
- if (record.getMapField(splits[1]) != null)
- {
+ } else if (splits.length == 2) {
+ if (record.getMapField(splits[1]) != null) {
for (String key : keys) {
if (record.getMapField(splits[1]).containsKey(key)) {
map.put(key, record.getMapField(splits[1]).get(key));
@@ -154,9 +133,8 @@ public class ConfigAccessor
/**
* get a single config entry
- *
* @param scope specification of the entity set to query
- * (e.g. cluster, resource, participant, etc.)
+ * (e.g. cluster, resource, participant, etc.)
* @param key the identifier of the configuration entry
* @return the configuration entry
*/
@@ -170,14 +148,13 @@ public class ConfigAccessor
/**
* get many config entries
- *
* @param scope scope specification of the entity set to query
- * (e.g. cluster, resource, participant, etc.)
+ * (e.g. cluster, resource, participant, etc.)
* @param keys the identifiers of the configuration entries
* @return the configuration entries, organized by key
*/
public Map<String, String> get(HelixConfigScope scope, List<String> keys) {
- if (scope == null || scope.getType()== null || !scope.isFullKey()) {
+ if (scope == null || scope.getType() == null || !scope.isFullKey()) {
LOG.error("fail to get configs. invalid config scope. scope: " + scope + ", keys: " + keys);
return null;
}
@@ -222,7 +199,6 @@ public class ConfigAccessor
/**
* Set config, create if not exist
* @deprecated replaced by {@link #set(HelixConfigScope, String, String)}
- *
* @param scope
* @param key
* @param value
@@ -237,31 +213,29 @@ public class ConfigAccessor
/**
* Set configs, create if not exist
* @deprecated replaced by {@link #set(HelixConfigScope, Map<String, String>)}
- *
* @param scope
* @param keyValueMap
*/
@Deprecated
- public void set(ConfigScope scope, Map<String, String> keyValueMap)
- {
- if (scope == null || scope.getScope() == null)
- {
+ public void set(ConfigScope scope, Map<String, String> keyValueMap) {
+ if (scope == null || scope.getScope() == null) {
LOG.error("Scope can't be null");
return;
}
String clusterName = scope.getClusterName();
- if (!ZKUtil.isClusterSetup(clusterName, zkClient))
- {
+ if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
throw new HelixException("cluster: " + clusterName + " is NOT setup.");
}
if (scope.getScope() == ConfigScopeProperty.PARTICIPANT) {
- String scopeStr = scope.getScopeStr();
- String instanceName = scopeStr.substring(scopeStr.lastIndexOf('/') + 1);
- if (!ZKUtil.isInstanceSetup(zkClient, scope.getClusterName(), instanceName, InstanceType.PARTICIPANT)) {
- throw new HelixException("instance: " + instanceName + " is NOT setup in cluster: " + clusterName);
- }
+ String scopeStr = scope.getScopeStr();
+ String instanceName = scopeStr.substring(scopeStr.lastIndexOf('/') + 1);
+ if (!ZKUtil.isInstanceSetup(zkClient, scope.getClusterName(), instanceName,
+ InstanceType.PARTICIPANT)) {
+ throw new HelixException("instance: " + instanceName + " is NOT setup in cluster: "
+ + clusterName);
+ }
}
// use "|" to delimit resource and partition. e.g. /MyCluster/CONFIGS/PARTICIPANT/MyDB|MyDB_0
@@ -270,20 +244,16 @@ public class ConfigAccessor
String id = splits[0].substring(splits[0].lastIndexOf('/') + 1);
ZNRecord update = new ZNRecord(id);
- if (splits.length == 1)
- {
- for (String key: keyValueMap.keySet()) {
+ if (splits.length == 1) {
+ for (String key : keyValueMap.keySet()) {
String value = keyValueMap.get(key);
update.setSimpleField(key, value);
}
- }
- else if (splits.length == 2)
- {
- if (update.getMapField(splits[1]) == null)
- {
+ } else if (splits.length == 2) {
+ if (update.getMapField(splits[1]) == null) {
update.setMapField(splits[1], new TreeMap<String, String>());
}
- for (String key: keyValueMap.keySet()) {
+ for (String key : keyValueMap.keySet()) {
String value = keyValueMap.get(key);
update.getMapField(splits[1]).put(key, value);
}
@@ -294,9 +264,8 @@ public class ConfigAccessor
/**
* Set config, creating it if it doesn't exist
- *
* @param scope scope specification of the entity set to query
- * (e.g. cluster, resource, participant, etc.)
+ * (e.g. cluster, resource, participant, etc.)
* @param key the identifier of the configuration entry
* @param value the configuration
*/
@@ -308,13 +277,11 @@ public class ConfigAccessor
/**
* Set multiple configs, creating them if they don't exist
- *
* @param scope scope specification of the entity set to query
- * (e.g. cluster, resource, participant, etc.)
+ * (e.g. cluster, resource, participant, etc.)
* @param keyValueMap configurations organized by their identifiers
*/
- public void set(HelixConfigScope scope, Map<String, String> keyValueMap)
- {
+ public void set(HelixConfigScope scope, Map<String, String> keyValueMap) {
if (scope == null || scope.getType() == null || !scope.isFullKey()) {
LOG.error("fail to set config. invalid config scope. scope: " + scope);
return;
@@ -326,11 +293,11 @@ public class ConfigAccessor
}
if (scope.getType() == ConfigScopeProperty.PARTICIPANT) {
- if (!ZKUtil.isInstanceSetup(zkClient, scope.getClusterName(), scope.getParticipantName(),
- InstanceType.PARTICIPANT)) {
- throw new HelixException("fail to set config. instance: " + scope.getParticipantName()
- + " is NOT setup in cluster: " + clusterName);
- }
+ if (!ZKUtil.isInstanceSetup(zkClient, scope.getClusterName(), scope.getParticipantName(),
+ InstanceType.PARTICIPANT)) {
+ throw new HelixException("fail to set config. instance: " + scope.getParticipantName()
+ + " is NOT setup in cluster: " + clusterName);
+ }
}
String zkPath = scope.getZkPath();
@@ -340,7 +307,7 @@ public class ConfigAccessor
if (mapKey == null) {
update.getSimpleFields().putAll(keyValueMap);
} else {
- update.setMapField(mapKey, keyValueMap);
+ update.setMapField(mapKey, keyValueMap);
}
ZKUtil.createOrUpdate(zkClient, zkPath, update, true, true);
return;
@@ -349,7 +316,6 @@ public class ConfigAccessor
/**
* Remove config
* @deprecated replaced by {@link #remove(HelixConfigScope, String)}
- *
* @param scope
* @param key
*/
@@ -361,22 +327,18 @@ public class ConfigAccessor
/**
* remove configs
* @deprecated replaced by {@link #remove(HelixConfigScope, List<String>)}
- *
* @param scope
* @param keys
*/
@Deprecated
- public void remove(ConfigScope scope, List<String> keys)
- {
- if (scope == null || scope.getScope() == null)
- {
+ public void remove(ConfigScope scope, List<String> keys) {
+ if (scope == null || scope.getScope() == null) {
LOG.error("Scope can't be null");
return;
}
String clusterName = scope.getClusterName();
- if (!ZKUtil.isClusterSetup(clusterName, zkClient))
- {
+ if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
throw new HelixException("cluster " + clusterName + " is not setup yet");
}
@@ -385,17 +347,13 @@ public class ConfigAccessor
String id = splits[0].substring(splits[0].lastIndexOf('/') + 1);
ZNRecord update = new ZNRecord(id);
- if (splits.length == 1)
- {
+ if (splits.length == 1) {
// subtract doesn't care about value, use empty string
for (String key : keys) {
update.setSimpleField(key, "");
}
- }
- else if (splits.length == 2)
- {
- if (update.getMapField(splits[1]) == null)
- {
+ } else if (splits.length == 2) {
+ if (update.getMapField(splits[1]) == null) {
update.setMapField(splits[1], new TreeMap<String, String>());
}
// subtract doesn't care about value, use empty string
@@ -410,9 +368,8 @@ public class ConfigAccessor
/**
* Remove a single config
- *
* @param scope scope specification of the entity set to query
- * (e.g. cluster, resource, participant, etc.)
+ * (e.g. cluster, resource, participant, etc.)
* @param key the identifier of the configuration entry
*/
public void remove(HelixConfigScope scope, String key) {
@@ -421,13 +378,11 @@ public class ConfigAccessor
/**
* Remove multiple configs
- *
* @param scope scope specification of the entity set to query
- * (e.g. cluster, resource, participant, etc.)
+ * (e.g. cluster, resource, participant, etc.)
* @param keys the identifiers of the configuration entries
*/
- public void remove(HelixConfigScope scope, List<String> keys)
- {
+ public void remove(HelixConfigScope scope, List<String> keys) {
if (scope == null || scope.getType() == null || !scope.isFullKey()) {
LOG.error("fail to remove. invalid scope: " + scope + ", keys: " + keys);
return;
@@ -462,27 +417,20 @@ public class ConfigAccessor
/**
* get config keys
* @deprecated replaced by {@link #getKeys(HelixConfigScope)}
- *
* @param type
* @param clusterName
* @param keys
* @return
*/
@Deprecated
- public List<String> getKeys(ConfigScopeProperty type,
- String clusterName,
- String... keys)
- {
- if (type == null || clusterName == null)
- {
+ public List<String> getKeys(ConfigScopeProperty type, String clusterName, String... keys) {
+ if (type == null || clusterName == null) {
LOG.error("clusterName|scope can't be null");
return Collections.emptyList();
}
- try
- {
- if (!ZKUtil.isClusterSetup(clusterName, zkClient))
- {
+ try {
+ if (!ZKUtil.isClusterSetup(clusterName, zkClient)) {
LOG.error("cluster " + clusterName + " is not setup yet");
return Collections.emptyList();
}
@@ -493,48 +441,35 @@ public class ConfigAccessor
String scopeStr = template.instantiate(type, args);
String[] splits = scopeStr.split("\\|");
List<String> retKeys = null;
- if (splits.length == 1)
- {
+ if (splits.length == 1) {
retKeys = zkClient.getChildren(splits[0]);
- }
- else
- {
+ } else {
ZNRecord record = zkClient.readData(splits[0]);
- if (splits[1].startsWith("SIMPLEKEYS"))
- {
+ if (splits[1].startsWith("SIMPLEKEYS")) {
retKeys = new ArrayList<String>(record.getSimpleFields().keySet());
- }
- else if (splits[1].startsWith("MAPKEYS"))
- {
+ } else if (splits[1].startsWith("MAPKEYS")) {
retKeys = new ArrayList<String>(record.getMapFields().keySet());
- }
- else if (splits[1].startsWith("MAPMAPKEYS"))
- {
+ } else if (splits[1].startsWith("MAPMAPKEYS")) {
retKeys = new ArrayList<String>(record.getMapField(splits[2]).keySet());
}
}
- if (retKeys == null)
- {
+ if (retKeys == null) {
LOG.error("Invalid scope: " + type + " or keys: " + Arrays.toString(args));
return Collections.emptyList();
}
Collections.sort(retKeys);
return retKeys;
- }
- catch (Exception e)
- {
+ } catch (Exception e) {
return Collections.emptyList();
}
}
-
/**
* Get list of config keys for a scope
- *
* @param scope
* @return a list of configuration keys
*/
@@ -544,7 +479,7 @@ public class ConfigAccessor
return null;
}
- if (!ZKUtil.isClusterSetup(scope.getClusterName(), zkClient)){
+ if (!ZKUtil.isClusterSetup(scope.getClusterName(), zkClient)) {
LOG.error("fail to getKeys. cluster " + scope.getClusterName() + " is not setup yet");
return Collections.emptyList();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/ConfigChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/ConfigChangeListener.java b/helix-core/src/main/java/org/apache/helix/ConfigChangeListener.java
index b9083ab..1dbf2fe 100644
--- a/helix-core/src/main/java/org/apache/helix/ConfigChangeListener.java
+++ b/helix-core/src/main/java/org/apache/helix/ConfigChangeListener.java
@@ -23,21 +23,16 @@ import java.util.List;
import org.apache.helix.model.InstanceConfig;
-
/**
- *
* @deprecated replaced by InstanceConfigChangeListener
*/
-public interface ConfigChangeListener
-{
+public interface ConfigChangeListener {
/**
* Invoked when participant config changes
- *
* @param configs
* @param changeContext
*/
- public void onConfigChange(List<InstanceConfig> configs,
- NotificationContext changeContext);
+ public void onConfigChange(List<InstanceConfig> configs, NotificationContext changeContext);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/ControllerChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/ControllerChangeListener.java b/helix-core/src/main/java/org/apache/helix/ControllerChangeListener.java
index 104f0bc..8e17703 100644
--- a/helix-core/src/main/java/org/apache/helix/ControllerChangeListener.java
+++ b/helix-core/src/main/java/org/apache/helix/ControllerChangeListener.java
@@ -22,11 +22,9 @@ package org.apache.helix;
/**
* Interface to implement to respond to controller changes.
*/
-public interface ControllerChangeListener
-{
+public interface ControllerChangeListener {
/**
* Invoked when controller changes
- *
* @param changeContext description of the event and state
*/
public void onControllerChange(NotificationContext changeContext);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/Criteria.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/Criteria.java b/helix-core/src/main/java/org/apache/helix/Criteria.java
index 101420c..da2b36e 100644
--- a/helix-core/src/main/java/org/apache/helix/Criteria.java
+++ b/helix-core/src/main/java/org/apache/helix/Criteria.java
@@ -22,12 +22,12 @@ package org.apache.helix;
/**
* Describes various properties that operations involving {@link Message} delivery will follow.
*/
-public class Criteria
-{
- public enum DataSource
- {
- IDEALSTATES, EXTERNALVIEW
+public class Criteria {
+ public enum DataSource {
+ IDEALSTATES,
+ EXTERNALVIEW
}
+
/**
* This can be CONTROLLER, PARTICIPANT, ROUTER Cannot be null
*/
@@ -69,8 +69,7 @@ public class Criteria
* Get the current source of truth
* @return either the ideal state or the external view
*/
- public DataSource getDataSource()
- {
+ public DataSource getDataSource() {
return _dataSource;
}
@@ -78,8 +77,7 @@ public class Criteria
* Set the current source of truth
* @param source ideal state or external view
*/
- public void setDataSource(DataSource source)
- {
+ public void setDataSource(DataSource source) {
_dataSource = source;
}
@@ -87,8 +85,7 @@ public class Criteria
* Determine if the message is excluded from being sent to the sender
* @return true if the self-sent message is excluded, false otherwise
*/
- public boolean isSelfExcluded()
- {
+ public boolean isSelfExcluded() {
return selfExcluded;
}
@@ -96,8 +93,7 @@ public class Criteria
* Indicate whether or not the sender will be excluded as a message recipient
* @param selfExcluded true if the sender should be excluded, false otherwise
*/
- public void setSelfExcluded(boolean selfExcluded)
- {
+ public void setSelfExcluded(boolean selfExcluded) {
this.selfExcluded = selfExcluded;
}
@@ -105,8 +101,7 @@ public class Criteria
* Determine the type of the recipient
* @return InstanceType (e.g. PARTICIPANT, CONTROLLER, SPECTATOR)
*/
- public InstanceType getRecipientInstanceType()
- {
+ public InstanceType getRecipientInstanceType() {
return recipientInstanceType;
}
@@ -114,8 +109,7 @@ public class Criteria
* Set the type of the recipient
* @param recipientInstanceType InstanceType (e.g. PARTICIPANT, CONTROLLER, SPECTATOR)
*/
- public void setRecipientInstanceType(InstanceType recipientInstanceType)
- {
+ public void setRecipientInstanceType(InstanceType recipientInstanceType) {
this.recipientInstanceType = recipientInstanceType;
}
@@ -123,18 +117,16 @@ public class Criteria
* Determine if this message should be processed only if an instance was up at send time
* @return true if the message will be processed by current live nodes, false otherwise
*/
- public boolean isSessionSpecific()
- {
+ public boolean isSessionSpecific() {
return sessionSpecific;
}
/**
* Indicate whether or not a message should be restricted to a session
* @param sessionSpecific true if the message can only be processed by live nodes at send time,
- * false otherwise
+ * false otherwise
*/
- public void setSessionSpecific(boolean sessionSpecific)
- {
+ public void setSessionSpecific(boolean sessionSpecific) {
this.sessionSpecific = sessionSpecific;
}
@@ -142,8 +134,7 @@ public class Criteria
* Get the name of the destination instance, available only for PARTICIPANT
* @return the instance name
*/
- public String getInstanceName()
- {
+ public String getInstanceName() {
return instanceName;
}
@@ -151,8 +142,7 @@ public class Criteria
* Set the name of the destination instance (PARTICIPANT only)
* @param instanceName the instance name or * for all instances
*/
- public void setInstanceName(String instanceName)
- {
+ public void setInstanceName(String instanceName) {
this.instanceName = instanceName;
}
@@ -160,8 +150,7 @@ public class Criteria
* Get the destination resource name
* @return destination resource name
*/
- public String getResource()
- {
+ public String getResource() {
return resourceName;
}
@@ -169,8 +158,7 @@ public class Criteria
* Set the destination resource name
* @param resourceName the resource name or * for all resources
*/
- public void setResource(String resourceName)
- {
+ public void setResource(String resourceName) {
this.resourceName = resourceName;
}
@@ -178,8 +166,7 @@ public class Criteria
* Get the destination partition name
* @return destination partition name
*/
- public String getPartition()
- {
+ public String getPartition() {
return partitionName;
}
@@ -187,8 +174,7 @@ public class Criteria
* Set the destination partition name
* @param partitionName the partition name, or * for all partitions of a resource
*/
- public void setPartition(String partitionName)
- {
+ public void setPartition(String partitionName) {
this.partitionName = partitionName;
}
@@ -196,8 +182,7 @@ public class Criteria
* Get the state of a resource partition
* @return the state of the resource partition
*/
- public String getPartitionState()
- {
+ public String getPartitionState() {
return partitionState;
}
@@ -205,14 +190,12 @@ public class Criteria
* Set the state of the resource partition
* @param partitionState the state of the resource partition
*/
- public void setPartitionState(String partitionState)
- {
+ public void setPartitionState(String partitionState) {
this.partitionState = partitionState;
}
@Override
- public String toString()
- {
+ public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("instanceName").append("=").append(instanceName);
sb.append("resourceName").append("=").append(resourceName);
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/CurrentStateChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/CurrentStateChangeListener.java b/helix-core/src/main/java/org/apache/helix/CurrentStateChangeListener.java
index d244600..9adafea 100644
--- a/helix-core/src/main/java/org/apache/helix/CurrentStateChangeListener.java
+++ b/helix-core/src/main/java/org/apache/helix/CurrentStateChangeListener.java
@@ -26,18 +26,15 @@ import org.apache.helix.model.CurrentState;
/**
* Interface to implement to respond to changes in the current state
*/
-public interface CurrentStateChangeListener
-{
+public interface CurrentStateChangeListener {
/**
* Invoked when current state changes
- *
* @param instanceName name of the instance whose state changed
* @param statesInfo a list of the current states
* @param changeContext the change event and state
*/
- public void onStateChange(String instanceName,
- List<CurrentState> statesInfo,
- NotificationContext changeContext);
+ public void onStateChange(String instanceName, List<CurrentState> statesInfo,
+ NotificationContext changeContext);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/ExternalCommand.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/ExternalCommand.java b/helix-core/src/main/java/org/apache/helix/ExternalCommand.java
index b24085d..5eb89f1 100644
--- a/helix-core/src/main/java/org/apache/helix/ExternalCommand.java
+++ b/helix-core/src/main/java/org/apache/helix/ExternalCommand.java
@@ -37,8 +37,7 @@ import org.apache.log4j.Logger;
* Wrapper for running commands outside of the JVM
* @see {@link Process}
*/
-public class ExternalCommand
-{
+public class ExternalCommand {
public static final String MODULE = ExternalCommand.class.getName();
public static final Logger LOG = Logger.getLogger(MODULE);
@@ -51,64 +50,55 @@ public class ExternalCommand
/**
* Stream redirector
*/
- private static class InputReader extends Thread
- {
+ private static class InputReader extends Thread {
private static final int BUFFER_SIZE = 2048;
private final InputStream _in;
private final ByteArrayOutputStream _out;
private boolean _running = false;
- InputReader(InputStream in)
- {
+ InputReader(InputStream in) {
_in = in;
_out = new ByteArrayOutputStream();
}
@Override
- public void run()
- {
+ public void run() {
_running = true;
byte[] buf = new byte[BUFFER_SIZE];
int n = 0;
- try
- {
- while((n = _in.read(buf)) != -1)
+ try {
+ while ((n = _in.read(buf)) != -1)
_out.write(buf, 0, n);
- }
- catch(IOException e)
- {
+ } catch (IOException e) {
LOG.error("error while reading external command", e);
}
_running = false;
}
- public byte[] getOutput()
- {
- if(_running)
+ public byte[] getOutput() {
+ if (_running)
throw new IllegalStateException("wait for process to be completed");
return _out.toByteArray();
}
}
+
/**
* Initialize with a {@link ProcessBuilder}
* @param processBuilder initialized {@link ProcessBuilder} object
*/
- public ExternalCommand(ProcessBuilder processBuilder)
- {
+ public ExternalCommand(ProcessBuilder processBuilder) {
_processBuilder = processBuilder;
}
/**
-* After creating the command, you have to start it...
-*
-* @throws IOException
-*/
- public void start() throws IOException
- {
+ * After creating the command, you have to start it...
+ * @throws IOException
+ */
+ public void start() throws IOException {
_process = _processBuilder.start();
_out = new InputReader(new BufferedInputStream(_process.getInputStream()));
_err = new InputReader(new BufferedInputStream(_process.getErrorStream()));
@@ -118,42 +108,37 @@ public class ExternalCommand
}
/**
-* @see {@link ProcessBuilder#environment()}
-*/
- public Map<String, String> getEnvironment()
- {
+ * @see {@link ProcessBuilder#environment()}
+ */
+ public Map<String, String> getEnvironment() {
return _processBuilder.environment();
}
/**
-* @see {@link ProcessBuilder#directory()}
-*/
- public File getWorkingDirectory()
- {
+ * @see {@link ProcessBuilder#directory()}
+ */
+ public File getWorkingDirectory() {
return _processBuilder.directory();
}
/**
-* @see {@link ProcessBuilder#directory(File)}
-*/
- public void setWorkingDirectory(File directory)
- {
+ * @see {@link ProcessBuilder#directory(File)}
+ */
+ public void setWorkingDirectory(File directory) {
_processBuilder.directory(directory);
}
/**
-* @see {@link ProcessBuilder#redirectErrorStream()}
-*/
- public boolean getRedirectErrorStream()
- {
+ * @see {@link ProcessBuilder#redirectErrorStream()}
+ */
+ public boolean getRedirectErrorStream() {
return _processBuilder.redirectErrorStream();
}
/**
-* @see {@link ProcessBuilder#redirectErrorStream(boolean)}
-*/
- public void setRedirectErrorStream(boolean redirectErrorStream)
- {
+ * @see {@link ProcessBuilder#redirectErrorStream(boolean)}
+ */
+ public void setRedirectErrorStream(boolean redirectErrorStream) {
_processBuilder.redirectErrorStream(redirectErrorStream);
}
@@ -162,8 +147,7 @@ public class ExternalCommand
* @return bytes from the output stream
* @throws InterruptedException the process was interrupted before completion
*/
- public byte[] getOutput() throws InterruptedException
- {
+ public byte[] getOutput() throws InterruptedException {
waitFor();
return _out.getOutput();
}
@@ -173,89 +157,72 @@ public class ExternalCommand
* @return bytes from the error stream
* @throws InterruptedException the process was interrupted before completion
*/
- public byte[] getError() throws InterruptedException
- {
+ public byte[] getError() throws InterruptedException {
waitFor();
return _err.getOutput();
}
/**
-* Returns the output as a string.
-*
-* @param encoding string encoding scheme, e.g. "UTF-8"
-* @return encoded string
-* @throws InterruptedException the process was interrupted before completion
-* @throws UnsupportedEncodingException the encoding scheme is invalid
-*/
+ * Returns the output as a string.
+ * @param encoding string encoding scheme, e.g. "UTF-8"
+ * @return encoded string
+ * @throws InterruptedException the process was interrupted before completion
+ * @throws UnsupportedEncodingException the encoding scheme is invalid
+ */
public String getStringOutput(String encoding) throws InterruptedException,
- UnsupportedEncodingException
- {
+ UnsupportedEncodingException {
return new String(getOutput(), encoding);
}
/**
-* Returns the output as a string. Uses encoding "UTF-8".
-*
-* @return utf8 encoded string
-* @throws InterruptedException the process was interrupted before completion
-*/
- public String getStringOutput() throws InterruptedException
- {
- try
- {
+ * Returns the output as a string. Uses encoding "UTF-8".
+ * @return utf8 encoded string
+ * @throws InterruptedException the process was interrupted before completion
+ */
+ public String getStringOutput() throws InterruptedException {
+ try {
return getStringOutput("UTF-8");
- }
- catch(UnsupportedEncodingException e)
- {
+ } catch (UnsupportedEncodingException e) {
// should not happen
throw new RuntimeException(e);
}
}
/**
-* Returns the error as a string.
-*
-* @param encoding the encoding scheme, e.g. "UTF-8"
-* @return error as string
-* @throws InterruptedException the process was interrupted before completion
-* @throws UnsupportedEncodingException the encoding scheme is invalid
-*/
+ * Returns the error as a string.
+ * @param encoding the encoding scheme, e.g. "UTF-8"
+ * @return error as string
+ * @throws InterruptedException the process was interrupted before completion
+ * @throws UnsupportedEncodingException the encoding scheme is invalid
+ */
public String getStringError(String encoding) throws InterruptedException,
- UnsupportedEncodingException
- {
+ UnsupportedEncodingException {
return new String(getError(), encoding);
}
/**
-* Returns the error as a string. Uses encoding "UTF-8".
-*
-* @return error as string
-* @throws InterruptedException the process was interrupted before completion
-*/
- public String getStringError() throws InterruptedException
- {
- try
- {
+ * Returns the error as a string. Uses encoding "UTF-8".
+ * @return error as string
+ * @throws InterruptedException the process was interrupted before completion
+ */
+ public String getStringError() throws InterruptedException {
+ try {
return getStringError("UTF-8");
- }
- catch(UnsupportedEncodingException e)
- {
+ } catch (UnsupportedEncodingException e) {
// should not happen
throw new RuntimeException(e);
}
}
/**
-* Properly waits until everything is complete: joins on the thread that
-* reads the output, joins on the thread that reads the error and finally
-* wait for the process to be finished.
-* @return the status code of the process.
-*
-* @throws InterruptedException the process was interrupted before completion
-*/
- public int waitFor() throws InterruptedException
- {
- if(_process == null)
+ * Properly waits until everything is complete: joins on the thread that
+ * reads the output, joins on the thread that reads the error and finally
+ * wait for the process to be finished.
+ * @return the status code of the process.
+ * @throws InterruptedException the process was interrupted before completion
+ */
+ public int waitFor() throws InterruptedException {
+ if (_process == null)
throw new IllegalStateException("you must call start first");
_out.join();
@@ -264,28 +231,25 @@ public class ExternalCommand
}
/**
-* Properly waits until everything is complete: joins on the thread that
-* reads the output, joins on the thread that reads the error and finally
-* wait for the process to be finished.
-* If the process has not completed before the timeout, throws a
-* {@link TimeoutException}
-* @return the status code of the process.
-*
-* @throws TimeoutException the process timed out
-* @throws InterruptedException the process was interrupted before completion
-*/
- public int waitFor(long timeout) throws InterruptedException, TimeoutException
- {
- if(_process == null)
+ * Properly waits until everything is complete: joins on the thread that
+ * reads the output, joins on the thread that reads the error and finally
+ * wait for the process to be finished.
+ * If the process has not completed before the timeout, throws a {@link TimeoutException}
+ * @return the status code of the process.
+ * @throws TimeoutException the process timed out
+ * @throws InterruptedException the process was interrupted before completion
+ */
+ public int waitFor(long timeout) throws InterruptedException, TimeoutException {
+ if (_process == null)
throw new IllegalStateException("you must call start first");
-// Chronos c = new Chronos();
+ // Chronos c = new Chronos();
_out.join(timeout);
-// timeout -= c.tick();
+ // timeout -= c.tick();
if (timeout <= 0)
throw new TimeoutException("Wait timed out");
_err.join(timeout);
-// timeout -= c.tick();
+ // timeout -= c.tick();
if (timeout <= 0)
throw new TimeoutException("Wait timed out");
@@ -298,9 +262,8 @@ public class ExternalCommand
* @see {@link Process#exitValue()}
* @return the return code of the process
*/
- public int exitValue()
- {
- if(_process == null)
+ public int exitValue() {
+ if (_process == null)
throw new IllegalStateException("you must call start first");
return _process.exitValue();
@@ -309,97 +272,81 @@ public class ExternalCommand
/**
* see {@link Process#destroy()}
*/
- public void destroy()
- {
- if(_process == null)
+ public void destroy() {
+ if (_process == null)
throw new IllegalStateException("you must call start first");
_process.destroy();
}
/**
-* Creates an external process from the command. It is not started and you have to call
-* start on it!
-*
-* @param commands the command to execute
-* @return the process */
- public static ExternalCommand create(String... commands)
- {
+ * Creates an external process from the command. It is not started and you have to call
+ * start on it!
+ * @param commands the command to execute
+ * @return the process
+ */
+ public static ExternalCommand create(String... commands) {
ExternalCommand ec = new ExternalCommand(new ProcessBuilder(commands));
return ec;
}
/**
-* Creates an external process from the command. It is not started and you have to call
-* start on it!
-*
-* @param commands the command to execute
-* @return the process */
- public static ExternalCommand create(List<String> commands)
- {
+ * Creates an external process from the command. It is not started and you have to call
+ * start on it!
+ * @param commands the command to execute
+ * @return the process
+ */
+ public static ExternalCommand create(List<String> commands) {
ExternalCommand ec = new ExternalCommand(new ProcessBuilder(commands));
return ec;
}
/**
-* Creates an external process from the command. The command is executed.
-*
-* @param commands the commands to execute
-* @return the process
-* @throws IOException if there is an error */
- public static ExternalCommand start(String... commands) throws IOException
- {
+ * Creates an external process from the command. The command is executed.
+ * @param commands the commands to execute
+ * @return the process
+ * @throws IOException if there is an error
+ */
+ public static ExternalCommand start(String... commands) throws IOException {
ExternalCommand ec = new ExternalCommand(new ProcessBuilder(commands));
ec.start();
return ec;
}
/**
- * Executes the external command in the given working directory and waits for it to be
- * finished.
- *
- * @param workingDirectory the root directory from where to run the command
- * @param command the command to execute (should be relative to the working directory
- * @param args the arguments to the command
- * @return the process
- */
- public static ExternalCommand execute(File workingDirectory,
- String command,
- String... args)
- throws IOException, InterruptedException
- {
- try
- {
+ * Executes the external command in the given working directory and waits for it to be
+ * finished.
+ * @param workingDirectory the root directory from where to run the command
+ * @param command the command to execute (should be relative to the working directory
+ * @param args the arguments to the command
+ * @return the process
+ */
+ public static ExternalCommand execute(File workingDirectory, String command, String... args)
+ throws IOException, InterruptedException {
+ try {
return executeWithTimeout(workingDirectory, command, 0, args);
- }
- catch (TimeoutException e)
- {
+ } catch (TimeoutException e) {
// Can't happen!
throw new IllegalStateException(MODULE + ".execute: Unexpected timeout occurred!");
}
}
-/**
- * Executes the external command in the given working directory and waits (until timeout
- * is elapsed) for it to be finished.
- *
- * @param workingDirectory
- * the root directory from where to run the command
- * @param command
- * the command to execute (should be relative to the working directory
- * @param timeout
- * the maximum amount of time to wait for this external command (in ms). If
- * this value is less than or equal to 0, timeout is ignored
- * @param args
- * the arguments to the command
- * @return the process
- */
- public static ExternalCommand executeWithTimeout(File workingDirectory,
- String command,
- long timeout,
- String... args)
- throws IOException, InterruptedException, TimeoutException
- {
+ /**
+ * Executes the external command in the given working directory and waits (until timeout
+ * is elapsed) for it to be finished.
+ * @param workingDirectory
+ * the root directory from where to run the command
+ * @param command
+ * the command to execute (should be relative to the working directory
+ * @param timeout
+ * the maximum amount of time to wait for this external command (in ms). If
+ * this value is less than or equal to 0, timeout is ignored
+ * @param args
+ * the arguments to the command
+ * @return the process
+ */
+ public static ExternalCommand executeWithTimeout(File workingDirectory, String command,
+ long timeout, String... args) throws IOException, InterruptedException, TimeoutException {
List<String> arguments = new ArrayList<String>(args.length + 1);
arguments.add(new File(workingDirectory, command).getAbsolutePath());
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/ExternalViewChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/ExternalViewChangeListener.java b/helix-core/src/main/java/org/apache/helix/ExternalViewChangeListener.java
index 9aff2da..024cc88 100644
--- a/helix-core/src/main/java/org/apache/helix/ExternalViewChangeListener.java
+++ b/helix-core/src/main/java/org/apache/helix/ExternalViewChangeListener.java
@@ -26,16 +26,14 @@ import org.apache.helix.model.ExternalView;
/**
* Interface to implement to be notified of changes to the external view
*/
-public interface ExternalViewChangeListener
-{
+public interface ExternalViewChangeListener {
/**
* Invoked when external view changes
- *
* @param externalViewList a list of ExternalViews
* @param changeContext the change event and state
*/
public void onExternalViewChange(List<ExternalView> externalViewList,
- NotificationContext changeContext);
+ NotificationContext changeContext);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/GroupCommit.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/GroupCommit.java b/helix-core/src/main/java/org/apache/helix/GroupCommit.java
index f90ec9d..4d1bf68 100644
--- a/helix-core/src/main/java/org/apache/helix/GroupCommit.java
+++ b/helix-core/src/main/java/org/apache/helix/GroupCommit.java
@@ -32,48 +32,42 @@ import org.apache.log4j.Logger;
/**
* Support committing updates to data such that they are ordered for each key
*/
-public class GroupCommit
-{
- private static Logger LOG = Logger.getLogger(GroupCommit.class);
- private static class Queue
- {
- final AtomicReference<Thread> _running = new AtomicReference<Thread>();
+public class GroupCommit {
+ private static Logger LOG = Logger.getLogger(GroupCommit.class);
+
+ private static class Queue {
+ final AtomicReference<Thread> _running = new AtomicReference<Thread>();
final ConcurrentLinkedQueue<Entry> _pending = new ConcurrentLinkedQueue<Entry>();
}
- private static class Entry
- {
- final String _key;
+ private static class Entry {
+ final String _key;
final ZNRecord _record;
- AtomicBoolean _sent = new AtomicBoolean(false);
+ AtomicBoolean _sent = new AtomicBoolean(false);
- Entry(String key, ZNRecord record)
- {
+ Entry(String key, ZNRecord record) {
_key = key;
_record = record;
}
}
- private final Queue[] _queues = new Queue[100];
-
+ private final Queue[] _queues = new Queue[100];
+
// potential memory leak if we add resource and remove resource
// TODO: move the cache logic to data accessor
-// private final Map<String, ZNRecord> _cache = new ConcurrentHashMap<String, ZNRecord>();
+ // private final Map<String, ZNRecord> _cache = new ConcurrentHashMap<String, ZNRecord>();
/**
* Set up a group committer and its associated queues
*/
- public GroupCommit()
- {
+ public GroupCommit() {
// Don't use Arrays.fill();
- for (int i = 0; i < _queues.length; ++i)
- {
+ for (int i = 0; i < _queues.length; ++i) {
_queues[i] = new Queue();
}
}
- private Queue getQueue(String key)
- {
+ private Queue getQueue(String key) {
return _queues[(key.hashCode() & Integer.MAX_VALUE) % _queues.length];
}
@@ -85,20 +79,17 @@ public class GroupCommit
* @param record the data to be merged in
* @return true if successful, false otherwise
*/
- public boolean commit(BaseDataAccessor<ZNRecord> accessor, int options, String key, ZNRecord record)
- {
+ public boolean commit(BaseDataAccessor<ZNRecord> accessor, int options, String key,
+ ZNRecord record) {
Queue queue = getQueue(key);
Entry entry = new Entry(key, record);
queue._pending.add(entry);
- while (!entry._sent.get())
- {
- if (queue._running.compareAndSet(null, Thread.currentThread()))
- {
+ while (!entry._sent.get()) {
+ if (queue._running.compareAndSet(null, Thread.currentThread())) {
ArrayList<Entry> processed = new ArrayList<Entry>();
- try
- {
+ try {
if (queue._pending.peek() == null)
return true;
@@ -107,51 +98,44 @@ public class GroupCommit
processed.add(first);
String mergedKey = first._key;
-// ZNRecord merged = _cache.get(mergedKey);
+ // ZNRecord merged = _cache.get(mergedKey);
ZNRecord merged = null;
-
- try
- {
+
+ try {
// accessor will fallback to zk if not found in cache
merged = accessor.get(mergedKey, null, options);
- }
- catch (ZkNoNodeException e)
- {
+ } catch (ZkNoNodeException e) {
// OK.
}
-
+
/**
- * If the local cache does not contain a value, need to check if there is a
+ * If the local cache does not contain a value, need to check if there is a
* value in ZK; use it as initial value if exists
*/
- if (merged == null)
- {
-// ZNRecord valueOnZk = null;
-// try
-// {
-// valueOnZk = accessor.get(mergedKey, null, 0);
-// }
-// catch(Exception e)
-// {
-// LOG.info(e);
-// }
-// if(valueOnZk != null)
-// {
-// merged = valueOnZk;
-// merged.merge(first._record);
-// }
-// else // Zk path has null data. use the first record as initial record.
+ if (merged == null) {
+ // ZNRecord valueOnZk = null;
+ // try
+ // {
+ // valueOnZk = accessor.get(mergedKey, null, 0);
+ // }
+ // catch(Exception e)
+ // {
+ // LOG.info(e);
+ // }
+ // if(valueOnZk != null)
+ // {
+ // merged = valueOnZk;
+ // merged.merge(first._record);
+ // }
+ // else // Zk path has null data. use the first record as initial record.
{
merged = new ZNRecord(first._record);
}
- }
- else
- {
+ } else {
merged.merge(first._record);
}
Iterator<Entry> it = queue._pending.iterator();
- while (it.hasNext())
- {
+ while (it.hasNext()) {
Entry ent = it.next();
if (!ent._key.equals(mergedKey))
continue;
@@ -164,30 +148,20 @@ public class GroupCommit
accessor.set(mergedKey, merged, options);
// accessor.set(mergedKey, merged, BaseDataAccessor.Option.PERSISTENT);
// _cache.put(mergedKey, merged);
- }
- finally
- {
+ } finally {
queue._running.set(null);
- for (Entry e : processed)
- {
- synchronized (e)
- {
+ for (Entry e : processed) {
+ synchronized (e) {
e._sent.set(true);
e.notify();
}
}
}
- }
- else
- {
- synchronized (entry)
- {
- try
- {
+ } else {
+ synchronized (entry) {
+ try {
entry.wait(10);
- }
- catch (InterruptedException e)
- {
+ } catch (InterruptedException e) {
e.printStackTrace();
return false;
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/HealthStateChangeListener.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HealthStateChangeListener.java b/helix-core/src/main/java/org/apache/helix/HealthStateChangeListener.java
index 607d5c9..9202330 100644
--- a/helix-core/src/main/java/org/apache/helix/HealthStateChangeListener.java
+++ b/helix-core/src/main/java/org/apache/helix/HealthStateChangeListener.java
@@ -26,17 +26,14 @@ import org.apache.helix.model.HealthStat;
/**
* Interface to implement to listen for when a health status event is triggered.
*/
-public interface HealthStateChangeListener
-{
+public interface HealthStateChangeListener {
/**
* Invoked when health stats change
- *
* @param instanceName the instance where the health status changed
* @param reports the health statuses
* @param changeContext event properties and state
*/
- public void onHealthChange(String instanceName,
- List<HealthStat> reports,
- NotificationContext changeContext);
+ public void onHealthChange(String instanceName, List<HealthStat> reports,
+ NotificationContext changeContext);
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
index 4791703..d5c62fa 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixAdmin.java
@@ -28,18 +28,15 @@ import org.apache.helix.model.ClusterConstraints.ConstraintType;
/*
* Helix cluster management
*/
-public interface HelixAdmin
-{
+public interface HelixAdmin {
/**
* Get a list of clusters under "/"
- *
* @return a list of cluster names
*/
List<String> getClusters();
/**
* Get a list of instances under a cluster
- *
* @param clusterName
* @return a list of instance names
*/
@@ -47,7 +44,6 @@ public interface HelixAdmin
/**
* Get an instance config
- *
* @param clusterName
* @param instanceName
* @return InstanceConfig corresponding to the specified instance
@@ -56,7 +52,6 @@ public interface HelixAdmin
/**
* Get a list of resources in a cluster
- *
* @param clusterName
* @return a list of resource names in the cluster
*/
@@ -64,14 +59,13 @@ public interface HelixAdmin
/**
* Add a cluster
- *
* @param clusterName
* @return true if successfully created, or if cluster already exists
- */
+ */
boolean addCluster(String clusterName);
+
/**
* Add a cluster
- *
* @param clusterName
* @param recreateIfExists If the cluster already exists, it will delete it and recreate
* @return true if successfully created, or if cluster already exists
@@ -80,7 +74,6 @@ public interface HelixAdmin
/**
* Add a cluster and also add this cluster as a resource group in the super cluster
- *
* @param clusterName
* @param grandCluster
*/
@@ -88,44 +81,33 @@ public interface HelixAdmin
/**
* Add a resource to a cluster, using the default ideal state mode AUTO
- *
* @param clusterName
* @param resourceName
* @param numPartitions
* @param stateModelRef
*/
- void addResource(String clusterName,
- String resourceName,
- int numPartitions,
- String stateModelRef);
+ void addResource(String clusterName, String resourceName, int numPartitions, String stateModelRef);
+
/**
- *
* @param clusterName
* @param resourceName
* @param idealstate
*/
- void addResource(String clusterName,
- String resourceName,
- IdealState idealstate);
+ void addResource(String clusterName, String resourceName, IdealState idealstate);
/**
* Add a resource to a cluster
- *
* @param clusterName
* @param resourceName
* @param numPartitions
* @param stateModelRef
* @param rebalancerMode
*/
- void addResource(String clusterName,
- String resourceName,
- int numPartitions,
- String stateModelRef,
- String rebalancerMode);
+ void addResource(String clusterName, String resourceName, int numPartitions,
+ String stateModelRef, String rebalancerMode);
/**
* Add a resource to a cluster, using a bucket size > 1
- *
* @param clusterName
* @param resourceName
* @param numPartitions
@@ -133,16 +115,11 @@ public interface HelixAdmin
* @param rebalancerMode
* @param bucketSize
*/
- void addResource(String clusterName,
- String resourceName,
- int numPartitions,
- String stateModelRef,
- String rebalancerMode,
- int bucketSize);
-
+ void addResource(String clusterName, String resourceName, int numPartitions,
+ String stateModelRef, String rebalancerMode, int bucketSize);
+
/**
* Add a resource to a cluster, using a bucket size > 1
- *
* @param clusterName
* @param resourceName
* @param numPartitions
@@ -151,17 +128,11 @@ public interface HelixAdmin
* @param bucketSize
* @param maxPartitionsPerInstance
*/
- void addResource(String clusterName,
- String resourceName,
- int numPartitions,
- String stateModelRef,
- String rebalancerMode,
- int bucketSize,
- int maxPartitionsPerInstance);
-
+ void addResource(String clusterName, String resourceName, int numPartitions,
+ String stateModelRef, String rebalancerMode, int bucketSize, int maxPartitionsPerInstance);
+
/**
* Add an instance to a cluster
- *
* @param clusterName
* @param instanceConfig
*/
@@ -169,7 +140,6 @@ public interface HelixAdmin
/**
* Drop an instance from a cluster
- *
* @param clusterName
* @param instanceConfig
*/
@@ -177,7 +147,6 @@ public interface HelixAdmin
/**
* Get ideal state for a resource
- *
* @param clusterName
* @param resourceName
* @return
@@ -186,18 +155,14 @@ public interface HelixAdmin
/**
* Set ideal state for a resource
- *
* @param clusterName
* @param resourceName
* @param idealState
*/
- void setResourceIdealState(String clusterName,
- String resourceName,
- IdealState idealState);
+ void setResourceIdealState(String clusterName, String resourceName, IdealState idealState);
/**
* Disable or enable an instance
- *
* @param clusterName
* @param instanceName
* @param enabled
@@ -206,22 +171,17 @@ public interface HelixAdmin
/**
* Disable or enable a list of partitions on an instance
- *
* @param enabled
* @param clusterName
* @param instanceName
* @param resourceName
* @param partitionNames
*/
- void enablePartition(boolean enabled,
- String clusterName,
- String instanceName,
- String resourceName,
- List<String> partitionNames);
+ void enablePartition(boolean enabled, String clusterName, String instanceName,
+ String resourceName, List<String> partitionNames);
/**
* Disable or enable a cluster
- *
* @param clusterName
* @param enabled
*/
@@ -229,23 +189,18 @@ public interface HelixAdmin
/**
* Reset a list of partitions in error state for an instance
- *
* The partitions are assume to be in error state and reset will bring them from error
* to initial state. An error to initial state transition is required for reset.
- *
* @param clusterName
* @param instanceName
* @param resourceName
* @param partitionNames
*/
- void resetPartition(String clusterName,
- String instanceName,
- String resourceName,
- List<String> partitionNames);
+ void resetPartition(String clusterName, String instanceName, String resourceName,
+ List<String> partitionNames);
/**
* Reset all the partitions in error state for a list of instances
- *
* @param clusterName
* @param instanceNames
*/
@@ -253,7 +208,6 @@ public interface HelixAdmin
/**
* Reset all partitions in error state for a list of resources
- *
* @param clusterName
* @param resourceNames
*/
@@ -261,18 +215,14 @@ public interface HelixAdmin
/**
* Add a state model definition
- *
* @param clusterName
* @param stateModelDef
* @param record
*/
- void addStateModelDef(String clusterName,
- String stateModelDef,
- StateModelDefinition record);
+ void addStateModelDef(String clusterName, String stateModelDef, StateModelDefinition record);
/**
* Drop a resource from a cluster
- *
* @param clusterName
* @param resourceName
*/
@@ -280,7 +230,6 @@ public interface HelixAdmin
/**
* Add a statistics to a cluster
- *
* @param clusterName
* @param statName
*/
@@ -288,7 +237,6 @@ public interface HelixAdmin
/**
* Add an alert to a cluster
- *
* @param clusterName
* @param alertName
*/
@@ -296,7 +244,6 @@ public interface HelixAdmin
/**
* Drop statistics from a cluster
- *
* @param clusterName
* @param statName
*/
@@ -304,7 +251,6 @@ public interface HelixAdmin
/**
* Drop an alert from a cluster
- *
* @param clusterName
* @param alertName
*/
@@ -312,7 +258,6 @@ public interface HelixAdmin
/**
* Get a list of state model definitions in a cluster
- *
* @param clusterName
* @return
*/
@@ -320,7 +265,6 @@ public interface HelixAdmin
/**
* Get a state model definition in a cluster
- *
* @param clusterName
* @param stateModelName
* @return StateModelDefinition identified by stateModelName
@@ -329,7 +273,6 @@ public interface HelixAdmin
/**
* Get external view for a resource
- *
* @param clusterName
* @param resourceName
* @return ExternalView for the resource
@@ -338,14 +281,12 @@ public interface HelixAdmin
/**
* Drop a cluster
- *
* @param clusterName
*/
void dropCluster(String clusterName);
/**
* Set configuration values
- *
* @param scope
* @param properties
*/
@@ -353,7 +294,6 @@ public interface HelixAdmin
/**
* Remove configuration values
- *
* @param scope
* @param keys
*/
@@ -361,7 +301,6 @@ public interface HelixAdmin
/**
* Get configuration values
- *
* @param scope
* @param keys
* @return configuration values ordered by the provided keys
@@ -370,7 +309,6 @@ public interface HelixAdmin
/**
* Get configuration keys
- *
* @param scope
* @return keys mapping to valid configuration values
*/
@@ -378,7 +316,6 @@ public interface HelixAdmin
/**
* Rebalance a resource in cluster
- *
* @param clusterName
* @param resourceName
* @param replica
@@ -387,113 +324,97 @@ public interface HelixAdmin
/**
* Add ideal state using a json format file
- *
* @param clusterName
* @param resourceName
* @param idealStateFile
* @throws IOException
*/
- void addIdealState(String clusterName, String resourceName, String idealStateFile) throws IOException;
+ void addIdealState(String clusterName, String resourceName, String idealStateFile)
+ throws IOException;
/**
* Add state model definition using a json format file
- *
* @param clusterName
* @param stateModelDefName
* @param stateModelDefFile
* @throws IOException error reading the state model definition file
*/
- void addStateModelDef(String clusterName,
- String stateModelDefName,
- String stateModelDefFile) throws IOException;
+ void addStateModelDef(String clusterName, String stateModelDefName, String stateModelDefFile)
+ throws IOException;
/**
* Add a constraint item; create if not exist
- *
* @param clusterName
* @param constraintType
* @param constraintId
* @param constraintItem
*/
- void setConstraint(String clusterName,
- ConstraintType constraintType,
- String constraintId,
- ConstraintItem constraintItem);
+ void setConstraint(String clusterName, ConstraintType constraintType, String constraintId,
+ ConstraintItem constraintItem);
/**
* Remove a constraint item
- *
* @param clusterName
* @param constraintType
* @param constraintId
*/
- void removeConstraint(String clusterName,
- ConstraintType constraintType,
- String constraintId);
-
+ void removeConstraint(String clusterName, ConstraintType constraintType, String constraintId);
+
/**
* Get all constraints for a type
- *
* @param clusterName
* @param constraintType
* @return constraints of constraintType
*/
- ClusterConstraints getConstraints(String clusterName,
- ConstraintType constraintType);
-
+ ClusterConstraints getConstraints(String clusterName, ConstraintType constraintType);
+
/**
- *
* @param clusterName
* @param currentIdealState
* @param instanceNames
*/
- void rebalance(String clusterName,
- IdealState currentIdealState,
- List<String> instanceNames);
+ void rebalance(String clusterName, IdealState currentIdealState, List<String> instanceNames);
+
/**
- *
* @param clusterName
* @param resourceName
* @param replica
* @param instances
*/
- void rebalance(String clusterName, String resourceName, int replica,
- List<String> instances);
+ void rebalance(String clusterName, String resourceName, int replica, List<String> instances);
+
/**
- *
* @param clusterName
* @param resourceName
* @param replica
* @param keyPrefix
* @param group the group identifier of instances to rebalance
*/
- void rebalance(String clusterName, String resourceName, int replica,
- String keyPrefix, String group);
+ void rebalance(String clusterName, String resourceName, int replica, String keyPrefix,
+ String group);
+
/**
- *
* @param clusterName
* @param tag
*/
List<String> getInstancesInClusterWithTag(String clusterName, String tag);
-
+
/**
- *
* @param clusterName
* @param instanceNames
* @param tag
*/
void addInstanceTag(String clusterName, String instanceName, String tag);
-
+
/**
- *
* @param clusterName
* @param instanceNames
* @param tag
*/
void removeInstanceTag(String clusterName, String instanceName, String tag);
-
+
/**
- * Release resources
+ * Release resources
*/
void close();
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/HelixConstants.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HelixConstants.java b/helix-core/src/main/java/org/apache/helix/HelixConstants.java
index f8fded8..6b6287c 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixConstants.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixConstants.java
@@ -22,11 +22,9 @@ package org.apache.helix;
/*
* Identifying constants of the components in a Helix-managed cluster
*/
-public interface HelixConstants
-{
+public interface HelixConstants {
// TODO: ChangeType and PropertyType are duplicated, consider unifying
- enum ChangeType
- {
+ enum ChangeType {
// @formatter:off
IDEAL_STATE,
CONFIG,
@@ -41,15 +39,13 @@ public interface HelixConstants
// @formatter:on
}
- enum StateModelToken
- {
+ enum StateModelToken {
ANY_LIVEINSTANCE
}
- enum ClusterConfigType
- {
+ enum ClusterConfigType {
HELIX_DISABLE_PIPELINE_TRIGGERS
}
-
+
static final String DEFAULT_STATE_MODEL_FACTORY = "DEFAULT";
}
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java b/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
index 5ad48ea..ad2c4bc 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixDataAccessor.java
@@ -28,14 +28,10 @@ import org.I0Itec.zkclient.DataUpdater;
* Interface used to interact with Helix Data Types like IdealState, Config,
* LiveInstance, Message, ExternalView etc PropertyKey represent the HelixData
* type. See {@link PropertyKey.Builder} to get more information on building a propertyKey.
- *
- *
*/
-public interface HelixDataAccessor
-{
+public interface HelixDataAccessor {
/**
* Create a helix property only if it does not exist.
- *
* @param key
* @param value
* @return true if creation was successful. False if already exists or if it
@@ -49,7 +45,6 @@ public interface HelixDataAccessor
* assumes the node exists and only tries to update it only if the call fail
* it will create the node. So there is a performance cost if always ends up
* creating the node.
- *
* @param key
* @param value
* @true if the operation was successful
@@ -58,7 +53,6 @@ public interface HelixDataAccessor
/**
* Updates a property using newvalue.merge(oldvalue)
- *
* @param key
* @param value
* @return true if the update was successful
@@ -68,7 +62,6 @@ public interface HelixDataAccessor
/**
* Return the property value, it must be refer to a single Helix Property. i.e
* PropertyKey.isLeaf() must return true.
- *
* @param key
* @return value, Null if absent or on error
*/
@@ -77,7 +70,6 @@ public interface HelixDataAccessor
/**
* Return a list of property values, each of which must be refer to a single Helix
* Property. Property may be bucketized.
- *
* @param keys
* @return
*/
@@ -85,7 +77,6 @@ public interface HelixDataAccessor
/**
* Removes the property
- *
* @param key
* @return true if removal was successful or node does not exist. false if the
* node existed and failed to remove it
@@ -95,7 +86,6 @@ public interface HelixDataAccessor
/**
* Return the child names for a property. PropertyKey needs to refer to a
* collection like instances, resources. PropertyKey.isLeaf must be false
- *
* @param key
* @return SubPropertyNames
*/
@@ -104,7 +94,6 @@ public interface HelixDataAccessor
/**
* Get the child values for a property. PropertyKey needs to refer to just one
* level above the non leaf. PropertyKey.isCollection must be true.
- *
* @param key
* @return subPropertyValues
*/
@@ -113,7 +102,6 @@ public interface HelixDataAccessor
/**
* Same as getChildValues except that it converts list into a map using the id
* of the HelixProperty
- *
* @param key
* @return a map of property identifiers to typed properties
*/
@@ -122,45 +110,38 @@ public interface HelixDataAccessor
/**
* Adds multiple children to a parent.
- *
* @param keys
* @param children
* @return array where true means the child was added and false means it was not
*/
- <T extends HelixProperty> boolean[] createChildren(List<PropertyKey> keys,
- List<T> children);
+ <T extends HelixProperty> boolean[] createChildren(List<PropertyKey> keys, List<T> children);
/**
* Sets multiple children under one parent
- *
* @param keys
* @param children
* @return array where true means the child was set and false means it was not
*/
<T extends HelixProperty> boolean[] setChildren(List<PropertyKey> keys, List<T> children);
-
+
/**
* Updates multiple children under one parent
* TODO: change to use property-keys instead of paths
- *
* @param paths
* @param updaters
* @return array where true means the child was updated and false means it was not
*/
<T extends HelixProperty> boolean[] updateChildren(List<String> paths,
- List<DataUpdater<ZNRecord>> updaters,
- int options);
-
+ List<DataUpdater<ZNRecord>> updaters, int options);
+
/**
* Get key builder for the accessor
- *
* @return instantiated PropertyKey.Builder
*/
PropertyKey.Builder keyBuilder();
/**
* Get underlying base data accessor
- *
* @return a data accessor that can process ZNRecord objects
*/
BaseDataAccessor<ZNRecord> getBaseDataAccessor();
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/HelixDefinedState.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HelixDefinedState.java b/helix-core/src/main/java/org/apache/helix/HelixDefinedState.java
index 88cf068..d822fa2 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixDefinedState.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixDefinedState.java
@@ -19,24 +19,22 @@ package org.apache.helix;
* under the License.
*/
-
/**
-* helix defined states
-*
-* ERROR : when errors happen during state transitions, transit to ERROR state
-* participant will also invoke state-model.on-err(), ignore errors in state-model.on-err()
-* when drop resource in ERROR state and not disabled, controller sends ERROR->DROPPED transition
-* if errors happen in ERROR->DROPPED transition, participant will disable resource/partition
-* when disable resource/partition in ERROR state, resource/partition will be marked disabled
-* but controller not send any transitions
-* when reset resource/partition in ERROR state and not disabled
-* controller send ERROR->initial-state transition
-* if errors happen in ERROR->initial-state transition, remain in ERROR state
-* DROPPED : when drop resource in a non-ERROR state and not disabled
-* controller sends all the transitions from current-state to initial-state
-* then sends initial-state->DROPPED transition
-* @see HELIX-43: add support for dropping partitions in error state
-*/
+ * helix defined states
+ * ERROR : when errors happen during state transitions, transit to ERROR state
+ * participant will also invoke state-model.on-err(), ignore errors in state-model.on-err()
+ * when drop resource in ERROR state and not disabled, controller sends ERROR->DROPPED transition
+ * if errors happen in ERROR->DROPPED transition, participant will disable resource/partition
+ * when disable resource/partition in ERROR state, resource/partition will be marked disabled
+ * but controller not send any transitions
+ * when reset resource/partition in ERROR state and not disabled
+ * controller send ERROR->initial-state transition
+ * if errors happen in ERROR->initial-state transition, remain in ERROR state
+ * DROPPED : when drop resource in a non-ERROR state and not disabled
+ * controller sends all the transitions from current-state to initial-state
+ * then sends initial-state->DROPPED transition
+ * @see HELIX-43: add support for dropping partitions in error state
+ */
public enum HelixDefinedState {
ERROR,
DROPPED
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/f414aad4/helix-core/src/main/java/org/apache/helix/HelixException.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/HelixException.java b/helix-core/src/main/java/org/apache/helix/HelixException.java
index 0a64194..8693026 100644
--- a/helix-core/src/main/java/org/apache/helix/HelixException.java
+++ b/helix-core/src/main/java/org/apache/helix/HelixException.java
@@ -22,18 +22,15 @@ package org.apache.helix;
/**
* Base class for an exception thrown by Helix due to inconsistencies caught by Helix itself.
*/
-public class HelixException extends RuntimeException
-{
+public class HelixException extends RuntimeException {
private static final long serialVersionUID = 6558251214364526257L;
- public HelixException(String message)
- {
+ public HelixException(String message) {
super(message);
}
- public HelixException(Throwable cause)
- {
+ public HelixException(Throwable cause) {
super(cause);
}
}