Zen2: Cluster state publication pipeline #32584
Conversation
ywelsch
added
>enhancement
v7.0.0
:Distributed Coordination/Cluster Coordination
|
Pinging @elastic/es-distributed |
| * and an optional {@link Join}. | ||
| */ | ||
| public class PublishWithJoinResponse extends TransportResponse { | ||
| private final PublishResponse publishResponse; |
There was a problem hiding this comment.
PublishResponse is also a TransportResponse (via TermVersionResponse) but I think because it's wrapped up like this it could just be a standalone object.
There was a problem hiding this comment.
I've removed and inlined TermVersionResponse, see a21e8c7
| assertTrue(node1.coordinationState.electionWon()); | ||
| } | ||
|
|
||
| public void testSimpleClusterStatePublishing() throws InterruptedException { |
There was a problem hiding this comment.
Could this make finer assertions about the state that the publication is in? More precise requests follow...
| publication.pendingPublications.entrySet().stream().collect(shuffle()).forEach(e -> { | ||
| PublishResponse publishResponse = nodeResolver.apply(e.getKey()).coordinationState.handlePublishRequest( | ||
| publication.publishRequest); | ||
| e.getValue().onResponse(new PublishWithJoinResponse(publishResponse, Optional.empty())); |
There was a problem hiding this comment.
Could this assert that publication.applyCommit is null iff n1 hasn't been processed?
There was a problem hiding this comment.
sure, fixed in f09659c
| assertEquals(publication.applyCommit.getTerm(), publication.publishRequest.getAcceptedState().term()); | ||
| assertEquals(publication.applyCommit.getVersion(), publication.publishRequest.getAcceptedState().version()); | ||
| publication.pendingCommits.entrySet().stream().collect(shuffle()).forEach(e -> { | ||
| nodeResolver.apply(e.getKey()).coordinationState.handleCommit(publication.applyCommit); |
There was a problem hiding this comment.
Could we assert publication.completed == false and publication.committed == false at the top of this lambda?
There was a problem hiding this comment.
sure, fixed in f09659c
| assertTrue(publication.completed); | ||
| assertTrue(publication.committed); | ||
|
|
||
| assertThat(ackListener.await(0L, TimeUnit.SECONDS).size(), equalTo(3)); |
There was a problem hiding this comment.
Could this be:
assertThat(ackListener.await(0L, TimeUnit.SECONDS), containsInAnyOrder(n1, n2, n3));
| } | ||
|
|
||
| @Override | ||
| protected void onCompletion(boolean committed) { |
There was a problem hiding this comment.
Can we assert that this is only ever called once?
There was a problem hiding this comment.
sure, done in f09659c
| isCompleted = true; | ||
| onCompletion(true); | ||
| assert applyCommitRequest.isPresent(); | ||
| logger.trace("onPossibleCompletion: [{}] was successful, applying new state locally", this); |
There was a problem hiding this comment.
not sure it makes sense to talk about applying the state here
|
|
||
| protected abstract Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse); | ||
|
|
||
| protected abstract void onPossibleJoin(DiscoveryNode sourceNode, PublishWithJoinResponse response); |
There was a problem hiding this comment.
Does this need the whole PublishWithJoinResponse or could it just take the Optional<Join>? Or maybe even the Join only if present?
There was a problem hiding this comment.
I don't know yet. I've added a TODO in 43e3399
|
|
||
| private class PublicationTarget { | ||
| private final DiscoveryNode discoveryNode; | ||
| private final PublicationTargetStateMachine publicationTargetStateMachine = new PublicationTargetStateMachine(); |
There was a problem hiding this comment.
This was useful for development, but now things are more stable shall we merge this into PublicationTarget?
There was a problem hiding this comment.
ok, fixed in b7b4e39
| } | ||
| publicationTargetStateMachine.setState(PublicationTargetState.SENT_PUBLISH_REQUEST); | ||
| Publication.this.sendPublishRequest(discoveryNode, publishRequest, new PublicationTarget.PublishResponseHandler()); | ||
| // TODO Can this ^ fail with an exception? Target should be failed if so. |
There was a problem hiding this comment.
Does this question have an answer? I think any failures are passed to the response handler.
There was a problem hiding this comment.
I'm deferring this TODO until we integrate this with Legislator.
There was a problem hiding this comment.
I find it a little strange that we don't have any tests with more than 2 nodes in the configuration, to demonstrate that we only need a majority of responses to commit. I don't think this is a gap in coverage so much as in documentation. I suggested a fix for this, and also found one private thing that I think shouldn't be.
| private boolean ackIsPending = true; | ||
| private PublicationTargetState state = PublicationTargetState.NOT_STARTED; | ||
|
|
||
| private PublicationTarget(DiscoveryNode discoveryNode) { |
There was a problem hiding this comment.
Not 100% sure of the rules here, but private seems too restricted for this.
| } | ||
|
|
||
| public void testClusterStatePublishingTimesOutAfterCommit() throws InterruptedException { | ||
| VotingConfiguration config = new VotingConfiguration(Sets.newHashSet(n1.getId(), n2.getId())); |
There was a problem hiding this comment.
Could we (randomly?) add n3 to the configuration here?
There was a problem hiding this comment.
sure, fixed in 799157d
|
@elasticmachine retest this please |
|
Thanks @DaveCTurner |
This PR implements the state machine on the master to publish a cluster state.
Relates to #32006