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 2020/06/15 12:47:40 UTC
[GitHub] [beam] nielm opened a new pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
nielm opened a new pull request #12010:
URL: https://github.com/apache/beam/pull/12010
In a multi-threaded runner, WriteToSpannerFn.setup() is called for
each instance on each thread, causing multiple session pools to be
created.
Use a ref-counted SpannerAccessor per database to ensure that there
is only one session pool per database per process.
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Apex | Dataflow | Flink | Samza | Spark
--- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/)
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/)
XLang | --- | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/)
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://builds.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://builds.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-648956095
Retest this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj merged pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
chamikaramj merged pull request #12010:
URL: https://github.com/apache/beam/pull/12010
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] allenpradeep commented on a change in pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
allenpradeep commented on a change in pull request #12010:
URL: https://github.com/apache/beam/pull/12010#discussion_r445110010
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerAccessor.java
##########
@@ -31,33 +31,73 @@
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.util.ReleaseInfo;
import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** Manages lifecycle of {@link DatabaseClient} and {@link Spanner} instances. */
class SpannerAccessor implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(SpannerAccessor.class);
+
// A common user agent token that indicates that this request was originated from Apache Beam.
private static final String USER_AGENT_PREFIX = "Apache_Beam_Java";
+ // Only create one SpannerAccessor for each different SpannerConfig.
+ private static final ConcurrentHashMap<SpannerConfig, SpannerAccessor> spannerAccessors =
+ new ConcurrentHashMap<>();
+
+ // Keep reference counts of each SpannerAccessor's usage so that we can close
+ // it when it is no longer in use.
+ private static final ConcurrentHashMap<SpannerConfig, AtomicInteger> refcounts =
+ new ConcurrentHashMap<>();
+
private final Spanner spanner;
private final DatabaseClient databaseClient;
private final BatchClient batchClient;
private final DatabaseAdminClient databaseAdminClient;
+ private final SpannerConfig spannerConfig;
private SpannerAccessor(
Spanner spanner,
DatabaseClient databaseClient,
DatabaseAdminClient databaseAdminClient,
- BatchClient batchClient) {
+ BatchClient batchClient,
+ SpannerConfig spannerConfig) {
this.spanner = spanner;
this.databaseClient = databaseClient;
this.databaseAdminClient = databaseAdminClient;
this.batchClient = batchClient;
+ this.spannerConfig = spannerConfig;
}
- static SpannerAccessor create(SpannerConfig spannerConfig) {
Review comment:
NIT: Should we maintain this function for compatibility sake? There might be external customers who may be using create call in their template. Would they be affected when they start using this code.
Internally, this can call getOrCreate()
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-650532755
> The added `SpannerAccessorTest` (at line 20) seems to be breaking `beam_PreCommit_SQL_Java11_Commit`. For example, see: https://ci-beam.apache.org/job/beam_PreCommit_SQL_Java11_Commit/178/java/new/
Thanks - no idea how this passed tests before! PR #12111 created.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-649895281
Run Java PreCommit
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm commented on a change in pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm commented on a change in pull request #12010:
URL: https://github.com/apache/beam/pull/12010#discussion_r445045164
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
##########
@@ -1373,10 +1375,22 @@ public void processElement(ProcessContext c) {
@VisibleForTesting
static class WriteToSpannerFn extends DoFn<Iterable<MutationGroup>, Void> {
- private transient SpannerAccessor spannerAccessor;
private final SpannerConfig spannerConfig;
private final FailureMode failureMode;
+ // Only create one SpannerAccessor for each different SpannerConfig.
Review comment:
Good call -- moved to SpannerAccessor.
To make it clearer, SpannerAccessor.create(SpannerConfig) has been renamed to getOrCreate(SpannerConfig).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robinyqiu commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
robinyqiu commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-650446658
The added `SpannerAccessorTest` seems to be breaking `beam_PreCommit_SQL_Java11_Commit`. For example, see: https://ci-beam.apache.org/job/beam_PreCommit_SQL_Java11_Commit/178/java/new/
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm edited a comment on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm edited a comment on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-648956095
Retest this please
(implemented Allens comments, and fixed tests)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm commented on a change in pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm commented on a change in pull request #12010:
URL: https://github.com/apache/beam/pull/12010#discussion_r445045365
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
##########
@@ -1423,18 +1438,46 @@ public void processElement(ProcessContext c) {
}
@Setup
- public void setup() throws Exception {
- // set up non-serializable values here.
- spannerAccessor = SpannerAccessor.create(spannerConfig);
+ public void setup() {
+ spannerAccessor = spannerAccessors.get(spannerConfig);
+ if (spannerAccessor == null) {
+ synchronized (spannerAccessors) {
Review comment:
as above, moved refcounting into SpannerAccessor.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-648516228
Retest this please
I'll wait for Allen's review here. Thanks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robinyqiu edited a comment on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
robinyqiu edited a comment on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-650446658
The added `SpannerAccessorTest` (at line 20) seems to be breaking `beam_PreCommit_SQL_Java11_Commit`. For example, see: https://ci-beam.apache.org/job/beam_PreCommit_SQL_Java11_Commit/178/java/new/
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-649008321
Retest this please
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] allenpradeep commented on a change in pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
allenpradeep commented on a change in pull request #12010:
URL: https://github.com/apache/beam/pull/12010#discussion_r441173677
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
##########
@@ -1373,10 +1375,22 @@ public void processElement(ProcessContext c) {
@VisibleForTesting
static class WriteToSpannerFn extends DoFn<Iterable<MutationGroup>, Void> {
- private transient SpannerAccessor spannerAccessor;
private final SpannerConfig spannerConfig;
private final FailureMode failureMode;
+ // Only create one SpannerAccessor for each different SpannerConfig.
Review comment:
Setup for SpannerAccessor are done from BatchSpannerRead, CreateTransactionFn, NaiveSpannerRead, ReadSpannerSchema.
Should this go into SpannerAccessor directly and when create() is called, should that take care of creating if absent rather than SpannerIO? let me know your thoughts.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
##########
@@ -1423,18 +1438,46 @@ public void processElement(ProcessContext c) {
}
@Setup
- public void setup() throws Exception {
- // set up non-serializable values here.
- spannerAccessor = SpannerAccessor.create(spannerConfig);
+ public void setup() {
+ spannerAccessor = spannerAccessors.get(spannerConfig);
+ if (spannerAccessor == null) {
+ synchronized (spannerAccessors) {
Review comment:
I guess this could could be common code in SpannerAccessor(may be part of SpannerAccessor.create()) and could be shared as SpannerAccessor is called from other places as well.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm commented on pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm commented on pull request #12010:
URL: https://github.com/apache/beam/pull/12010#issuecomment-644113718
R: @allenpradeep
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] nielm commented on a change in pull request #12010: [BEAM-10259] Use ref-counted connection to Spanner to prevent multiple connections.
Posted by GitBox <gi...@apache.org>.
nielm commented on a change in pull request #12010:
URL: https://github.com/apache/beam/pull/12010#discussion_r445436926
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerAccessor.java
##########
@@ -31,33 +31,73 @@
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.beam.sdk.options.ValueProvider;
import org.apache.beam.sdk.util.ReleaseInfo;
import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/** Manages lifecycle of {@link DatabaseClient} and {@link Spanner} instances. */
class SpannerAccessor implements AutoCloseable {
+ private static final Logger LOG = LoggerFactory.getLogger(SpannerAccessor.class);
+
// A common user agent token that indicates that this request was originated from Apache Beam.
private static final String USER_AGENT_PREFIX = "Apache_Beam_Java";
+ // Only create one SpannerAccessor for each different SpannerConfig.
+ private static final ConcurrentHashMap<SpannerConfig, SpannerAccessor> spannerAccessors =
+ new ConcurrentHashMap<>();
+
+ // Keep reference counts of each SpannerAccessor's usage so that we can close
+ // it when it is no longer in use.
+ private static final ConcurrentHashMap<SpannerConfig, AtomicInteger> refcounts =
+ new ConcurrentHashMap<>();
+
private final Spanner spanner;
private final DatabaseClient databaseClient;
private final BatchClient batchClient;
private final DatabaseAdminClient databaseAdminClient;
+ private final SpannerConfig spannerConfig;
private SpannerAccessor(
Spanner spanner,
DatabaseClient databaseClient,
DatabaseAdminClient databaseAdminClient,
- BatchClient batchClient) {
+ BatchClient batchClient,
+ SpannerConfig spannerConfig) {
this.spanner = spanner;
this.databaseClient = databaseClient;
this.databaseAdminClient = databaseAdminClient;
this.batchClient = batchClient;
+ this.spannerConfig = spannerConfig;
}
- static SpannerAccessor create(SpannerConfig spannerConfig) {
Review comment:
Its a package-private class -- noone else will be using it.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org