Backport of #38311: Move TokenService to seqno powered cas

Author: bleskes

server/src/main/java/org/elasticsearch/action/search/SearchRequestBuilder.java

@@ -223,6 +223,15 @@ public SearchRequestBuilder setVersion(boolean version) {
         return this;
     }
 
+    /**
+     * Should each {@link org.elasticsearch.search.SearchHit} be returned with the
+     * sequence number and primary term of the last modification of the document.
+     */
+    public SearchRequestBuilder seqNoAndPrimaryTerm(boolean seqNoAndPrimaryTerm) {
+        sourceBuilder().seqNoAndPrimaryTerm(seqNoAndPrimaryTerm);
+        return this;
+    }
+
     /**
      * Sets the boost a specific index will receive when the query is executed against it.
      *

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authc/TokenService.java

@@ -35,6 +35,7 @@
 import org.elasticsearch.action.support.WriteRequest.RefreshPolicy;
 import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.AckedClusterStateUpdateTask;
@@ -863,13 +864,17 @@ private void innerRefresh(String tokenDocId, Authentication userAuth, ActionList
                             try (StreamInput in = StreamInput.wrap(Base64.getDecoder().decode(authString))) {
                                 in.setVersion(authVersion);
                                 Authentication authentication = new Authentication(in);
-                                UpdateRequest updateRequest =
+                                UpdateRequestBuilder updateRequest =
                                     client.prepareUpdate(SecurityIndexManager.SECURITY_INDEX_NAME, TYPE, tokenDocId)
-                                        .setVersion(response.getVersion())
                                         .setDoc("refresh_token", Collections.singletonMap("refreshed", true))
-                                        .setRefreshPolicy(RefreshPolicy.WAIT_UNTIL)
-                                        .request();
-                                executeAsyncWithOrigin(client.threadPool().getThreadContext(), SECURITY_ORIGIN, updateRequest,
+                                        .setRefreshPolicy(RefreshPolicy.WAIT_UNTIL);
+                                if (clusterService.state().nodes().getMinNodeVersion().onOrAfter(Version.V_6_7_0)) {
+                                    updateRequest.setIfSeqNo(response.getSeqNo());
+                                    updateRequest.setIfPrimaryTerm(response.getPrimaryTerm());
+                                } else {
+                                    updateRequest.setVersion(response.getVersion());
+                                }
+                                executeAsyncWithOrigin(client.threadPool().getThreadContext(), SECURITY_ORIGIN, updateRequest.request(),
                                     ActionListener.<UpdateResponse>wrap(
                                         updateResponse -> createUserToken(authentication, userAuth, listener, metadata, true),
                                         e -> {