Fixes for Polaris race conditions

NodeConverter had a large amount of dead code - including a race condition! - that was deleted.

Author: rjrudin

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/QueryBatcherImpl.java

@@ -1004,6 +1004,8 @@ public void run() {
               }
             };
             threadPool.execute(processBatch);
+			  // Polaris warns of a race condition here, presumably based on the check on remainingCapacity. But the
+			  // cost of executing a needless IteratorTask is small enough to accept the possible race condition.
             // If the queue is almost full, stop producing and add a task to continue later
             if (isSingleThreaded && threadPool.getQueue().remainingCapacity() <= 2 && iterator.hasNext()) {
               threadPool.execute(new IteratorTask(batcher));

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/RowBatcherImpl.java

@@ -381,9 +381,9 @@ public synchronized void start(JobTicket ticket) {
 		// not the same value that a user would have provided via withBatchSize. And we don't want to log it when it's
 		// -1, which will be the case for a single batch.
 		if (logger.isDebugEnabled() && this.batchSize > 0) {
-			logger.debug("batch count: {}, calculated batch size: {}", batchCount, batchSize);
+			logger.debug("batch count: {}, calculated batch size: {}", this.batchCount, this.batchSize);
 		} else {
-			logger.info("batch count: {}", batchCount);
+			logger.info("batch count: {}", this.batchCount);
 		}
 
         if (this.hostInfos != null && getMoveMgr().getConnectionType() == DatabaseClient.ConnectionType.DIRECT) {

marklogic-client-api/src/main/java/com/marklogic/client/dataservices/impl/IOEndpointImpl.java

@@ -300,9 +300,6 @@ static class CallerThreadPoolExecutor<I,O> extends ThreadPoolExecutor {
                 this.bulkIOEndpointCaller = bulkIOEndpointCaller;
             }
 
-            Boolean isAwaitingTermination() {
-                return this.awaitingTermination;
-            }
             synchronized void awaitTermination() throws InterruptedException {
                 if (bulkIOEndpointCaller.getCallContextQueue().isEmpty() && getActiveCount()<=1) {
                     shutdown();

marklogic-client-api/src/main/java/com/marklogic/client/impl/HTTPSamlAuthInterceptor.java

@@ -4,62 +4,77 @@
 
 package com.marklogic.client.impl;
 
-import java.io.IOException;
-import java.time.Instant;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicBoolean;
-
 import com.marklogic.client.DatabaseClientFactory.SAMLAuthContext.AuthorizerCallback;
 import com.marklogic.client.DatabaseClientFactory.SAMLAuthContext.ExpiringSAMLAuth;
 import com.marklogic.client.DatabaseClientFactory.SAMLAuthContext.RenewerCallback;
-
 import okhttp3.Interceptor;
 import okhttp3.Request;
 import okhttp3.Response;
 
+import java.io.IOException;
+import java.time.Instant;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
 public class HTTPSamlAuthInterceptor implements Interceptor {
 
-    private String authorizationTokenValue;
-    private AuthorizerCallback authorizer;
+    private final AuthorizerCallback authorizer;
+	private final RenewerCallback renewer;
+
+	private String authorizationTokenValue;
     private ExpiringSAMLAuth expiringSAMLAuth;
     private long threshold;
-    private RenewerCallback renewer;
     private AtomicBoolean isCallbackExecuting;
 
     public HTTPSamlAuthInterceptor(String authToken) {
-        this.authorizationTokenValue = authToken;
+		this.authorizationTokenValue = authToken;
+		this.authorizer = null;
+		this.renewer = null;
     }
 
     public HTTPSamlAuthInterceptor(AuthorizerCallback authorizer) {
         this.authorizer = authorizer;
+		this.renewer = null;
     }
 
     public HTTPSamlAuthInterceptor(ExpiringSAMLAuth authorization, RenewerCallback renew) {
         expiringSAMLAuth = authorization;
         renewer = renew;
         isCallbackExecuting = new AtomicBoolean(false);
+		this.authorizer = null;
     }
 
     @Override
     public Response intercept(Chain chain) throws IOException {
-        Request request = chain.request();
-        if (authorizer != null) {
-            if(expiringSAMLAuth == null) {
-                authorizeCallbackWrapper(null);
-            } else if(threshold<=Instant.now().getEpochSecond()){
-                authorizeCallbackWrapper(expiringSAMLAuth.getExpiry());
-            }
-        } else if (renewer != null && threshold <= Instant.now().getEpochSecond() && isCallbackExecuting.compareAndSet(false, true)) {
-                RenewCallbackWrapper renewCallbackWrapper = new RenewCallbackWrapper(expiringSAMLAuth);
-                Executors.defaultThreadFactory().newThread(renewCallbackWrapper).start();
-        }
-        String samlHeaderValue = RESTServices.AUTHORIZATION_TYPE_SAML + " " + RESTServices.AUTHORIZATION_PARAM_TOKEN
-                + "=" + authorizationTokenValue;
-        Request authenticatedRequest = request.newBuilder().header(RESTServices.HEADER_AUTHORIZATION, samlHeaderValue)
-                .build();
-        return chain.proceed(authenticatedRequest);
+		if (authorizer != null) {
+			authorizeRequest();
+		} else if (renewer != null && threshold <= Instant.now().getEpochSecond() && isCallbackExecuting.compareAndSet(false, true)) {
+			RenewCallbackWrapper renewCallbackWrapper = new RenewCallbackWrapper(expiringSAMLAuth);
+			Executors.defaultThreadFactory().newThread(renewCallbackWrapper).start();
+		}
+
+		Request authenticatedRequest = chain.request().newBuilder()
+			.header(RESTServices.HEADER_AUTHORIZATION, buildSamlHeader())
+			.build();
+
+		return chain.proceed(authenticatedRequest);
     }
 
+	private synchronized void authorizeRequest() {
+		if (expiringSAMLAuth == null) {
+			authorizeCallbackWrapper(null);
+		} else if (threshold <= Instant.now().getEpochSecond()) {
+			authorizeCallbackWrapper(expiringSAMLAuth.getExpiry());
+		}
+	}
+
+	private synchronized String buildSamlHeader() {
+		return String.format("%s %s=%s",
+			RESTServices.AUTHORIZATION_TYPE_SAML,
+			RESTServices.AUTHORIZATION_PARAM_TOKEN,
+			this.authorizationTokenValue);
+	}
+
     private synchronized void authorizeCallbackWrapper(Instant expiry) {
 
         if(expiry == null && expiringSAMLAuth != null) {