@Test public void testAddNode() throws Throwable { WaitableConnector connector = new WaitableConnector(); NodeImpl node = (NodeImpl) newNodeBuilder( NodeId.of("A"), new NodeEndpoint("A", "localhost", 2333), new NodeEndpoint("B", "localhost", 2334), new NodeEndpoint("C", "localhost", 2335)) .setConnector(connector) .setTaskExecutor(taskExecutor) .setGroupConfigChangeTaskExecutor(groupConfigChangeTaskExecutor) .build(); node.start(); node.electionTimeout(); node.processRequestVoteResult(new RequestVoteResult(1, true)).get(); Future<GroupConfigChangeTaskReference> future = cachedThreadTaskExecutor.submit(() -> node.addNode(new NodeEndpoint("D", "localhost", 2336))); connector.awaitAppendEntriesRpc(); node.onReceiveAppendEntriesResult(new AppendEntriesResultMessage( new AppendEntriesResult("", 1, true), NodeId.of("D"), createAppendEntriesRpc(1))); GroupConfigChangeTaskReference reference = future.get(); connector.awaitAppendEntriesRpc(); node.onReceiveAppendEntriesResult(new AppendEntriesResultMessage( new AppendEntriesResult("", 1, true), NodeId.of("B"), createAppendEntriesRpc(2))); node.onReceiveAppendEntriesResult(new AppendEntriesResultMessage( new AppendEntriesResult("", 1, true), NodeId.of("C"), createAppendEntriesRpc(2))); Assert.assertEquals(GroupConfigChangeTaskResult.OK, reference.getResult(1000L)); checkWithinTaskExecutor(node, () -> Assert.assertEquals(4, node.getContext().group().getCountOfMajor())); }
@Override @Nonnull public GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint) { Preconditions.checkNotNull(endpoint); ensureLeader(); if (context.selfId().equals(endpoint.getId())) { throw new IllegalArgumentException("new node cannot be self"); } NewNodeCatchUpTask newNodeCatchUpTask = new NewNodeCatchUpTask(newNodeCatchUpTaskContext, endpoint, context.config()); if (!newNodeCatchUpTaskGroup.add(newNodeCatchUpTask)) { throw new IllegalArgumentException("node " + endpoint.getId() + " is adding"); } NewNodeCatchUpTaskResult newNodeCatchUpTaskResult; try { newNodeCatchUpTaskResult = newNodeCatchUpTask.call(); switch (newNodeCatchUpTaskResult.getState()) { case REPLICATION_FAILED: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.REPLICATION_FAILED); case TIMEOUT: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.TIMEOUT); } } catch (Exception e) { if (!(e instanceof InterruptedException)) { logger.warn("failed to catch up new node " + endpoint.getId(), e); } return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.ERROR); } GroupConfigChangeTaskResult result = awaitPreviousGroupConfigChangeTask(); if (result != null) { return new FixedResultGroupConfigTaskReference(result); } synchronized (this) { if (!groupConfigChangeTaskHolder.isEmpty()) { throw new IllegalStateException("group config change concurrently"); } AddNodeTask addNodeTask = new AddNodeTask(groupConfigChangeTaskContext, endpoint, newNodeCatchUpTaskResult); Future<GroupConfigChangeTaskResult> future = context.groupConfigChangeTaskExecutor().submit(addNodeTask); GroupConfigChangeTaskReference reference = new FutureGroupConfigChangeTaskReference(future); groupConfigChangeTaskHolder = new GroupConfigChangeTaskHolder(addNodeTask, reference); return reference; } }
NodeImpl implements Node { @Override @Nonnull public GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint) { Preconditions.checkNotNull(endpoint); ensureLeader(); if (context.selfId().equals(endpoint.getId())) { throw new IllegalArgumentException("new node cannot be self"); } NewNodeCatchUpTask newNodeCatchUpTask = new NewNodeCatchUpTask(newNodeCatchUpTaskContext, endpoint, context.config()); if (!newNodeCatchUpTaskGroup.add(newNodeCatchUpTask)) { throw new IllegalArgumentException("node " + endpoint.getId() + " is adding"); } NewNodeCatchUpTaskResult newNodeCatchUpTaskResult; try { newNodeCatchUpTaskResult = newNodeCatchUpTask.call(); switch (newNodeCatchUpTaskResult.getState()) { case REPLICATION_FAILED: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.REPLICATION_FAILED); case TIMEOUT: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.TIMEOUT); } } catch (Exception e) { if (!(e instanceof InterruptedException)) { logger.warn("failed to catch up new node " + endpoint.getId(), e); } return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.ERROR); } GroupConfigChangeTaskResult result = awaitPreviousGroupConfigChangeTask(); if (result != null) { return new FixedResultGroupConfigTaskReference(result); } synchronized (this) { if (!groupConfigChangeTaskHolder.isEmpty()) { throw new IllegalStateException("group config change concurrently"); } AddNodeTask addNodeTask = new AddNodeTask(groupConfigChangeTaskContext, endpoint, newNodeCatchUpTaskResult); Future<GroupConfigChangeTaskResult> future = context.groupConfigChangeTaskExecutor().submit(addNodeTask); GroupConfigChangeTaskReference reference = new FutureGroupConfigChangeTaskReference(future); groupConfigChangeTaskHolder = new GroupConfigChangeTaskHolder(addNodeTask, reference); return reference; } } }
NodeImpl implements Node { @Override @Nonnull public GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint) { Preconditions.checkNotNull(endpoint); ensureLeader(); if (context.selfId().equals(endpoint.getId())) { throw new IllegalArgumentException("new node cannot be self"); } NewNodeCatchUpTask newNodeCatchUpTask = new NewNodeCatchUpTask(newNodeCatchUpTaskContext, endpoint, context.config()); if (!newNodeCatchUpTaskGroup.add(newNodeCatchUpTask)) { throw new IllegalArgumentException("node " + endpoint.getId() + " is adding"); } NewNodeCatchUpTaskResult newNodeCatchUpTaskResult; try { newNodeCatchUpTaskResult = newNodeCatchUpTask.call(); switch (newNodeCatchUpTaskResult.getState()) { case REPLICATION_FAILED: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.REPLICATION_FAILED); case TIMEOUT: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.TIMEOUT); } } catch (Exception e) { if (!(e instanceof InterruptedException)) { logger.warn("failed to catch up new node " + endpoint.getId(), e); } return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.ERROR); } GroupConfigChangeTaskResult result = awaitPreviousGroupConfigChangeTask(); if (result != null) { return new FixedResultGroupConfigTaskReference(result); } synchronized (this) { if (!groupConfigChangeTaskHolder.isEmpty()) { throw new IllegalStateException("group config change concurrently"); } AddNodeTask addNodeTask = new AddNodeTask(groupConfigChangeTaskContext, endpoint, newNodeCatchUpTaskResult); Future<GroupConfigChangeTaskResult> future = context.groupConfigChangeTaskExecutor().submit(addNodeTask); GroupConfigChangeTaskReference reference = new FutureGroupConfigChangeTaskReference(future); groupConfigChangeTaskHolder = new GroupConfigChangeTaskHolder(addNodeTask, reference); return reference; } } NodeImpl(NodeContext context); }
NodeImpl implements Node { @Override @Nonnull public GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint) { Preconditions.checkNotNull(endpoint); ensureLeader(); if (context.selfId().equals(endpoint.getId())) { throw new IllegalArgumentException("new node cannot be self"); } NewNodeCatchUpTask newNodeCatchUpTask = new NewNodeCatchUpTask(newNodeCatchUpTaskContext, endpoint, context.config()); if (!newNodeCatchUpTaskGroup.add(newNodeCatchUpTask)) { throw new IllegalArgumentException("node " + endpoint.getId() + " is adding"); } NewNodeCatchUpTaskResult newNodeCatchUpTaskResult; try { newNodeCatchUpTaskResult = newNodeCatchUpTask.call(); switch (newNodeCatchUpTaskResult.getState()) { case REPLICATION_FAILED: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.REPLICATION_FAILED); case TIMEOUT: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.TIMEOUT); } } catch (Exception e) { if (!(e instanceof InterruptedException)) { logger.warn("failed to catch up new node " + endpoint.getId(), e); } return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.ERROR); } GroupConfigChangeTaskResult result = awaitPreviousGroupConfigChangeTask(); if (result != null) { return new FixedResultGroupConfigTaskReference(result); } synchronized (this) { if (!groupConfigChangeTaskHolder.isEmpty()) { throw new IllegalStateException("group config change concurrently"); } AddNodeTask addNodeTask = new AddNodeTask(groupConfigChangeTaskContext, endpoint, newNodeCatchUpTaskResult); Future<GroupConfigChangeTaskResult> future = context.groupConfigChangeTaskExecutor().submit(addNodeTask); GroupConfigChangeTaskReference reference = new FutureGroupConfigChangeTaskReference(future); groupConfigChangeTaskHolder = new GroupConfigChangeTaskHolder(addNodeTask, reference); return reference; } } NodeImpl(NodeContext context); @Override synchronized void registerStateMachine(@Nonnull StateMachine stateMachine); @Override @Nonnull RoleNameAndLeaderId getRoleNameAndLeaderId(); @Override void addNodeRoleListener(@Nonnull NodeRoleListener listener); @Override synchronized void start(); @Override void appendLog(@Nonnull byte[] commandBytes); @Override @Nonnull GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint); @Override @Nonnull GroupConfigChangeTaskReference removeNode(@Nonnull NodeId id); @Subscribe void onReceiveRequestVoteRpc(RequestVoteRpcMessage rpcMessage); @Subscribe void onReceiveRequestVoteResult(RequestVoteResult result); @Subscribe void onReceiveAppendEntriesRpc(AppendEntriesRpcMessage rpcMessage); @Subscribe void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage); @Subscribe void onReceiveInstallSnapshotRpc(InstallSnapshotRpcMessage rpcMessage); @Subscribe void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage); @Subscribe void onGroupConfigEntryFromLeaderAppend(GroupConfigEntryFromLeaderAppendEvent event); @Subscribe void onGroupConfigEntryCommitted(GroupConfigEntryCommittedEvent event); @Subscribe void onGroupConfigEntryBatchRemoved(GroupConfigEntryBatchRemovedEvent event); @Subscribe void onGenerateSnapshot(SnapshotGenerateEvent event); @Subscribe void onReceiveDeadEvent(DeadEvent deadEvent); @Override synchronized void stop(); }
NodeImpl implements Node { @Override @Nonnull public GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint) { Preconditions.checkNotNull(endpoint); ensureLeader(); if (context.selfId().equals(endpoint.getId())) { throw new IllegalArgumentException("new node cannot be self"); } NewNodeCatchUpTask newNodeCatchUpTask = new NewNodeCatchUpTask(newNodeCatchUpTaskContext, endpoint, context.config()); if (!newNodeCatchUpTaskGroup.add(newNodeCatchUpTask)) { throw new IllegalArgumentException("node " + endpoint.getId() + " is adding"); } NewNodeCatchUpTaskResult newNodeCatchUpTaskResult; try { newNodeCatchUpTaskResult = newNodeCatchUpTask.call(); switch (newNodeCatchUpTaskResult.getState()) { case REPLICATION_FAILED: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.REPLICATION_FAILED); case TIMEOUT: return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.TIMEOUT); } } catch (Exception e) { if (!(e instanceof InterruptedException)) { logger.warn("failed to catch up new node " + endpoint.getId(), e); } return new FixedResultGroupConfigTaskReference(GroupConfigChangeTaskResult.ERROR); } GroupConfigChangeTaskResult result = awaitPreviousGroupConfigChangeTask(); if (result != null) { return new FixedResultGroupConfigTaskReference(result); } synchronized (this) { if (!groupConfigChangeTaskHolder.isEmpty()) { throw new IllegalStateException("group config change concurrently"); } AddNodeTask addNodeTask = new AddNodeTask(groupConfigChangeTaskContext, endpoint, newNodeCatchUpTaskResult); Future<GroupConfigChangeTaskResult> future = context.groupConfigChangeTaskExecutor().submit(addNodeTask); GroupConfigChangeTaskReference reference = new FutureGroupConfigChangeTaskReference(future); groupConfigChangeTaskHolder = new GroupConfigChangeTaskHolder(addNodeTask, reference); return reference; } } NodeImpl(NodeContext context); @Override synchronized void registerStateMachine(@Nonnull StateMachine stateMachine); @Override @Nonnull RoleNameAndLeaderId getRoleNameAndLeaderId(); @Override void addNodeRoleListener(@Nonnull NodeRoleListener listener); @Override synchronized void start(); @Override void appendLog(@Nonnull byte[] commandBytes); @Override @Nonnull GroupConfigChangeTaskReference addNode(@Nonnull NodeEndpoint endpoint); @Override @Nonnull GroupConfigChangeTaskReference removeNode(@Nonnull NodeId id); @Subscribe void onReceiveRequestVoteRpc(RequestVoteRpcMessage rpcMessage); @Subscribe void onReceiveRequestVoteResult(RequestVoteResult result); @Subscribe void onReceiveAppendEntriesRpc(AppendEntriesRpcMessage rpcMessage); @Subscribe void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage); @Subscribe void onReceiveInstallSnapshotRpc(InstallSnapshotRpcMessage rpcMessage); @Subscribe void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage); @Subscribe void onGroupConfigEntryFromLeaderAppend(GroupConfigEntryFromLeaderAppendEvent event); @Subscribe void onGroupConfigEntryCommitted(GroupConfigEntryCommittedEvent event); @Subscribe void onGroupConfigEntryBatchRemoved(GroupConfigEntryBatchRemovedEvent event); @Subscribe void onGenerateSnapshot(SnapshotGenerateEvent event); @Subscribe void onReceiveDeadEvent(DeadEvent deadEvent); @Override synchronized void stop();  }