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

.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);
+    }
   }
 
   /**

pom.xml

@@ -46,7 +46,6 @@ limitations under the License.
     <module>bigtable-client-core-parent</module>
     <module>bigtable-hbase-1.x-parent</module>
     <module>bigtable-hbase-2.x-parent</module>
-    <module>bigtable-dataflow-parent</module>
     <module>bigtable-test</module>
     <module>hbase-migration-tools</module>
   </modules>
@@ -55,16 +54,16 @@ limitations under the License.
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 
     <!-- core dependency versions -->
-    <bigtable.version>2.81.0</bigtable.version>
-    <google-cloud-bigtable-emulator.version>0.218.0</google-cloud-bigtable-emulator.version>
+    <bigtable.version>2.85.0</bigtable.version>
+    <google-cloud-bigtable-emulator.version>0.222.0</google-cloud-bigtable-emulator.version>
     <bigtable-metrics-api.version>1.29.2</bigtable-metrics-api.version>
     <!-- Optional dep for bigtable-metrics-api used for tests -->
     <dropwizard-metrics.version>4.2.22</dropwizard-metrics.version>
 
     <!-- shared library versions pulled in by the bigtable veneer; pinned to satisfy
          RequireUpperBoundDeps (slf4j) and to match the google-cloud-bigtable deps-bom (conscrypt) -->
     <slf4j.version>2.0.16</slf4j.version>
-    <conscrypt.version>2.6.0</conscrypt.version>
+    <conscrypt.version>2.6.2</conscrypt.version>
 
     <!--  keeping at this version to align with hbase-->
     <reload4j.version>1.2.24</reload4j.version>
@@ -85,7 +84,7 @@ limitations under the License.
     <truth.version>1.1.5</truth.version>
     <hamcrest.version>1.3</hamcrest.version>
     <mockito.version>4.11.0</mockito.version>
-    <beam.version>2.72.0</beam.version>
+    <beam.version>2.77.0</beam.version>
     <!-- referred from bigtable-beam-import and bigtable-emulator -->
     <guava.version>33.4.0-jre</guava.version>
     <opencensus.version>0.31.1</opencensus.version>
@@ -94,8 +93,8 @@ limitations under the License.
          while still respecting global skipTests override. -->
     <skipTests>false</skipTests>
     <skipUnitTests>${skipTests}</skipUnitTests>
-    <maven.compiler.source>1.8</maven.compiler.source>
-    <maven.compiler.target>1.8</maven.compiler.target>
+    <maven.compiler.source>11</maven.compiler.source>
+    <maven.compiler.target>11</maven.compiler.target>
   </properties>
 
   <dependencyManagement>
@@ -316,6 +315,16 @@ limitations under the License.
     </plugins>
   </build>
   <profiles>
+    <profile>
+      <!-- beam-sdks-java-core 2.61+ requires Java 11; exclude beam modules from Java 8 builds -->
+      <id>beam</id>
+      <activation>
+        <jdk>[11,)</jdk>
+      </activation>
+      <modules>
+        <module>bigtable-dataflow-parent</module>
+      </modules>
+    </profile>
     <profile>
       <id>useBigtableSnapshots</id>
       <!-- use the snapshot veneer artifact to run the tests-->