Forbid null ack timeouts (#113443)

We've asserted that `ackTimeout` is non-null for a while now without
issues, so this commit adds code to enforce this invariant.

Author: DaveCTurner

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

@@ -15,6 +15,8 @@
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.core.TimeValue;
 
+import java.util.Objects;
+
 /**
  * An extension interface to {@link ClusterStateUpdateTask} that allows the caller to be notified after the master has
  * computed, published, accepted, committed, and applied the cluster state update AND only after the rest of the nodes
@@ -54,7 +56,7 @@ protected AckedClusterStateUpdateTask(
     ) {
         super(priority, masterNodeTimeout);
         this.listener = (ActionListener<AcknowledgedResponse>) listener;
-        this.ackTimeout = ackTimeout;
+        this.ackTimeout = Objects.requireNonNull(ackTimeout);
     }
 
     /**
@@ -81,10 +83,6 @@ protected AcknowledgedResponse newResponse(boolean acknowledged) {
         return AcknowledgedResponse.of(acknowledged);
     }
 
-    /**
-     * Called once the acknowledgement timeout defined by
-     * {@link AckedClusterStateUpdateTask#ackTimeout()} has expired
-     */
     public void onAckTimeout() {
         listener.onResponse(newResponse(false));
     }
@@ -94,9 +92,6 @@ public void onFailure(Exception e) {
         listener.onFailure(e);
     }
 
-    /**
-     * Acknowledgement timeout, maximum time interval to wait for acknowledgements
-     */
     public final TimeValue ackTimeout() {
         return ackTimeout;
     }

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

@@ -20,39 +20,37 @@
 public interface ClusterStateAckListener {
 
     /**
-     * Called to determine which nodes the acknowledgement is expected from.
+     * Called to determine the nodes from which an acknowledgement is expected.
+     * <p>
+     * This method will be called multiple times to determine the set of acking nodes, so it is crucial for it to return consistent results:
+     * Given the same listener instance and the same node parameter, the method implementation should return the same result.
      *
-     * As this method will be called multiple times to determine the set of acking nodes,
-     * it is crucial for it to return consistent results: Given the same listener instance
-     * and the same node parameter, the method implementation should return the same result.
-     *
-     * @param discoveryNode a node
-     * @return true if the node is expected to send ack back, false otherwise
+     * @return {@code true} if and only if this task will wait for an ack from the given node.
      */
     boolean mustAck(DiscoveryNode discoveryNode);
 
     /**
-     * Called once all the nodes have acknowledged the cluster state update request. Must be
-     * very lightweight execution, since it gets executed on the cluster service thread.
+     * Called once all the selected nodes have acknowledged the cluster state update request. Must be very lightweight execution, since it
+     * is executed on the cluster service thread.
      */
     void onAllNodesAcked();
 
     /**
-     * Called after all the nodes have acknowledged the cluster state update request but at least one of them failed. Must be
-     * very lightweight execution, since it gets executed on the cluster service thread.
+     * Called after all the nodes have acknowledged the cluster state update request but at least one of them failed. Must be very
+     * lightweight execution, since it is executed on the cluster service thread.
      *
-     * @param e optional error that might have been thrown
+     * @param e exception representing the failure.
      */
     void onAckFailure(Exception e);
 
     /**
-     * Called once the acknowledgement timeout defined by
-     * {@link AckedClusterStateUpdateTask#ackTimeout()} has expired
+     * Called if the acknowledgement timeout defined by {@link ClusterStateAckListener#ackTimeout()} expires while still waiting for acks.
      */
     void onAckTimeout();
 
     /**
-     * @return acknowledgement timeout, maximum time interval to wait for acknowledgements
+     * @return acknowledgement timeout, i.e. the maximum time interval to wait for acknowledgements. Return {@link TimeValue#MINUS_ONE} if
+     *         the request should wait indefinitely for acknowledgements.
      */
     TimeValue ackTimeout();
 

server/src/main/java/org/elasticsearch/cluster/service/MasterService.java

@@ -47,7 +47,6 @@
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
-import org.elasticsearch.core.UpdateForV9;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskAwareRequest;
@@ -656,7 +655,7 @@ public void onAckSuccess() {
             }
         }
 
-        public void onAckFailure(@Nullable Exception e) {
+        public void onAckFailure(Exception e) {
             try (ThreadContext.StoredContext ignore = context.get()) {
                 restoreResponseHeaders.run();
                 listener.onAckFailure(e);
@@ -688,6 +687,7 @@ public TimeValue ackTimeout() {
     private static class TaskAckListener {
 
         private final ContextPreservingAckListener contextPreservingAckListener;
+        private final TimeValue ackTimeout;
         private final CountDown countDown;
         private final DiscoveryNode masterNode;
         private final ThreadPool threadPool;
@@ -702,6 +702,7 @@ private static class TaskAckListener {
             ThreadPool threadPool
         ) {
             this.contextPreservingAckListener = contextPreservingAckListener;
+            this.ackTimeout = Objects.requireNonNull(contextPreservingAckListener.ackTimeout());
             this.clusterStateVersion = clusterStateVersion;
             this.threadPool = threadPool;
             this.masterNode = nodes.getMasterNode();
@@ -716,14 +717,7 @@ private static class TaskAckListener {
             this.countDown = new CountDown(countDown + 1); // we also wait for onCommit to be called
         }
 
-        @UpdateForV9 // properly forbid ackTimeout == null after enough time has passed to be sure it's not used in production
         public void onCommit(TimeValue commitTime) {
-            TimeValue ackTimeout = contextPreservingAckListener.ackTimeout();
-            if (ackTimeout == null) {
-                assert false : "ackTimeout must always be present: " + contextPreservingAckListener;
-                ackTimeout = TimeValue.ZERO;
-            }
-
             if (ackTimeout.millis() < 0) {
                 if (countDown.countDown()) {
                     finish();

server/src/main/java/org/elasticsearch/cluster/service/MasterServiceTaskQueue.java

@@ -36,6 +36,9 @@ public interface MasterServiceTaskQueue<T extends ClusterStateTaskListener> {
      *                ?master_timeout}, which is typically available from {@link MasterNodeRequest#masterNodeTimeout()}. Tasks that
      *                correspond with internal actions should normally have no timeout since it is usually better to wait patiently in the
      *                queue until processed rather than to fail, especially if the only reasonable reaction to a failure is to retry.
+     *                <p>
+     *                The values {@code null} and {@link MasterNodeRequest#INFINITE_MASTER_NODE_TIMEOUT} (i.e. {@link TimeValue#MINUS_ONE})
+     *                both indicate that the task should never time out.
      */
     void submitTask(String source, T task, @Nullable TimeValue timeout);
 }