Interpret ?timeout=-1 as infinite ack timeout

[DaveCTurner]

APIs which perform cluster state updates typically accept the
?master_timeout= and ?timeout= parameters to respectively set the
pending task queue timeout and the acking timeout for the cluster state
update. Both of these parameters accept the value -1, but
?master_timeout=-1 means to wait indefinitely whereas ?timeout=-1
means the same thing as ?timeout=0, namely that acking times out
immediately on commit.

There are some situations where it makes sense to wait for as long as
possible for nodes to ack a cluster state update. In practice this wait
is bounded by other mechanisms (e.g. the lag detector will remove the
node from the cluster after a couple of minutes of failing to apply
cluster state updates) but these are not really the concern of clients.

Therefore with this commit we change the meaning of ?timeout=-1 to
mean that the acking timeout is infinite.

[github-actions]

Documentation preview:

• ✨ Changed pages

[elasticsearchmachine]

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

[elasticsearchmachine]

Hi @DaveCTurner, I've created a changelog YAML for you. Note that since this PR is labelled >breaking, you need to update the changelog YAML to fill out the extended information sections.

[DaveCTurner]

We've reached consensus that this is an acceptable change to make so this is good to review now.

[ywangd]

LGTM

Sorry for the delayed review. It is always a great learning opportunity to see changes around master coordination. Thanks!

[ywangd]

Not really related to this PR but for my learning. I am trying to understand the sequence of how cluster state happens. Is the following order correct?

1. Master sends cluster state publish requests to all nodes
2. After receiving publish response from a quorum of nodes, this onCommit is called.
3. Master sends apply commit requests to all nodes
4. On each apply commit response, master calls onNodeAck.

When step 2 completes and all nodes responded in step 4, the overall request is considered as acknowledged.

Reading the code here, it seems to me that when onCommit is called, there is a chance that step 4 has already completed (since it checks the countDown and call finish). But I am not sure how that can happen since onCommit is called before any apply commit request can be sent (code)? Or is it to take care of single node cluster? I must be missing something (or even many things). I'd appreciate if you could help clarify it. Thanks!

[DaveCTurner]

The sequence is right.

You could be right about onCommit never actually finishing the acking today. FWIW this code was added in #31303 (6.4.0) and we've rewritten much of the surrounding code since then. That said, it's a fairly delicate argument to prove this, whereas the "obviously correct" code as written today is robust and not meaningfully less efficient.

In particular, it's hard to see that onCommit is called before any ApplyCommit is sent. The relevant code is org.elasticsearch.cluster.coordination.Publication.PublicationTarget#handlePublishResponse:

        void handlePublishResponse(PublishResponse publishResponse) {
            assert isWaitingForQuorum() : this;
            logger.trace("handlePublishResponse: handling [{}] from [{}])", publishResponse, discoveryNode);
            if (applyCommitRequest.isPresent()) {
                sendApplyCommit();
            } else {
                try {
                    Publication.this.handlePublishResponse(discoveryNode, publishResponse).ifPresent(applyCommit -> {
                        assert applyCommitRequest.isPresent() == false;
                        applyCommitRequest = Optional.of(applyCommit);
                        ackListener.onCommit(TimeValue.timeValueMillis(currentTimeSupplier.getAsLong() - startTime));
                        publicationTargets.stream()
                            .filter(PublicationTarget::isWaitingForQuorum)
                            .forEach(PublicationTarget::sendApplyCommit);
                    });
                } catch (Exception e) {
                    setFailed(e);
                    onPossibleCommitFailure();
                }
            }
        }

As written, you could have the committing thread setting applyCommitRequest, then pausing before calling ackListener.onCommit, while another thread concurrently processes a later PublishResponse, discovers that applyCommitRequest.isPresent() and sends the ApplyCommit. But then if you look at how this code is called eventually you discover that this all happens underneath Coordinator#mutex so these things cannot happen concurrently. But relying on a mutex in Coordinator to protect against concurrency in Publication as part of the correctness argument for MasterService is too deeply opaque for my liking.

[ywangd]

Thanks a lot for the explanation. TIL 🙇

[ywangd]

It would be great to also explain how master_timeout is computed. By reading the code, it seems to start when the task is added to the queue and expires if the task does not get processed by the master. And I believe during this waiting, the task is visible via the PendingTasks API?

[elasticsearchmachine]

Hi @DaveCTurner, I've updated the changelog YAML for you. Note that since this PR is labelled >breaking, you need to update the changelog YAML to fill out the extended information sections.