chore: bump java-bigtable to 2.85.0, beam to 2.77.0, emulator to 0.222.0 (#4640)

* chore: bump java-bigtable to 2.85.0, beam to 2.77.0, emulator to 0.222.0

- java-bigtable: 2.81.0 → 2.85.0
- google-cloud-bigtable-emulator: 0.218.0 → 0.222.0
- beam: 2.72.0 → 2.77.0
- conscrypt: 2.6.0 → 2.6.2 (to match bigtable 2.85.0 deps-bom)
- maven.compiler.source/target: 1.8 → 11 (beam 2.77+ requires Java 11)

Add dev.cel relocation to 1.x-shaded and 2.x-shaded: bigtable 2.85.0
pulls in grpc-xds which transitively introduces dev.cel:0.13.0; without
relocation those classes leak un-relocated into the shaded jars and
trip verify-shaded-jar-entries.

Upgrade kokoro java8 CI jobs to use the java11 docker image (presubmit,
continuous, nightly). Move bigtable-dataflow-parent into a JDK-activated
profile so it is excluded from any remaining Java 8 build environments.

Change-Id: I558368876e3e46827097598f426abc8469e503cf

* test: deflake replication endpoint and buffered mutator tests

Replication endpoint tests (TestUtils, shared by 1.x and 2.x):

- waitForReplication polled 5 times with a doubling sleep, but the final
  8s sleep was not followed by another check, so the effective budget was
  7.5s and a failure cost 8s of dead time. Replace it with a deadline
  based poll (60s, every 500ms) that always re-checks after sleeping.
- The replicated-entry counter is static and shared by every test in the
  class. Resetting it in @before is not enough because replication is
  asynchronous: entries still in flight from the previous test arrive
  after the reset and count towards the next test's target. This is worst
  after the 10k-row test. Drain replication before resetting the counter,
  and retry the table comparison in assertTableEventuallyEquals so the
  assertion waits on the replicated data rather than the counter.
- Close the ResultScanners in assertTableEquals; they were leaked, which
  matters now that the comparison runs in a retry loop.

TestBufferedMutator:

- testAutoFlushOff/testAutoFlushOn write on one Connection and read back
  on another, asserting on the first response. flush() makes the write
  durable but does not guarantee a read on a separate connection observes
  it immediately, so the read could return 0 cells. Retry the read; it
  only absorbs that delay and still fails if the write never appears.
- Close the ResultScanner in testBulkMutation.

Change-Id: I5a1936b0037965f6306f545164c014a0a6306c3d

* test: raise modifyTableAsync timeout in TestModifyTable

testModifyTable_HBase2_Async timed out waiting 60s on modifyTableAsync.
modifyTableAsync chains a plain thenApply off getDescriptor, and because
ApiFutureUtils completes its futures with a directExecutor, that
continuation runs on a gRPC callback thread -- where it then blocks on
the modifyFamilies RPC. With the suite running admin operations across
four classes in parallel these pile up and starve the callback pool, so
the call can take well over a minute. Raise the wait to 5 minutes; the
1800s fork timeout still bounds a genuine hang.

This only stops the test failing. The blocking continuation is still
there and can stall modifyTable/modifyTableAsync for callers.

Change-Id: Ibdd83e80ad47f429791d8e189dead12d2f96ec54

Author: mutianf

.kokoro/continuous/java8.cfg

@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-kokoro-resources/java8"
+  value: "gcr.io/cloud-devrel-kokoro-resources/java11"
 }
 
 env_vars: {

.kokoro/nightly/java8.cfg

@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-kokoro-resources/java8"
+  value: "gcr.io/cloud-devrel-kokoro-resources/java11"
 }
 
 env_vars: {

.kokoro/presubmit/java8.cfg

@@ -3,7 +3,7 @@
 # Configure the docker image for kokoro-trampoline.
 env_vars: {
   key: "TRAMPOLINE_IMAGE"
-  value: "gcr.io/cloud-devrel-kokoro-resources/java8"
+  value: "gcr.io/cloud-devrel-kokoro-resources/java11"
 }
 
 env_vars: {

bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/TestBufferedMutator.java

@@ -21,9 +21,11 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang.RandomStringUtils;
 import org.apache.hadoop.hbase.client.BufferedMutator;
 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
@@ -61,7 +63,7 @@ public void testAutoFlushOff() throws Exception {
       // Bigtable pushes the change right away.  This test would be flaky.
       //      Assert.assertEquals("Expecting no results", 0, tableForRead.get(get).size());
       mutator.flush();
-      Assert.assertEquals("Expecting one result", 1, tableForRead.get(get).size());
+      assertEventuallyHasCells(tableForRead, get, 1);
     }
   }
 
@@ -71,7 +73,7 @@ public void testAutoFlushOn() throws Exception {
         Connection c = createNewConnection();
         Table tableForRead = c.getTable(sharedTestEnv.getDefaultTableName()); ) {
       mutator.put(getPut());
-      Assert.assertEquals("Expecting one result", 1, tableForRead.get(getGet()).size());
+      assertEventuallyHasCells(tableForRead, getGet(), 1);
     }
   }
 
@@ -190,8 +192,10 @@ public void testBulkMutation() throws Exception {
 
       Scan scan = new Scan();
       scan.setRowPrefixFilter(Bytes.toBytes(rowKeyPrefix));
-      ResultScanner resultScanner = tableForRead.getScanner(scan);
-      Result[] results = resultScanner.next(20);
+      Result[] results;
+      try (ResultScanner resultScanner = tableForRead.getScanner(scan)) {
+        results = resultScanner.next(20);
+      }
 
       assertEquals("mutations should have been applied now", mutations.size(), results.length);
       for (int i = 0; i < results.length; i++) {
@@ -204,6 +208,30 @@ public void testBulkMutation() throws Exception {
   final byte[] rowKey = dataHelper.randomData("testrow-");
   final byte[] value = dataHelper.randomData("testValue-");
 
+  /**
+   * Reads until {@code get} returns {@code expectedCells} cells, or fails once the timeout expires.
+   *
+   * <p>These tests read back through a second {@link Connection} to prove the write reached
+   * Bigtable rather than a client side buffer. A write is durable once the mutator returns, but a
+   * read issued on a different connection is not guaranteed to observe it immediately, so asserting
+   * on the very first response makes the test flaky. Retrying only absorbs that delay: if the write
+   * never becomes visible, this still fails with the same expected/actual mismatch.
+   */
+  private static void assertEventuallyHasCells(Table table, Get get, int expectedCells)
+      throws IOException, InterruptedException {
+    long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
+    int actualCells;
+    do {
+      actualCells = table.get(get).size();
+      if (actualCells == expectedCells) {
+        return;
+      }
+      Thread.sleep(250);
+    } while (System.nanoTime() < deadline);
+
+    assertEquals("Expecting " + expectedCells + " result(s)", expectedCells, actualCells);
+  }
+
   private Get getGet() {
     Get get = new Get(rowKey);
     get.addColumn(COLUMN_FAMILY, qualifier);

bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml

@@ -389,6 +389,10 @@ limitations under the License.
                   <pattern>org.jspecify.annotations</pattern>
                   <shadedPattern>com.google.bigtable.repackaged.org.jspecify.annotations</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>dev.cel</pattern>
+                  <shadedPattern>com.google.bigtable.repackaged.dev.cel</shadedPattern>
+                </relocation>
               </relocations>
             </configuration>
           </execution>

bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestModifyTable.java

@@ -70,7 +70,9 @@ public void testModifyTable_HBase2_Async() throws IOException {
           @Override
           public Void apply(@Nullable HTableDescriptor descriptor) {
             try (Admin admin = getConnection().getAdmin()) {
-              admin.modifyTableAsync(descriptor).get(60, TimeUnit.SECONDS);
+              // Generous: modifyTableAsync blocks a gRPC callback thread while it waits on the
+              // modifyFamilies RPC, so it can stall when the suite runs admin ops in parallel.
+              admin.modifyTableAsync(descriptor).get(5, TimeUnit.MINUTES);
             } catch (Exception e) {
               throw new RuntimeException(e);
             }

bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml

@@ -380,6 +380,10 @@ limitations under the License.
                   <pattern>org.jspecify.annotations</pattern>
                   <shadedPattern>com.google.bigtable.repackaged.org.jspecify.annotations</shadedPattern>
                 </relocation>
+                <relocation>
+                  <pattern>dev.cel</pattern>
+                  <shadedPattern>com.google.bigtable.repackaged.dev.cel</shadedPattern>
+                </relocation>
               </relocations>
             </configuration>
           </execution>

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java

@@ -149,7 +149,11 @@ public static void setUpCluster() throws Exception {
   }
 
   @Before
-  public void setupTestCase() throws IOException {
+  public void setupTestCase() throws IOException, InterruptedException {
+
+    // Drain any replication still in flight from the previous test, so that it is not counted
+    // against this test's target by the reset below.
+    TestUtils.waitForReplicationToQuiesce(TestReplicationEndpoint.replicatedEntries::get);
 
     // Create and set the empty tables
     TableName table1 = TableName.valueOf(UUID.randomUUID().toString());

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java

@@ -157,7 +157,11 @@ public static void tearDown() throws Exception {
   }
 
   @Before
-  public void setupTestCase() throws IOException {
+  public void setupTestCase() throws IOException, InterruptedException {
+
+    // Drain any replication still in flight from the previous test, so that it is not counted
+    // against this test's target by the reset below.
+    TestUtils.waitForReplicationToQuiesce(TestReplicationEndpoint.replicatedEntries::get);
 
     // Create and set the empty tables
     TableName table1 = TableName.valueOf(UUID.randomUUID().toString());

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/utils/TestUtils.java

@@ -25,7 +25,9 @@
 import java.util.NavigableMap;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.BooleanSupplier;
+import java.util.function.IntSupplier;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.TableName;
@@ -65,6 +67,15 @@ public class TestUtils {
   public static final byte[] VALUE = "Value".getBytes();
   public static final long TIMESTAMP = 1000l;
 
+  /** How long to wait for asynchronous replication before giving up. */
+  private static final long REPLICATION_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(60);
+
+  /** How long to wait between polls while waiting on asynchronous replication. */
+  private static final long POLL_INTERVAL_MILLIS = 500;
+
+  /** Consecutive unchanged polls required before replication is considered idle. */
+  private static final int QUIESCE_STABLE_POLLS = 3;
+
   // ONLY STATIC METHODS
   private TestUtils() {}
 
@@ -180,52 +191,77 @@ public static void assertEquals(RowMutations expected, RowMutations actual) {
   /** Scans the 2 Tables and compares the data. Fails assertion in case of a mismatch. */
   public static void assertTableEquals(Table expected, Table actual) throws IOException {
 
-    ResultScanner expectedScanner = expected.getScanner(new Scan().setMaxVersions());
-    ResultScanner actualScanner = actual.getScanner(new Scan().setMaxVersions());
+    try (ResultScanner expectedScanner = expected.getScanner(new Scan().setMaxVersions());
+        ResultScanner actualScanner = actual.getScanner(new Scan().setMaxVersions())) {
 
-    Iterator<Result> expectedIterator = expectedScanner.iterator();
-    Iterator<Result> actualIterator = actualScanner.iterator();
+      Iterator<Result> expectedIterator = expectedScanner.iterator();
+      Iterator<Result> actualIterator = actualScanner.iterator();
 
-    // Both databases should return exact same results in exact same order.
-    while (expectedIterator.hasNext()) {
-      // Check size mismatch
-      if (!actualIterator.hasNext()) {
+      // Both databases should return exact same results in exact same order.
+      while (expectedIterator.hasNext()) {
+        // Check size mismatch
+        if (!actualIterator.hasNext()) {
+          Assert.fail(
+              "Expected more rows. CBT missing results starting at "
+                  + Bytes.toStringBinary(expectedIterator.next().getRow()));
+        }
+
+        // Both the databases have next, so compare the results
+        assertEquals(expectedIterator.next(), actualIterator.next());
+      }
+      // Verify that actual has no more data left
+      if (actualIterator.hasNext()) {
         Assert.fail(
-            "Expected more rows. CBT missing results starting at "
-                + Bytes.toStringBinary(expectedIterator.next().getRow()));
+            "Expected less rows, extra results in CBT starting from "
+                + actualIterator.next().toString());
       }
-
-      // Both the databases have next, so compare the results
-      assertEquals(expectedIterator.next(), actualIterator.next());
-    }
-    // Verify that actual has no more data left
-    if (actualIterator.hasNext()) {
-      Assert.fail(
-          "Expected less rows, extra results in CBT starting from "
-              + actualIterator.next().toString());
     }
   }
 
   /** Waits for replication to replicate numReplicatedEntries */
   public static void waitForReplication(BooleanSupplier isReplicationCurrent)
       throws InterruptedException {
-    int sleepTimeInMillis = 500;
-    for (int i = 0; i < 5; i++) {
-
-      // First check if replication has already happened
+    long start = System.nanoTime();
+    long deadline = start + TimeUnit.MILLISECONDS.toNanos(REPLICATION_TIMEOUT_MILLIS);
+    while (true) {
       if (isReplicationCurrent.getAsBoolean()) {
-        LOG.info("Replication is caught up after " + i + " sleep attempts");
+        LOG.info("Replication caught up after " + elapsedMillis(start) + "ms");
         return;
       }
-
-      // Give time for replication to catch up
-      LOG.info("sleeping for " + sleepTimeInMillis);
-      Thread.sleep(sleepTimeInMillis);
-      // Exponential backoff
-      sleepTimeInMillis = sleepTimeInMillis * 2;
+      if (System.nanoTime() >= deadline) {
+        Assert.fail(
+            "Timed out waiting for replication to catch up after " + elapsedMillis(start) + "ms");
+      }
+      Thread.sleep(POLL_INTERVAL_MILLIS);
     }
+  }
 
-    Assert.fail("Timed out waiting for replication to catch up.");
+  private static long elapsedMillis(long startNanos) {
+    return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+  }
+
+  /**
+   * Waits until {@code replicatedEntries} stops advancing, draining entries left in flight by an
+   * earlier test.
+   *
+   * <p>The replication endpoint's entry counter is static and shared by every test in a class.
+   * Resetting it between tests is not enough on its own: replication is asynchronous, so entries
+   * belonging to the previous test keep arriving after the reset and count towards the next test's
+   * target. Call this before resetting the counter so it starts from a genuinely idle state.
+   *
+   * <p>This is best effort and does not fail on timeout; callers still verify the data itself.
+   */
+  public static void waitForReplicationToQuiesce(IntSupplier replicatedEntries)
+      throws InterruptedException {
+    long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(REPLICATION_TIMEOUT_MILLIS);
+    int previous = -1;
+    int stablePolls = 0;
+    while (stablePolls < QUIESCE_STABLE_POLLS && System.nanoTime() < deadline) {
+      int current = replicatedEntries.getAsInt();
+      stablePolls = (current == previous) ? stablePolls + 1 : 0;
+      previous = current;
+      Thread.sleep(POLL_INTERVAL_MILLIS);
+    }
   }
 
   /**
@@ -236,7 +272,22 @@ public static void assertTableEventuallyEquals(
       Table expected, Table actual, BooleanSupplier isReplicationCurrent)
       throws InterruptedException, IOException {
     waitForReplication(isReplicationCurrent);
-    TestUtils.assertTableEquals(expected, actual);
+
+    // The entry counter the predicate reads is only a coarse progress signal: it is shared across
+    // tests and counts entries for every table, so it can be satisfied before this table's own
+    // writes land. Retry the comparison so the assertion waits on the replicated data itself.
+    long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(REPLICATION_TIMEOUT_MILLIS);
+    while (true) {
+      try {
+        TestUtils.assertTableEquals(expected, actual);
+        return;
+      } catch (AssertionError e) {
+        if (System.nanoTime() >= deadline) {
+          throw e;
+        }
+      }
+      Thread.sleep(POLL_INTERVAL_MILLIS);
+    }
   }
 
   /**