Fixed an issue in serial version logic.

Fixed a case where multiple threads running parallel requests to an
older server might result in unsupported protocol errors.

Author: connelly38

driver/src/main/java/oracle/nosql/driver/http/Client.java

@@ -162,7 +162,7 @@ public class Client {
      */
     private ExecutorService threadPool;
 
-    private short serialVersion = DEFAULT_SERIAL_VERSION;
+    private volatile short serialVersion = DEFAULT_SERIAL_VERSION;
 
     /* for one-time messages */
     private final HashSet<String> oneTimeMessages;
@@ -489,7 +489,7 @@ public Result execute(Request kvRequest) {
             }
 
             ResponseHandler responseHandler = null;
-
+            short serialVersionUsed = serialVersion;
             ByteBuf buffer = null;
             long networkLatency;
             try {
@@ -514,7 +514,7 @@ public Result execute(Request kvRequest) {
                  */
                 kvRequest.setCheckRequestSize(false);
 
-                writeContent(buffer, kvRequest);
+                serialVersionUsed = writeContent(buffer, kvRequest);
 
                 /*
                  * If on-premise the authProvider will always be a
@@ -574,7 +574,7 @@ public Result execute(Request kvRequest) {
                 int resSize = wireContent.readerIndex();
                 networkLatency = System.currentTimeMillis() - networkLatency;
 
-                if (serialVersion < 3) {
+                if (serialVersionUsed < 3) {
                     /* so we can emit a one-time message if the app */
                     /* tries to access modificationTime */
                     if (res instanceof GetResult) {
@@ -690,7 +690,7 @@ public Result execute(Request kvRequest) {
                 continue;
             } catch (UnsupportedProtocolException upe) {
                 /* reduce protocol version and try again */
-                if (decrementSerialVersion() == true) {
+                if (decrementSerialVersion(serialVersionUsed) == true) {
                     exception = upe;
                     logInfo(logger, "Got unsupported protocol error " +
                             "from server: decrementing serial version to " +
@@ -933,15 +933,17 @@ boolean timeoutRequest(long startTime,
      *
      * @throws IOException
      */
-    void writeContent(ByteBuf content, Request kvRequest)
+    private short writeContent(ByteBuf content, Request kvRequest)
         throws IOException {
 
         final NettyByteOutputStream bos = new NettyByteOutputStream(content);
-        bos.writeShort(serialVersion);
+        final short versionUsed = serialVersion;
+        bos.writeShort(versionUsed);
         kvRequest.createSerializer(factory).
             serialize(kvRequest,
-                      serialVersion,
+                      versionUsed,
                       bos);
+        return versionUsed;
     }
 
     /**
@@ -1244,7 +1246,10 @@ StatsControl getStatsControl() {
      * @return true: version was decremented
      *         false: already at lowest version number.
      */
-    public boolean decrementSerialVersion() {
+    private synchronized boolean decrementSerialVersion(short versionUsed) {
+        if (serialVersion != versionUsed) {
+            return true;
+        }
         if (serialVersion == V3) {
             serialVersion = V2;
             return true;

driver/src/test/java/oracle/nosql/driver/InternalsTest.java

@@ -0,0 +1,88 @@
+/*-
+ * Copyright (c) 2011, 2022 Oracle and/or its affiliates. All rights reserved.
+ *
+ * Licensed under the Universal Permissive License v 1.0 as shown at
+ *  https://oss.oracle.com/licenses/upl/
+ */
+
+package oracle.nosql.driver;
+
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import oracle.nosql.driver.ops.ListTablesRequest;
+
+public class InternalsTest extends ProxyTestBase {
+
+    @Override
+    protected void perTestHandleConfig(NoSQLHandleConfig config) {
+        config.setConnectionPoolSize(10);
+        config.setNumThreads(8);
+    }
+
+    @Before
+    @Override
+    /*
+     * Override the default beforeTest() so we do not make
+     * any requests at all to the server before the actual test
+     * is run. The default runs a ListTables request.
+     */
+    public void beforeTest() throws Exception {
+        handle = getHandle(endpoint);
+        /* do NOT list tables here */
+    }
+
+    @Test
+    public void serialSetupTest() {
+        /*
+         * Start N threads. Make them all run at the same time.
+         * Each thread will issue requests. Verify that none of them
+         * throw UnsupportedProtocolException. This verifies that the
+         * logic inside Client properly manages setting the serial version
+         * on the first request(s) that may happen in parallel.
+         *
+         * This test really only does anything if the server it's running
+         * against is an older version. But it's ok to run with the current
+         * server version anyway.
+         */
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicInteger errors = new AtomicInteger(0);
+        final int numThreads = 5;
+        Thread[] threads = new Thread[numThreads];
+        for (int x=0; x<numThreads; x++) {
+            threads[x] = new Thread(() -> {
+                try { latch.await(); } catch (Exception e) {}
+                runRequests(errors);
+            });
+            threads[x].start();
+        }
+
+        latch.countDown();
+
+        /* wait for threads to finish */
+        for(int x=0; x<numThreads; x++) {
+            try { threads[x].join(); } catch (Exception e) {}
+        }
+        if (errors.get() > 0) {
+            fail("Got " + errors.get() + " errors");
+        }
+    }
+
+    private void runRequests(AtomicInteger errors) {
+        ListTablesRequest req = new ListTablesRequest()
+              .setLimit(1);
+        try {
+            for (int x=0; x<5; x++) {
+                handle.listTables(req);
+            }
+        } catch (Exception e) {
+            System.out.println("Got exception: " + e);
+            errors.incrementAndGet();
+        }
+    }
+}

driver/src/test/java/oracle/nosql/driver/ProxyTestBase.java

@@ -387,15 +387,13 @@ static private TableResult dropTableWithoutWait(NoSQLHandle nosqlHandle,
     }
 
     protected NoSQLHandle getHandle(String ep) {
-
         NoSQLHandleConfig config = new NoSQLHandleConfig(ep);
         serviceURL = config.getServiceURL();
         return setupHandle(config);
     }
 
     /* Set configuration values for the handle */
     protected NoSQLHandle setupHandle(NoSQLHandleConfig config) {
-
         /*
          * 5 retries, default retry algorithm
          */
@@ -412,12 +410,7 @@ protected NoSQLHandle setupHandle(NoSQLHandleConfig config) {
 
         NoSQLHandle h = getHandle(config);
 
-        /* this will set up the right protocol serial version */
-        try {
-            getTable("noop", h);
-        } catch (Exception e) {
-            /* ignore errors */
-        }
+        /* serial version will be set by default ListTables() in beforeTest */
 
         return h;
     }