You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/02/16 23:31:02 UTC
[GitHub] [beam] nancyxu123 opened a new pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
nancyxu123 opened a new pull request #16876:
URL: https://github.com/apache/beam/pull/16876
*Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] thiagotnunes commented on a change in pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
thiagotnunes commented on a change in pull request #16876:
URL: https://github.com/apache/beam/pull/16876#discussion_r810724553
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.java
##########
@@ -0,0 +1,659 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.it;
+
+import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.Key;
+import com.google.cloud.spanner.KeySet;
+import com.google.cloud.spanner.Mutation;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.DataChangeRecord;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
+import org.apache.beam.sdk.state.BagState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end test of Cloud Spanner Change Streams with strict commit timestamp and transaction
+ * ordering.
+ */
+@RunWith(JUnit4.class)
+public class SpannerChangeStreamOrderedByTimestampAndTransactionIdIT {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.class);
+
+ @ClassRule public static final IntegrationTestEnv ENV = new IntegrationTestEnv();
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static String projectId;
+ private static String instanceId;
+ private static String databaseId;
+ private static String tableName;
+ private static String changeStreamName;
+ private static DatabaseClient databaseClient;
+
+ @BeforeClass
+ public static void setup() throws InterruptedException, ExecutionException, TimeoutException {
+ projectId = ENV.getProjectId();
+ instanceId = ENV.getInstanceId();
+ databaseId = ENV.getDatabaseId();
+ tableName = ENV.createSingersTable();
+ changeStreamName = ENV.createChangeStreamFor(tableName);
+ databaseClient = ENV.getDatabaseClient();
+ }
+
+ @Test
+ public void testTransactionBoundaries() {
+ final SpannerConfig spannerConfig =
+ SpannerConfig.create()
+ .withProjectId(projectId)
+ .withInstanceId(instanceId)
+ .withDatabaseId(databaseId);
+ // Commit a initial transaction to get the timestamp to start reading from.
+ List<Mutation> mutations = new ArrayList<>();
+ mutations.add(insertRecordMutation(0, "FirstName0", "LastName0"));
+ final long timeIncrementInSeconds = 5;
Review comment:
Could we reduce this time or even better not have it? If not, could we write some code to check for a condition in a loop and sleep on smaller intervals until a timeout is reached?
I ask this, because introducing sleeps adds an artificial lower bound to how fast this test can be, which penalizes the whole integration test suite performance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #16876:
URL: https://github.com/apache/beam/pull/16876#issuecomment-1051089073
Run Java postcommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nancyxu123 commented on a change in pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
nancyxu123 commented on a change in pull request #16876:
URL: https://github.com/apache/beam/pull/16876#discussion_r813443755
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/ChangeStreamTestPipelineOptions.java
##########
@@ -31,7 +31,7 @@
void setProjectId(String value);
@Description("Instance ID to write to in Spanner")
- @Default.String("beam-test")
+ @Default.String("changestream")
Review comment:
Ah, I did not mean to change this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nancyxu123 commented on a change in pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
nancyxu123 commented on a change in pull request #16876:
URL: https://github.com/apache/beam/pull/16876#discussion_r813355701
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.java
##########
@@ -0,0 +1,659 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.it;
+
+import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.Key;
+import com.google.cloud.spanner.KeySet;
+import com.google.cloud.spanner.Mutation;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.DataChangeRecord;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
+import org.apache.beam.sdk.state.BagState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end test of Cloud Spanner Change Streams with strict commit timestamp and transaction
+ * ordering.
+ */
+@RunWith(JUnit4.class)
+public class SpannerChangeStreamOrderedByTimestampAndTransactionIdIT {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.class);
+
+ @ClassRule public static final IntegrationTestEnv ENV = new IntegrationTestEnv();
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static String projectId;
+ private static String instanceId;
+ private static String databaseId;
+ private static String tableName;
+ private static String changeStreamName;
+ private static DatabaseClient databaseClient;
+
+ @BeforeClass
+ public static void setup() throws InterruptedException, ExecutionException, TimeoutException {
+ projectId = ENV.getProjectId();
+ instanceId = ENV.getInstanceId();
+ databaseId = ENV.getDatabaseId();
+ tableName = ENV.createSingersTable();
+ changeStreamName = ENV.createChangeStreamFor(tableName);
+ databaseClient = ENV.getDatabaseClient();
+ }
+
+ @Test
+ public void testTransactionBoundaries() {
+ final SpannerConfig spannerConfig =
+ SpannerConfig.create()
+ .withProjectId(projectId)
+ .withInstanceId(instanceId)
+ .withDatabaseId(databaseId);
+ // Commit a initial transaction to get the timestamp to start reading from.
+ List<Mutation> mutations = new ArrayList<>();
+ mutations.add(insertRecordMutation(0, "FirstName0", "LastName0"));
+ final long timeIncrementInSeconds = 5;
Review comment:
Sure, reduced the sleep time to 2 seconds.
I think the sleep time is necessary, since i'm testing that the buffer flushes at advancing timestamps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] thiagotnunes commented on a change in pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
thiagotnunes commented on a change in pull request #16876:
URL: https://github.com/apache/beam/pull/16876#discussion_r813438960
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.java
##########
@@ -0,0 +1,581 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.it;
+
+import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.Key;
+import com.google.cloud.spanner.KeySet;
+import com.google.cloud.spanner.Mutation;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.DataChangeRecord;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
+import org.apache.beam.sdk.state.BagState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end test of Cloud Spanner Change Streams with strict commit timestamp and transaction
+ * ordering.
+ */
+@RunWith(JUnit4.class)
+public class SpannerChangeStreamOrderedByTimestampAndTransactionIdIT {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.class);
+
+ @ClassRule public static final IntegrationTestEnv ENV = new IntegrationTestEnv();
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static String projectId;
+ private static String instanceId;
+ private static String databaseId;
+ private static String tableName;
+ private static String changeStreamName;
+ private static DatabaseClient databaseClient;
+
+ @BeforeClass
+ public static void setup() throws InterruptedException, ExecutionException, TimeoutException {
+ projectId = ENV.getProjectId();
+ instanceId = ENV.getInstanceId();
+ databaseId = ENV.getDatabaseId();
+ tableName = ENV.createSingersTable();
+ changeStreamName = ENV.createChangeStreamFor(tableName);
+ databaseClient = ENV.getDatabaseClient();
+ }
+
+ @Test
+ public void testTransactionBoundaries() {
+ final SpannerConfig spannerConfig =
+ SpannerConfig.create()
+ .withProjectId(projectId)
+ .withInstanceId(instanceId)
+ .withDatabaseId(databaseId);
+ // Commit a initial transaction to get the timestamp to start reading from.
+ List<Mutation> mutations = new ArrayList<>();
+ mutations.add(insertRecordMutation(0, "FirstName0", "LastName0"));
+ final long timeIncrementInSeconds = 2;
+ final Timestamp startTimestamp = databaseClient.write(mutations);
+ writeTransactionsToDatabase();
+
+ // Sleep the time increment interval.
+ try {
+ Thread.sleep(timeIncrementInSeconds * 1000);
+ } catch (InterruptedException e) {
+ LOG.error(e.toString());
Review comment:
nit: `LOG.error(e.toString(), e)`
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/ChangeStreamTestPipelineOptions.java
##########
@@ -31,7 +31,7 @@
void setProjectId(String value);
@Description("Instance ID to write to in Spanner")
- @Default.String("beam-test")
+ @Default.String("changestream")
Review comment:
We can not change this, since the tests run in beam's instance
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamOrderedWithinKeyGloballyIT.java
##########
@@ -124,7 +123,7 @@ public void testOrderedWithinKey() {
try {
Thread.sleep(timeIncrementInSeconds * 1000);
} catch (InterruptedException e) {
- System.out.println(e);
+ LOG.error(e.toString());
Review comment:
nit: `LOG.error(e.toString(), e)`
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.java
##########
@@ -0,0 +1,581 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.spanner.changestreams.it;
+
+import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.Key;
+import com.google.cloud.spanner.KeySet;
+import com.google.cloud.spanner.Mutation;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.coders.BooleanCoder;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerConfig;
+import org.apache.beam.sdk.io.gcp.spanner.SpannerIO;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.DataChangeRecord;
+import org.apache.beam.sdk.io.gcp.spanner.changestreams.model.Mod;
+import org.apache.beam.sdk.state.BagState;
+import org.apache.beam.sdk.state.StateSpec;
+import org.apache.beam.sdk.state.StateSpecs;
+import org.apache.beam.sdk.state.TimeDomain;
+import org.apache.beam.sdk.state.Timer;
+import org.apache.beam.sdk.state.TimerSpec;
+import org.apache.beam.sdk.state.TimerSpecs;
+import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * End-to-end test of Cloud Spanner Change Streams with strict commit timestamp and transaction
+ * ordering.
+ */
+@RunWith(JUnit4.class)
+public class SpannerChangeStreamOrderedByTimestampAndTransactionIdIT {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(SpannerChangeStreamOrderedByTimestampAndTransactionIdIT.class);
+
+ @ClassRule public static final IntegrationTestEnv ENV = new IntegrationTestEnv();
+ @Rule public final transient TestPipeline pipeline = TestPipeline.create();
+
+ private static String projectId;
+ private static String instanceId;
+ private static String databaseId;
+ private static String tableName;
+ private static String changeStreamName;
+ private static DatabaseClient databaseClient;
+
+ @BeforeClass
+ public static void setup() throws InterruptedException, ExecutionException, TimeoutException {
+ projectId = ENV.getProjectId();
+ instanceId = ENV.getInstanceId();
+ databaseId = ENV.getDatabaseId();
+ tableName = ENV.createSingersTable();
+ changeStreamName = ENV.createChangeStreamFor(tableName);
+ databaseClient = ENV.getDatabaseClient();
+ }
+
+ @Test
+ public void testTransactionBoundaries() {
+ final SpannerConfig spannerConfig =
+ SpannerConfig.create()
+ .withProjectId(projectId)
+ .withInstanceId(instanceId)
+ .withDatabaseId(databaseId);
+ // Commit a initial transaction to get the timestamp to start reading from.
+ List<Mutation> mutations = new ArrayList<>();
+ mutations.add(insertRecordMutation(0, "FirstName0", "LastName0"));
+ final long timeIncrementInSeconds = 2;
+ final Timestamp startTimestamp = databaseClient.write(mutations);
+ writeTransactionsToDatabase();
+
+ // Sleep the time increment interval.
+ try {
+ Thread.sleep(timeIncrementInSeconds * 1000);
+ } catch (InterruptedException e) {
+ LOG.error(e.toString());
+ }
+
+ // This will be the second batch of transactions that will have strict timestamp ordering
+ // per key.
+ writeTransactionsToDatabase();
+
+ // Sleep the time increment interval.
+ try {
+ Thread.sleep(timeIncrementInSeconds * 1000);
+ } catch (InterruptedException e) {
+ LOG.error(e.toString());
Review comment:
nit: `LOG.error(e.toString(), e)`
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/it/SpannerChangeStreamOrderedWithinKeyGloballyIT.java
##########
@@ -113,7 +112,7 @@ public void testOrderedWithinKey() {
try {
Thread.sleep(timeIncrementInSeconds * 1000);
} catch (InterruptedException e) {
- System.out.println(e);
+ LOG.error(e.toString());
Review comment:
nit: `LOG.error(e.toString(), e)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nancyxu123 commented on pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
nancyxu123 commented on pull request #16876:
URL: https://github.com/apache/beam/pull/16876#issuecomment-1051071598
Hmm it seems maybe to be a flaky error where something went wrong with the database creation? I tried running it on my personal laptop, and it passes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #16876:
URL: https://github.com/apache/beam/pull/16876#issuecomment-1050607619
Run Java PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #16876:
URL: https://github.com/apache/beam/pull/16876#issuecomment-1051063001
Note there's one test failure: https://ci-beam.apache.org/job/beam_PostCommit_Java_PR/978/testReport/org.apache.beam.sdk.io.gcp.spanner.changestreams.it/
Let me know if you need some help to figure out what's going on:)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nancyxu123 commented on pull request #16876: [BEAM-12164] Added integration test for transaction boundaries and transaction ID ordering. Made small fixes in ordered by key integration test.
Posted by GitBox <gi...@apache.org>.
nancyxu123 commented on pull request #16876:
URL: https://github.com/apache/beam/pull/16876#issuecomment-1052825186
> Run Java postcommit
Hmm Java postcommit failed again. My personal run keeps passing. This is weird, since the new test has exactly the same set up as the currently passing tests (i.e. SpannerChangeStreamOrderedWithinKeyGloballyIT). Is there a way for me to run Java postcommit myself?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org