@Test public void testHelixParticipant() throws Exception { ClusterMapConfig clusterMapConfig = new ClusterMapConfig(new VerifiableProperties(props)); HelixParticipant participant = new HelixParticipant(clusterMapConfig, helixManagerFactory, new MetricRegistry(), getDefaultZkConnectStr(clusterMapConfig), true); assertTrue(helixManagerFactory.getHelixManager(InstanceType.SPECTATOR).isConnected()); assertFalse(helixManagerFactory.getHelixManager(InstanceType.PARTICIPANT).isConnected()); participant.participate(Collections.emptyList(), null); MockHelixManagerFactory.MockHelixManager helixManager = helixManagerFactory.getHelixManager(InstanceType.PARTICIPANT); assertTrue(helixManager.isConnected()); assertEquals(stateModelDef, helixManager.getStateModelDef()); assertEquals(AmbryStateModelFactory.class, helixManager.getStateModelFactory().getClass()); participant.close(); assertFalse(helixManager.isConnected()); }
public HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry, String zkConnectStr, boolean isSoleParticipant) { this.clusterMapConfig = clusterMapConfig; this.zkConnectStr = zkConnectStr; participantMetrics = new HelixParticipantMetrics(metricRegistry, isSoleParticipant ? null : zkConnectStr, localPartitionAndState); clusterName = clusterMapConfig.clusterMapClusterName; instanceName = getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); if (clusterName.isEmpty()) { throw new IllegalStateException("Cluster name is empty in clusterMapConfig"); } manager = helixFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, zkConnectStr); replicaSyncUpManager = new AmbryReplicaSyncUpManager(clusterMapConfig); partitionStateChangeListeners = new HashMap<>(); try { HelixManager spectatorManager = helixFactory.getZkHelixManagerAndConnect(clusterName, instanceName, InstanceType.SPECTATOR, zkConnectStr); helixAdmin = spectatorManager.getClusterManagmentTool(); dataNodeConfigSource = helixFactory.getDataNodeConfigSource(clusterMapConfig, zkConnectStr, new DataNodeConfigSourceMetrics(metricRegistry)); } catch (Exception exception) { throw new IllegalStateException("Error setting up administration facilities", exception); } }
HelixParticipant implements ClusterParticipant, PartitionStateChangeListener { public HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry, String zkConnectStr, boolean isSoleParticipant) { this.clusterMapConfig = clusterMapConfig; this.zkConnectStr = zkConnectStr; participantMetrics = new HelixParticipantMetrics(metricRegistry, isSoleParticipant ? null : zkConnectStr, localPartitionAndState); clusterName = clusterMapConfig.clusterMapClusterName; instanceName = getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); if (clusterName.isEmpty()) { throw new IllegalStateException("Cluster name is empty in clusterMapConfig"); } manager = helixFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, zkConnectStr); replicaSyncUpManager = new AmbryReplicaSyncUpManager(clusterMapConfig); partitionStateChangeListeners = new HashMap<>(); try { HelixManager spectatorManager = helixFactory.getZkHelixManagerAndConnect(clusterName, instanceName, InstanceType.SPECTATOR, zkConnectStr); helixAdmin = spectatorManager.getClusterManagmentTool(); dataNodeConfigSource = helixFactory.getDataNodeConfigSource(clusterMapConfig, zkConnectStr, new DataNodeConfigSourceMetrics(metricRegistry)); } catch (Exception exception) { throw new IllegalStateException("Error setting up administration facilities", exception); } } }
HelixParticipant implements ClusterParticipant, PartitionStateChangeListener { public HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry, String zkConnectStr, boolean isSoleParticipant) { this.clusterMapConfig = clusterMapConfig; this.zkConnectStr = zkConnectStr; participantMetrics = new HelixParticipantMetrics(metricRegistry, isSoleParticipant ? null : zkConnectStr, localPartitionAndState); clusterName = clusterMapConfig.clusterMapClusterName; instanceName = getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); if (clusterName.isEmpty()) { throw new IllegalStateException("Cluster name is empty in clusterMapConfig"); } manager = helixFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, zkConnectStr); replicaSyncUpManager = new AmbryReplicaSyncUpManager(clusterMapConfig); partitionStateChangeListeners = new HashMap<>(); try { HelixManager spectatorManager = helixFactory.getZkHelixManagerAndConnect(clusterName, instanceName, InstanceType.SPECTATOR, zkConnectStr); helixAdmin = spectatorManager.getClusterManagmentTool(); dataNodeConfigSource = helixFactory.getDataNodeConfigSource(clusterMapConfig, zkConnectStr, new DataNodeConfigSourceMetrics(metricRegistry)); } catch (Exception exception) { throw new IllegalStateException("Error setting up administration facilities", exception); } } HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry,
      String zkConnectStr, boolean isSoleParticipant); }
HelixParticipant implements ClusterParticipant, PartitionStateChangeListener { public HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry, String zkConnectStr, boolean isSoleParticipant) { this.clusterMapConfig = clusterMapConfig; this.zkConnectStr = zkConnectStr; participantMetrics = new HelixParticipantMetrics(metricRegistry, isSoleParticipant ? null : zkConnectStr, localPartitionAndState); clusterName = clusterMapConfig.clusterMapClusterName; instanceName = getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); if (clusterName.isEmpty()) { throw new IllegalStateException("Cluster name is empty in clusterMapConfig"); } manager = helixFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, zkConnectStr); replicaSyncUpManager = new AmbryReplicaSyncUpManager(clusterMapConfig); partitionStateChangeListeners = new HashMap<>(); try { HelixManager spectatorManager = helixFactory.getZkHelixManagerAndConnect(clusterName, instanceName, InstanceType.SPECTATOR, zkConnectStr); helixAdmin = spectatorManager.getClusterManagmentTool(); dataNodeConfigSource = helixFactory.getDataNodeConfigSource(clusterMapConfig, zkConnectStr, new DataNodeConfigSourceMetrics(metricRegistry)); } catch (Exception exception) { throw new IllegalStateException("Error setting up administration facilities", exception); } } HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry,
      String zkConnectStr, boolean isSoleParticipant); @Override void setInitialLocalPartitions(Collection<String> localPartitions); @Override void participate(List<AmbryHealthReport> ambryHealthReports, Callback<StatsSnapshot> callback); @Override void registerPartitionStateChangeListener(StateModelListenerType listenerType,
      PartitionStateChangeListener partitionStateChangeListener); @Override boolean setReplicaSealedState(ReplicaId replicaId, boolean isSealed); @Override boolean setReplicaStoppedState(List<ReplicaId> replicaIds, boolean markStop); @Override void close(); @Override List<String> getSealedReplicas(); @Override List<String> getStoppedReplicas(); @Override List<String> getDisabledReplicas(); @Override ReplicaSyncUpManager getReplicaSyncUpManager(); @Override boolean updateDataNodeInfoInCluster(ReplicaId replicaId, boolean shouldExist); @Override Map<StateModelListenerType, PartitionStateChangeListener> getPartitionStateChangeListeners(); @Override void setReplicaDisabledState(ReplicaId replicaId, boolean disable); @Override boolean resetPartitionState(String partitionName); @Override boolean supportsStateChanges(); HelixAdmin getHelixAdmin(); @Override void onPartitionBecomeBootstrapFromOffline(String partitionName); @Override void onPartitionBecomeStandbyFromBootstrap(String partitionName); @Override void onPartitionBecomeLeaderFromStandby(String partitionName); @Override void onPartitionBecomeStandbyFromLeader(String partitionName); @Override void onPartitionBecomeInactiveFromStandby(String partitionName); @Override void onPartitionBecomeOfflineFromInactive(String partitionName); @Override void onPartitionBecomeDroppedFromOffline(String partitionName); @Override void onPartitionBecomeDroppedFromError(String partitionName); @Override void onPartitionBecomeOfflineFromError(String partitionName); @Override void onReset(String partitionName); }
HelixParticipant implements ClusterParticipant, PartitionStateChangeListener { public HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry, String zkConnectStr, boolean isSoleParticipant) { this.clusterMapConfig = clusterMapConfig; this.zkConnectStr = zkConnectStr; participantMetrics = new HelixParticipantMetrics(metricRegistry, isSoleParticipant ? null : zkConnectStr, localPartitionAndState); clusterName = clusterMapConfig.clusterMapClusterName; instanceName = getInstanceName(clusterMapConfig.clusterMapHostName, clusterMapConfig.clusterMapPort); if (clusterName.isEmpty()) { throw new IllegalStateException("Cluster name is empty in clusterMapConfig"); } manager = helixFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT, zkConnectStr); replicaSyncUpManager = new AmbryReplicaSyncUpManager(clusterMapConfig); partitionStateChangeListeners = new HashMap<>(); try { HelixManager spectatorManager = helixFactory.getZkHelixManagerAndConnect(clusterName, instanceName, InstanceType.SPECTATOR, zkConnectStr); helixAdmin = spectatorManager.getClusterManagmentTool(); dataNodeConfigSource = helixFactory.getDataNodeConfigSource(clusterMapConfig, zkConnectStr, new DataNodeConfigSourceMetrics(metricRegistry)); } catch (Exception exception) { throw new IllegalStateException("Error setting up administration facilities", exception); } } HelixParticipant(ClusterMapConfig clusterMapConfig, HelixFactory helixFactory, MetricRegistry metricRegistry,
      String zkConnectStr, boolean isSoleParticipant); @Override void setInitialLocalPartitions(Collection<String> localPartitions); @Override void participate(List<AmbryHealthReport> ambryHealthReports, Callback<StatsSnapshot> callback); @Override void registerPartitionStateChangeListener(StateModelListenerType listenerType,
      PartitionStateChangeListener partitionStateChangeListener); @Override boolean setReplicaSealedState(ReplicaId replicaId, boolean isSealed); @Override boolean setReplicaStoppedState(List<ReplicaId> replicaIds, boolean markStop); @Override void close(); @Override List<String> getSealedReplicas(); @Override List<String> getStoppedReplicas(); @Override List<String> getDisabledReplicas(); @Override ReplicaSyncUpManager getReplicaSyncUpManager(); @Override boolean updateDataNodeInfoInCluster(ReplicaId replicaId, boolean shouldExist); @Override Map<StateModelListenerType, PartitionStateChangeListener> getPartitionStateChangeListeners(); @Override void setReplicaDisabledState(ReplicaId replicaId, boolean disable); @Override boolean resetPartitionState(String partitionName); @Override boolean supportsStateChanges(); HelixAdmin getHelixAdmin(); @Override void onPartitionBecomeBootstrapFromOffline(String partitionName); @Override void onPartitionBecomeStandbyFromBootstrap(String partitionName); @Override void onPartitionBecomeLeaderFromStandby(String partitionName); @Override void onPartitionBecomeStandbyFromLeader(String partitionName); @Override void onPartitionBecomeInactiveFromStandby(String partitionName); @Override void onPartitionBecomeOfflineFromInactive(String partitionName); @Override void onPartitionBecomeDroppedFromOffline(String partitionName); @Override void onPartitionBecomeDroppedFromError(String partitionName); @Override void onPartitionBecomeOfflineFromError(String partitionName); @Override void onReset(String partitionName);  }