Deserialize publish requests on generic thread-pool

[nicktindall]

This PR moves the publish_state handler from the CLUSTER_COORDINATION thread pool to the GENERIC one. This means the initial handling of the publish request, including the deserialisation of the cluster state, happens on one of the GENERIC threads instead of the CLUSTER_COORDINATION thread. Once we have deserialised the cluster state and done some validation, we delegate to the CLUSTER_COORDINATION pool to apply the new state.

The consequences of this include

• The generic pool contains multiple threads, but the cluster coordination pool contains a single thread. So (in theory) multiple instances of the handler could now be executing concurrently, where previously these messages would be handled serially.
• The delegation to the cluster coordination thread for the apply means we add some more asynchrony.

Closes #106352

[elasticsearchmachine]

Hi @nicktindall, I've created a changelog YAML for you.

[nicktindall]

onSuccess will end up being called by the CLUSTER_COORDINATION thread only after a successful apply, so I wonder if we need to use compare-and-set here, I don't think it's possible for lastSeenCluster to have changed between this being dispatched AND the new state being successfully applied. I would think either

• the state has not changed and onSuccess is called correctly
• the state has changed and the apply fails due to the version check, so onSuccess is not called

[nicktindall]

The reason I raise this is because we could do away with the onSuccess callback if we were able to safely call lastSeenClusterState.set for both the full and diff payloads.

[DaveCTurner]

I think it can change in between, dispatch and execution, yes - there could be another update in flight when we read the value when then completes and updates it before we get to run.

I am however not sure whether this matters. I'll think about this a little more.

[nicktindall]

This one still concerns me a bit... the task of applying an update is done in three steps

1. deserialise payload (on GENERIC)
2. apply payload if valid (on CLUSTER_COORDINATION)
3. update lastSeen, send response (on CLUSTER_COORDINATION)

only (2) happens in the mutex. So you could have the following interleaving (for cluster states a - version 7 and b - version 8)

1. a-1
2. b-1
3. a-2 (succeeds, bump version to 7)
4. b-2 (succeeds, bump version to 8)
5. b-3 (set lastSeen to 8)
6. a-3 (set lastSeen to 7) - this means the next diff would be applied to 7, and a 7/9 hybrid would be applied locally?

The above could only happen if there were multiple threads in the CLUSTER_COORDINATION pool, but you said that is configurable.

Unless I've missed something.

Possible solutions

• Use compare-and-set also for the non-diff case
  • this feels wrong, I suspect there are times it is correct to updatelastSeen even though it changed since we took a reference to it
• Put an additional check to only bump lastSeen when term and version are >= existing term and version?
  • I don't like this because it'd be duplicating business logic
• Move (3) into the mutex?
  • This feels least bad to me

Do we run tests with CLUSTER_COORDINATION size > 1 ?

[DaveCTurner]

The sequence of 6 steps is indeed something that could have happened prior to #83576 and can again happen after this change - we could end up having applied state version 8 but with lastSeen at state version 7. But that's ok, each diff includes a UUID which identifies the base version which we check here:

elasticsearch/server/src/main/java/org/elasticsearch/cluster/ClusterState.java

Lines 1170 to 1172 in 9ba5651

	if (fromUuid.equals(state.stateUUID) == false) {
	throw new IncompatibleClusterStateVersionException(state.version, state.stateUUID, toVersion, fromUuid);
	}

That IncompatibleClusterStateVersionException is caught and handled here:

elasticsearch/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java

Lines 473 to 483 in 638a450

	if (transportException.unwrapCause() instanceof IncompatibleClusterStateVersionException) {
	logger.debug(
	() -> format(
	"resending full cluster state to node %s reason %s",
	destination,
	transportException.getDetailedMessage()
	)
	);
	sendFullClusterState(destination, delegate);
	return;
	}

IOW if we receive a diff between versions 8 & 9 but lastSeen is at version 7 then we reject the diff and the master sends us the full state at version 9 for us to apply. Somewhat inefficient for sure but still correct.

This case will be exercised in org.elasticsearch.cluster.coordination.CoordinatorTests in a cluster.runRandomly() call, but with rather low probability I think. We could try adding a test there which specifically checks this case.

[nicktindall]

Great! not a problem we need to solve here then :)

[elasticsearchmachine]

Pinging @elastic/es-distributed (Team:Distributed)

[DaveCTurner]

Some initial comments, mostly stylistic tho.

[DaveCTurner]

I think it can change in between, dispatch and execution, yes - there could be another update in flight when we read the value when then completes and updates it before we get to run.

I am however not sure whether this matters. I'll think about this a little more.

[DaveCTurner]

one nit otherwise LGTM

[DaveCTurner]

👍 except that this extra step is happening between steps 3 & 4 in the old list. org.elasticsearch.cluster.coordination.ApplyCommitRequest only carries the term and version, it's org.elasticsearch.cluster.coordination.PublishRequest which carries the state that's being published.

[DaveCTurner]

LGTM (one more tiny suggestion but no need for another review)