diff --git a/.kokoro/continuous/java8.cfg b/.kokoro/continuous/java8.cfg index 495cc7bacd..2ebfeba560 100644 --- a/.kokoro/continuous/java8.cfg +++ b/.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: { diff --git a/.kokoro/nightly/java8.cfg b/.kokoro/nightly/java8.cfg index 495cc7bacd..2ebfeba560 100644 --- a/.kokoro/nightly/java8.cfg +++ b/.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: { diff --git a/.kokoro/presubmit/java8.cfg b/.kokoro/presubmit/java8.cfg index 495cc7bacd..2ebfeba560 100644 --- a/.kokoro/presubmit/java8.cfg +++ b/.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: { diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/TestBufferedMutator.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/TestBufferedMutator.java index 97b92d618c..f25bf886d8 100644 --- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/TestBufferedMutator.java +++ b/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. + * + *

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); diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml index 9490148b45..eeb563a5a3 100644 --- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml +++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml @@ -389,6 +389,10 @@ limitations under the License. org.jspecify.annotations com.google.bigtable.repackaged.org.jspecify.annotations + + dev.cel + com.google.bigtable.repackaged.dev.cel + diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestModifyTable.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestModifyTable.java index 605c078647..c19f8fdf1e 100644 --- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/TestModifyTable.java +++ b/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); } diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml index 229e871eff..40ed8346ba 100644 --- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml +++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml @@ -380,6 +380,10 @@ limitations under the License. org.jspecify.annotations com.google.bigtable.repackaged.org.jspecify.annotations + + dev.cel + com.google.bigtable.repackaged.dev.cel + diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java index fad550fa40..5b76a8e22c 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java +++ b/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()); diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java index 3ec3bd728f..6688c5cca9 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java +++ b/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()); diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/utils/TestUtils.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/utils/TestUtils.java index 0355b2f952..bf97b95202 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/utils/TestUtils.java +++ b/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 expectedIterator = expectedScanner.iterator(); - Iterator actualIterator = actualScanner.iterator(); + Iterator expectedIterator = expectedScanner.iterator(); + Iterator 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. + * + *

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. + * + *

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); + } } /** diff --git a/pom.xml b/pom.xml index a8def01761..58c8fe87d1 100644 --- a/pom.xml +++ b/pom.xml @@ -46,7 +46,6 @@ limitations under the License. bigtable-client-core-parent bigtable-hbase-1.x-parent bigtable-hbase-2.x-parent - bigtable-dataflow-parent bigtable-test hbase-migration-tools @@ -55,8 +54,8 @@ limitations under the License. UTF-8 - 2.81.0 - 0.218.0 + 2.85.0 + 0.222.0 1.29.2 4.2.22 @@ -64,7 +63,7 @@ limitations under the License. 2.0.16 - 2.6.0 + 2.6.2 1.2.24 @@ -85,7 +84,7 @@ limitations under the License. 1.1.5 1.3 4.11.0 - 2.72.0 + 2.77.0 33.4.0-jre 0.31.1 @@ -94,8 +93,8 @@ limitations under the License. while still respecting global skipTests override. --> false ${skipTests} - 1.8 - 1.8 + 11 + 11 @@ -316,6 +315,16 @@ limitations under the License. + + + beam + + [11,) + + + bigtable-dataflow-parent + + useBigtableSnapshots