InteractionProtocol { public MasterQueue publishMaster(final Master master) { String masterName = master.getMasterName(); String zkMasterPath = _zkConf.getZkPath(PathDef.MASTER); cleanupOldMasterData(masterName, zkMasterPath); boolean isMaster; try { createEphemeral(master, zkMasterPath, new MasterMetaData(masterName, System.currentTimeMillis())); isMaster = true; LOG.info(masterName + " started as master"); } catch (ZkNodeExistsException e) { registerDataListener(master, new IZkDataListener() { @Override public void handleDataDeleted(final String dataPath) throws KattaException { master.handleMasterDisappearedEvent(); } @Override public void handleDataChange(String dataPath, Object data) throws Exception { } }, zkMasterPath); isMaster = false; LOG.info(masterName + " started as secondary master"); } MasterQueue queue = null; if (isMaster) { LOG.info("master '" + master.getMasterName() + "' published"); String queuePath = _zkConf.getZkPath(PathDef.MASTER_QUEUE); if (!_zkClient.exists(queuePath)) { _zkClient.createPersistent(queuePath); } queue = new MasterQueue(_zkClient, queuePath); } else { LOG.info("secondary master '" + master.getMasterName() + "' registered"); } return queue; } InteractionProtocol(ZkClient zkClient, ZkConfiguration zkConfiguration); void registerComponent(ConnectedComponent connectedComponent); void unregisterComponent(ConnectedComponent component); void disconnect(); int getRegisteredComponentCount(); int getRegisteredListenerCount(); List<String> registerChildListener(ConnectedComponent component, PathDef pathDef, IAddRemoveListener listener); List<String> registerChildListener(ConnectedComponent component, PathDef pathDef, String childName,
          IAddRemoveListener listener); void unregisterChildListener(ConnectedComponent component, PathDef pathDef); void unregisterChildListener(ConnectedComponent component, PathDef pathDef, String childName); void registerDataListener(ConnectedComponent component, PathDef pathDef, String childName,
          IZkDataListener listener); void unregisterDataChanges(ConnectedComponent component, PathDef pathDef, String childName); void unregisterDataChanges(ConnectedComponent component, String dataPath); List<String> getKnownNodes(); List<String> getLiveNodes(); List<String> getIndices(); MasterMetaData getMasterMD(); NodeMetaData getNodeMD(String node); Collection<String> getNodeShards(String nodeName); int getShardReplication(String shard); long getShardAnnounceTime(String node, String shard); Map<String, List<String>> getShard2NodesMap(Collection<String> shardNames); List<String> getShardNodes(String shard); List<String> getShard2NodeShards(); ReplicationReport getReplicationReport(IndexMetaData indexMD); MasterQueue publishMaster(final Master master); NodeQueue publishNode(Node node, NodeMetaData nodeMetaData); void publishIndex(IndexMetaData indexMD); void unpublishIndex(String indexName); IndexMetaData getIndexMD(String index); void updateIndexMD(IndexMetaData indexMD); void publishShard(Node node, String shardName); void unpublishShard(Node node, String shard); void setMetric(String nodeName, MetricsRecord metricsRecord); MetricsRecord getMetric(String nodeName); void addMasterOperation(MasterOperation operation); OperationId addNodeOperation(String nodeName, NodeOperation nodeOperation); OperationResult getNodeOperationResult(OperationId operationId, boolean remove); boolean isNodeOperationQueued(OperationId operationId); void registerNodeOperationListener(ConnectedComponent component, OperationId operationId,
          IZkDataListener dataListener); boolean indexExists(String indexName); void showStructure(final boolean all); void explainStructure(); void setVersion(Version version); Version getVersion(); void setFlag(String name); boolean flagExists(String name); void removeFlag(String name); ZkConfiguration getZkConfiguration(); ZkClient getZkClient();  }
@Test(timeout = 7000) public void testPublishMaster() throws Exception { Master master1 = mock(Master.class); Master master2 = mock(Master.class); when(master1.getMasterName()).thenReturn("master1"); when(master2.getMasterName()).thenReturn("master2"); MasterOperation operation = mock(MasterOperation.class); _protocol.addMasterOperation(operation); MasterQueue queue = _protocol.publishMaster(master1); assertNotNull(queue); assertNotNull(_protocol.getMasterMD()); assertEquals(1, queue.size()); assertNotNull(queue.peek()); queue = _protocol.publishMaster(master1); assertNotNull(queue); assertNotNull(_protocol.getMasterMD()); queue = _protocol.publishMaster(master2); assertNull(queue); assertNotNull(_protocol.getMasterMD()); }