[yugabyte#27062] YSQL: Fix ConcurrentTablespaceTest.testAlterIndexSetTablespace

Summary:
Currently, `ConcurrentTablespaceTest.testAlterIndexSetTablespace` has a concurrency bug due to the way `CyclicBarrier` is used.

The test starts two threads, one that does DDL and one that does DML, and has them each execute 12 statements. Before executing each statement, each thread calls `CyclicBarrier.await` to ensure they are executing the statements at the same time.

### The problem

The problem is when the DML thread can runs into retriable errors (which is expected)—if this happens, the statement is retried without incrementing the statement counter, but the thread calls `CyclicBarrier.await` before retrying the statement. This means that if thread A runs into an error on statement 10, it'll retry statement 10 but this retry will be synchronized with other threads' statement 11.

This causes an issue at the end of the test when the threads run into uneven numbers of errors; since `CyclicBarrier.await` waits until all threads call `await`, it's possible that one thread finishes executing, so it never calls `await`, but another thread is stuck waiting at `await`.

### The fix

We fix this by swapping out the `CyclicBarrier` for a `Phaser`--`CyclicBarrier` requires a fixed number of threads, while `Phaser` allows for a dynamic number of threads. Using the `Phaser`, we de-register threads that are done executing, allowing the remaining threads to continue execution afterwards.
Jira: DB-16543

Test Plan:
```
./yb_build.sh release --java-test org.yb.pgsql.ConcurrentTablespaceTest#testAlterIndexSetTablespace
```

Reviewers: sanketh, myang

Reviewed By: myang

Differential Revision: https://phorge.dev.yugabyte.com/D43992

Author: kai-franz

java/yb-pgsql/src/test/java/org/yb/pgsql/ConcurrentTablespaceTest.java

@@ -121,7 +121,7 @@ private Tablespace[] generateTestTablespaces() {
 
   private List<Thread> setupConcurrentDdlDmlThreads(String ddlTemplate) {
     final int totalThreads = numDmlThreads + numDdlThreads;
-    final CyclicBarrier barrier = new CyclicBarrier(totalThreads);
+    final Phaser phaser = new Phaser(totalThreads);
     final List<Thread> threads = new ArrayList<>();
 
     // Add the DDL thread.
@@ -131,14 +131,14 @@ private List<Thread> setupConcurrentDdlDmlThreads(String ddlTemplate) {
               connections[i],
               ddlTemplate,
               errorsDetected,
-              barrier,
+              phaser,
               numStmtsPerThread,
               tablespaces));
     }
 
     // Add the DML threads.
     for (int i = numDdlThreads; i < totalThreads; ++i) {
-      threads.add(new DMLRunner(connections[i], errorsDetected, barrier, numStmtsPerThread, i));
+      threads.add(new DMLRunner(connections[i], errorsDetected, phaser, numStmtsPerThread, i));
     }
     return threads;
   }
@@ -223,7 +223,7 @@ public void testTableCreationFailure() throws Exception {
     YBClient client = miniCluster.getClient();
     connections = setupConnections();
     final int totalThreads = numDmlThreads + numDdlThreads;
-    final CyclicBarrier barrier = new CyclicBarrier(totalThreads);
+    final Phaser phaser = new Phaser(totalThreads);
     final List<Thread> threads = new ArrayList<>();
     AtomicBoolean invalidPlacementError = new AtomicBoolean(false);
 
@@ -236,7 +236,7 @@ public void testTableCreationFailure() throws Exception {
             connections[0],
             "CREATE TABLE validplacementtable (a int) TABLESPACE %s",
             errorsDetected,
-            barrier,
+            phaser,
             1,
             new Tablespace[] {valid_ts}));
 
@@ -248,7 +248,7 @@ public void testTableCreationFailure() throws Exception {
             connections[1],
             "CREATE TABLE invalid_placementtable (a int) TABLESPACE %s",
             invalidPlacementError,
-            barrier,
+            phaser,
             1,
             new Tablespace[] {invalid_ts}));
 
@@ -268,19 +268,19 @@ public void testTableCreationFailure() throws Exception {
   public abstract class SQLRunner extends Thread {
     protected final Connection conn;
     protected final AtomicBoolean errorsDetected;
-    protected final CyclicBarrier barrier;
+    protected final Phaser phaser;
     protected final int numStmtsPerThread;
     protected int idx; // Only used by DMLRunner
 
     public SQLRunner(
         Connection conn,
         AtomicBoolean errorsDetected,
-        CyclicBarrier barrier,
+        Phaser phaser,
         int numStmtsPerThread,
         int idx) {
       this.conn = conn;
       this.errorsDetected = errorsDetected;
-      this.barrier = barrier;
+      this.phaser = phaser;
       this.numStmtsPerThread = numStmtsPerThread;
       this.idx = idx; // This field is not used in DDLRunner
     }
@@ -289,16 +289,22 @@ public SQLRunner(
     public void run() {
       int item_idx = 0;
       while (item_idx < numStmtsPerThread && !errorsDetected.get()) {
-        try (Statement lstmt = conn.createStatement()) {
-          barrier.await();
-          executeStatement(lstmt, item_idx);
+        try {
+          phaser.arriveAndAwaitAdvance();
+
+          try (Statement lstmt = conn.createStatement()) {
+            executeStatement(lstmt, item_idx);
+          }
+
           item_idx++;
         } catch (PSQLException e) {
           handlePSQLException(e);
-        } catch (SQLException | InterruptedException | BrokenBarrierException e) {
+        } catch (SQLException e) {
           logAndSetError(e);
         }
       }
+      // Thread finished: shrink the party count.
+      phaser.arriveAndDeregister();
     }
 
     protected abstract void executeStatement(Statement lstmt, int item_idx) throws SQLException;
@@ -324,7 +330,7 @@ private void handlePSQLException(PSQLException e) {
     protected void logAndSetError(Exception e) {
       LOG.info("SQL thread: Unexpected error: ", e);
       errorsDetected.set(true);
-      barrier.reset();
+      phaser.forceTermination();
     }
   }
 
@@ -337,10 +343,10 @@ public class DMLRunner extends SQLRunner {
     public DMLRunner(
         Connection conn,
         AtomicBoolean errorsDetected,
-        CyclicBarrier barrier,
+        Phaser phaser,
         int numStmtsPerThread,
         int idx) {
-      super(conn, errorsDetected, barrier, numStmtsPerThread, idx);
+      super(conn, errorsDetected, phaser, numStmtsPerThread, idx);
     }
 
     @Override
@@ -372,10 +378,10 @@ public DDLRunner(
         Connection conn,
         String sql,
         AtomicBoolean errorsDetected,
-        CyclicBarrier barrier,
+        Phaser phaser,
         int numStmtsPerThread,
         Tablespace[] tablespaces) {
-      super(conn, errorsDetected, barrier, numStmtsPerThread, 0); // idx is not used here
+      super(conn, errorsDetected, phaser, numStmtsPerThread, 0); // idx is not used here
       this.sql = sql;
       this.tablespaces = tablespaces;
     }