beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [1/4] beam git commit: Pre read api refactoring. Extract `SpannerConfig` and `AbstractSpannerFn`
Date Wed, 28 Jun 2017 03:16:50 GMT
Repository: beam
Updated Branches:
  refs/heads/master 0b19fb414 -> e5929bd13


Pre read api refactoring. Extract `SpannerConfig` and `AbstractSpannerFn`


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/454f1c42
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/454f1c42
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/454f1c42

Branch: refs/heads/master
Commit: 454f1c427353feeb858cdc62185ea3fced8d8a1f
Parents: 80c9263
Author: Mairbek Khadikov <mairbek@google.com>
Authored: Mon Jun 19 13:01:20 2017 -0700
Committer: Eugene Kirpichov <kirpichov@google.com>
Committed: Tue Jun 27 18:36:01 2017 -0700

----------------------------------------------------------------------
 .../sdk/io/gcp/spanner/AbstractSpannerFn.java   |  41 ++++
 .../beam/sdk/io/gcp/spanner/SpannerConfig.java  | 118 ++++++++++
 .../beam/sdk/io/gcp/spanner/SpannerIO.java      | 227 ++++---------------
 .../sdk/io/gcp/spanner/SpannerWriteGroupFn.java | 108 +++++++++
 .../beam/sdk/io/gcp/spanner/SpannerIOTest.java  |   8 +-
 5 files changed, 321 insertions(+), 181 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/454f1c42/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/AbstractSpannerFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/AbstractSpannerFn.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/AbstractSpannerFn.java
new file mode 100644
index 0000000..08f7fa9
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/AbstractSpannerFn.java
@@ -0,0 +1,41 @@
+package org.apache.beam.sdk.io.gcp.spanner;
+
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.Spanner;
+import com.google.cloud.spanner.SpannerOptions;
+import org.apache.beam.sdk.transforms.DoFn;
+
+/**
+ * Abstract {@link DoFn} that manages {@link Spanner} lifecycle. Use {@link
+ * AbstractSpannerFn#databaseClient} to access the Cloud Spanner database client.
+ */
+abstract class AbstractSpannerFn<InputT, OutputT> extends DoFn<InputT, OutputT>
{
+  private transient Spanner spanner;
+  private transient DatabaseClient databaseClient;
+
+  abstract SpannerConfig getSpannerConfig();
+
+  @Setup
+  public void setup() throws Exception {
+    SpannerConfig spannerConfig = getSpannerConfig();
+    SpannerOptions options = spannerConfig.buildSpannerOptions();
+    spanner = options.getService();
+    databaseClient = spanner.getDatabaseClient(DatabaseId
+        .of(options.getProjectId(), spannerConfig.getInstanceId().get(),
+            spannerConfig.getDatabaseId().get()));
+  }
+
+  @Teardown
+  public void teardown() throws Exception {
+    if (spanner == null) {
+      return;
+    }
+    spanner.close();
+    spanner = null;
+  }
+
+  protected DatabaseClient databaseClient() {
+    return databaseClient;
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/454f1c42/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
new file mode 100644
index 0000000..4cb8aa2
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerConfig.java
@@ -0,0 +1,118 @@
+package org.apache.beam.sdk.io.gcp.spanner;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.google.auto.value.AutoValue;
+import com.google.cloud.ServiceFactory;
+import com.google.cloud.spanner.Spanner;
+import com.google.cloud.spanner.SpannerOptions;
+import com.google.common.annotations.VisibleForTesting;
+import java.io.Serializable;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.ValueProvider;
+import org.apache.beam.sdk.transforms.display.DisplayData;
+
+/** Configuration for a Cloud Spanner client. */
+@AutoValue
+public abstract class SpannerConfig implements Serializable {
+
+  private static final long serialVersionUID = -5680874609304170301L;
+
+  @Nullable
+  abstract ValueProvider<String> getProjectId();
+
+  @Nullable
+  abstract ValueProvider<String> getInstanceId();
+
+  @Nullable
+  abstract ValueProvider<String> getDatabaseId();
+
+  @Nullable
+  @VisibleForTesting
+  abstract ServiceFactory<Spanner, SpannerOptions> getServiceFactory();
+
+  abstract Builder toBuilder();
+
+  SpannerOptions buildSpannerOptions() {
+    SpannerOptions.Builder builder = SpannerOptions.newBuilder();
+    if (getProjectId() != null) {
+      builder.setProjectId(getProjectId().get());
+    }
+    if (getServiceFactory() != null) {
+      builder.setServiceFactory(getServiceFactory());
+    }
+    return builder.build();
+  }
+
+  public static SpannerConfig create() {
+    return builder().build();
+  }
+
+  public static Builder builder() {
+    return new AutoValue_SpannerConfig.Builder();
+  }
+
+  public void validate(PipelineOptions options) {
+    checkNotNull(
+        getInstanceId(),
+        "SpannerIO.read() requires instance id to be set with withInstanceId method");
+    checkNotNull(
+        getDatabaseId(),
+        "SpannerIO.read() requires database id to be set with withDatabaseId method");
+  }
+
+  public void populateDisplayData(DisplayData.Builder builder) {
+    builder
+        .addIfNotNull(DisplayData.item("projectId", getProjectId()).withLabel("Output Project"))
+        .addIfNotNull(DisplayData.item("instanceId", getInstanceId()).withLabel("Output Instance"))
+        .addIfNotNull(DisplayData.item("databaseId", getDatabaseId()).withLabel("Output Database"));
+
+    if (getServiceFactory() != null) {
+      builder.addIfNotNull(
+          DisplayData.item("serviceFactory", getServiceFactory().getClass().getName())
+              .withLabel("Service Factory"));
+    }
+  }
+
+  /** Builder for {@link SpannerConfig}. */
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+
+    abstract Builder setProjectId(ValueProvider<String> projectId);
+
+    abstract Builder setInstanceId(ValueProvider<String> instanceId);
+
+    abstract Builder setDatabaseId(ValueProvider<String> databaseId);
+
+
+    abstract Builder setServiceFactory(ServiceFactory<Spanner, SpannerOptions> serviceFactory);
+
+    public abstract SpannerConfig build();
+  }
+
+  public SpannerConfig withProjectId(ValueProvider<String> projectId) {
+    return toBuilder().setProjectId(projectId).build();
+  }
+
+  public SpannerConfig withProjectId(String projectId) {
+    return withProjectId(ValueProvider.StaticValueProvider.of(projectId));
+  }
+
+  public SpannerConfig withInstanceId(ValueProvider<String> instanceId) {
+    return toBuilder().setInstanceId(instanceId).build();
+  }
+
+  public SpannerConfig withInstanceId(String instanceId) {
+    return withInstanceId(ValueProvider.StaticValueProvider.of(instanceId));
+  }
+
+  public SpannerConfig withDatabaseId(ValueProvider<String> databaseId) {
+    return toBuilder().setDatabaseId(databaseId).build();
+  }
+
+  public SpannerConfig withDatabaseId(String databaseId) {
+    return withDatabaseId(ValueProvider.StaticValueProvider.of(databaseId));
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/454f1c42/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
index 32bf1d0..791c7e7 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java
@@ -17,22 +17,13 @@
  */
 package org.apache.beam.sdk.io.gcp.spanner;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-
 import com.google.auto.value.AutoValue;
 import com.google.cloud.ServiceFactory;
-import com.google.cloud.ServiceOptions;
-import com.google.cloud.spanner.AbortedException;
-import com.google.cloud.spanner.DatabaseClient;
-import com.google.cloud.spanner.DatabaseId;
 import com.google.cloud.spanner.Mutation;
 import com.google.cloud.spanner.Spanner;
 import com.google.cloud.spanner.SpannerOptions;
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Iterables;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
+
 import javax.annotation.Nullable;
 
 import org.apache.beam.sdk.annotations.Experimental;
@@ -42,16 +33,8 @@ import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.display.DisplayData;
-import org.apache.beam.sdk.transforms.display.DisplayData.Builder;
-import org.apache.beam.sdk.util.BackOff;
-import org.apache.beam.sdk.util.BackOffUtils;
-import org.apache.beam.sdk.util.FluentBackoff;
-import org.apache.beam.sdk.util.Sleeper;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PDone;
-import org.joda.time.Duration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Experimental {@link PTransform Transforms} for reading from and writing to <a
@@ -123,37 +106,23 @@ public class SpannerIO {
   @AutoValue
   public abstract static class Write extends PTransform<PCollection<Mutation>, PDone>
{
 
-    @Nullable
-    abstract ValueProvider<String> getProjectId();
+    private static final long serialVersionUID = 1920175411827980145L;
 
-    @Nullable
-    abstract ValueProvider<String> getInstanceId();
-
-    @Nullable
-    abstract ValueProvider<String> getDatabaseId();
+    abstract SpannerConfig getSpannerConfig();
 
     abstract long getBatchSizeBytes();
 
-    @Nullable
-    @VisibleForTesting
-    abstract ServiceFactory<Spanner, SpannerOptions> getServiceFactory();
-
     abstract Builder toBuilder();
 
     @AutoValue.Builder
     abstract static class Builder {
 
-      abstract Builder setProjectId(ValueProvider<String> projectId);
-
-      abstract Builder setInstanceId(ValueProvider<String> instanceId);
+      abstract Builder setSpannerConfig(SpannerConfig spannerConfig);
 
-      abstract Builder setDatabaseId(ValueProvider<String> databaseId);
+      abstract SpannerConfig.Builder spannerConfigBuilder();
 
       abstract Builder setBatchSizeBytes(long batchSizeBytes);
 
-      @VisibleForTesting
-      abstract Builder setServiceFactory(ServiceFactory<Spanner, SpannerOptions> serviceFactory);
-
       abstract Write build();
     }
 
@@ -166,8 +135,15 @@ public class SpannerIO {
       return withProjectId(ValueProvider.StaticValueProvider.of(projectId));
     }
 
+    /**
+     * Returns a new {@link SpannerIO.Write} that will write to the specified Cloud Spanner
project.
+     *
+     * <p>Does not modify this object.
+     */
     public Write withProjectId(ValueProvider<String> projectId) {
-      return toBuilder().setProjectId(projectId).build();
+      Write.Builder builder = toBuilder();
+      builder.spannerConfigBuilder().setProjectId(projectId);
+      return builder.build();
     }
 
     /**
@@ -180,11 +156,30 @@ public class SpannerIO {
       return withInstanceId(ValueProvider.StaticValueProvider.of(instanceId));
     }
 
+    /**
+     * Returns a new {@link SpannerIO.Write} that will write to the specified Cloud Spanner
+     * instance.
+     *
+     * <p>Does not modify this object.
+     */
     public Write withInstanceId(ValueProvider<String> instanceId) {
-      return toBuilder().setInstanceId(instanceId).build();
+      Write.Builder builder = toBuilder();
+      builder.spannerConfigBuilder().setInstanceId(instanceId);
+      return builder.build();
     }
 
     /**
+     * Returns a new {@link SpannerIO.Write} that will write to the specified Cloud Spanner
+     * config.
+     *
+     * <p>Does not modify this object.
+     */
+    public Write withSpannerConfig(SpannerConfig spannerConfig) {
+      return toBuilder().setSpannerConfig(spannerConfig).build();
+    }
+
+
+    /**
      * Returns a new {@link SpannerIO.Write} with a new batch size limit.
      *
      * <p>Does not modify this object.
@@ -203,8 +198,16 @@ public class SpannerIO {
       return withDatabaseId(ValueProvider.StaticValueProvider.of(databaseId));
     }
 
+    /**
+     * Returns a new {@link SpannerIO.Write} that will write to the specified Cloud Spanner
+     * database.
+     *
+     * <p>Does not modify this object.
+     */
     public Write withDatabaseId(ValueProvider<String> databaseId) {
-      return toBuilder().setDatabaseId(databaseId).build();
+      Write.Builder builder = toBuilder();
+      builder.spannerConfigBuilder().setDatabaseId(databaseId);
+      return builder.build();
     }
 
     /**
@@ -216,17 +219,14 @@ public class SpannerIO {
 
     @VisibleForTesting
     Write withServiceFactory(ServiceFactory<Spanner, SpannerOptions> serviceFactory)
{
-      return toBuilder().setServiceFactory(serviceFactory).build();
+      Write.Builder builder = toBuilder();
+      builder.spannerConfigBuilder().setServiceFactory(serviceFactory);
+      return builder.build();
     }
 
     @Override
     public void validate(PipelineOptions options) {
-      checkNotNull(
-          getInstanceId(),
-          "SpannerIO.write() requires instance id to be set with withInstanceId method");
-      checkNotNull(
-          getDatabaseId(),
-          "SpannerIO.write() requires database id to be set with withDatabaseId method");
+      getSpannerConfig().validate(options);
     }
 
     @Override
@@ -237,22 +237,13 @@ public class SpannerIO {
       return PDone.in(input.getPipeline());
     }
 
+
     @Override
     public void populateDisplayData(DisplayData.Builder builder) {
       super.populateDisplayData(builder);
-      builder
-          .addIfNotNull(DisplayData.item("projectId", getProjectId()).withLabel("Output Project"))
-          .addIfNotNull(
-              DisplayData.item("instanceId", getInstanceId()).withLabel("Output Instance"))
-          .addIfNotNull(
-              DisplayData.item("databaseId", getDatabaseId()).withLabel("Output Database"))
-          .add(DisplayData.item("batchSizeBytes", getBatchSizeBytes())
-              .withLabel("Batch Size in Bytes"));
-      if (getServiceFactory() != null) {
-        builder.addIfNotNull(
-            DisplayData.item("serviceFactory", getServiceFactory().getClass().getName())
-                .withLabel("Service Factory"));
-      }
+      getSpannerConfig().populateDisplayData(builder);
+      builder.add(
+          DisplayData.item("batchSizeBytes", getBatchSizeBytes()).withLabel("Batch Size in
Bytes"));
     }
   }
 
@@ -278,123 +269,5 @@ public class SpannerIO {
     }
   }
 
-  /** Batches together and writes mutations to Google Cloud Spanner. */
-  @VisibleForTesting
-  static class SpannerWriteGroupFn extends DoFn<MutationGroup, Void> {
-    private static final Logger LOG = LoggerFactory.getLogger(SpannerWriteGroupFn.class);
-    private final Write spec;
-    private transient Spanner spanner;
-    private transient DatabaseClient dbClient;
-    // Current batch of mutations to be written.
-    private List<MutationGroup> mutations;
-    private long batchSizeBytes = 0;
-
-    private static final int MAX_RETRIES = 5;
-    private static final FluentBackoff BUNDLE_WRITE_BACKOFF =
-        FluentBackoff.DEFAULT
-            .withMaxRetries(MAX_RETRIES)
-            .withInitialBackoff(Duration.standardSeconds(5));
-
-    @VisibleForTesting SpannerWriteGroupFn(Write spec) {
-      this.spec = spec;
-    }
-
-    @Setup
-    public void setup() throws Exception {
-      SpannerOptions spannerOptions = getSpannerOptions();
-      spanner = spannerOptions.getService();
-      dbClient = spanner.getDatabaseClient(
-          DatabaseId.of(projectId(), spec.getInstanceId().get(), spec.getDatabaseId().get()));
-      mutations = new ArrayList<>();
-      batchSizeBytes = 0;
-    }
-
-    @ProcessElement
-    public void processElement(ProcessContext c) throws Exception {
-      MutationGroup m = c.element();
-      mutations.add(m);
-      batchSizeBytes += MutationSizeEstimator.sizeOf(m);
-      if (batchSizeBytes >= spec.getBatchSizeBytes()) {
-        flushBatch();
-      }
-    }
-
-    private String projectId() {
-      return spec.getProjectId() == null
-          ? ServiceOptions.getDefaultProjectId()
-          : spec.getProjectId().get();
-    }
-
-    @FinishBundle
-    public void finishBundle() throws Exception {
-      if (!mutations.isEmpty()) {
-        flushBatch();
-      }
-    }
-
-    @Teardown
-    public void teardown() throws Exception {
-      if (spanner == null) {
-        return;
-      }
-      spanner.close();
-      spanner = null;
-    }
-
-    private SpannerOptions getSpannerOptions() {
-      SpannerOptions.Builder spannerOptionsBuider = SpannerOptions.newBuilder();
-      if (spec.getServiceFactory() != null) {
-        spannerOptionsBuider.setServiceFactory(spec.getServiceFactory());
-      }
-      if (spec.getProjectId() != null) {
-        spannerOptionsBuider.setProjectId(spec.getProjectId().get());
-      }
-      return spannerOptionsBuider.build();
-    }
-
-    /**
-     * Writes a batch of mutations to Cloud Spanner.
-     *
-     * <p>If a commit fails, it will be retried up to {@link #MAX_RETRIES} times. If
the retry limit
-     * is exceeded, the last exception from Cloud Spanner will be thrown.
-     *
-     * @throws AbortedException if the commit fails or IOException or InterruptedException
if
-     *     backing off between retries fails.
-     */
-    private void flushBatch() throws AbortedException, IOException, InterruptedException
{
-      LOG.debug("Writing batch of {} mutations", mutations.size());
-      Sleeper sleeper = Sleeper.DEFAULT;
-      BackOff backoff = BUNDLE_WRITE_BACKOFF.backoff();
-
-      while (true) {
-        // Batch upsert rows.
-        try {
-          dbClient.writeAtLeastOnce(Iterables.concat(mutations));
-
-          // Break if the commit threw no exception.
-          break;
-        } catch (AbortedException exception) {
-          // Only log the code and message for potentially-transient errors. The entire exception
-          // will be propagated upon the last retry.
-          LOG.error(
-              "Error writing to Spanner ({}): {}", exception.getCode(), exception.getMessage());
-          if (!BackOffUtils.next(sleeper, backoff)) {
-            LOG.error("Aborting after {} retries.", MAX_RETRIES);
-            throw exception;
-          }
-        }
-      }
-      LOG.debug("Successfully wrote {} mutations", mutations.size());
-      mutations = new ArrayList<>();
-      batchSizeBytes = 0;
-    }
-
-    @Override
-    public void populateDisplayData(Builder builder) {
-      super.populateDisplayData(builder);
-      spec.populateDisplayData(builder);
-    }
-  }
-
   private SpannerIO() {} // Prevent construction.
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/454f1c42/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerWriteGroupFn.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerWriteGroupFn.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerWriteGroupFn.java
new file mode 100644
index 0000000..aed4832
--- /dev/null
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerWriteGroupFn.java
@@ -0,0 +1,108 @@
+package org.apache.beam.sdk.io.gcp.spanner;
+
+import com.google.cloud.spanner.AbortedException;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Iterables;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.beam.sdk.transforms.display.DisplayData;
+import org.apache.beam.sdk.util.BackOff;
+import org.apache.beam.sdk.util.BackOffUtils;
+import org.apache.beam.sdk.util.FluentBackoff;
+import org.apache.beam.sdk.util.Sleeper;
+import org.joda.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Batches together and writes mutations to Google Cloud Spanner. */
+@VisibleForTesting class SpannerWriteGroupFn extends AbstractSpannerFn<MutationGroup,
Void> {
+  private static final Logger LOG = LoggerFactory.getLogger(SpannerWriteGroupFn.class);
+  private final SpannerIO.Write spec;
+  // Current batch of mutations to be written.
+  private List<MutationGroup> mutations;
+  private long batchSizeBytes = 0;
+
+  private static final int MAX_RETRIES = 5;
+  private static final FluentBackoff BUNDLE_WRITE_BACKOFF =
+      FluentBackoff.DEFAULT
+          .withMaxRetries(MAX_RETRIES)
+          .withInitialBackoff(Duration.standardSeconds(5));
+
+  @VisibleForTesting SpannerWriteGroupFn(SpannerIO.Write spec) {
+    this.spec = spec;
+  }
+
+  @Override SpannerConfig getSpannerConfig() {
+    return spec.getSpannerConfig();
+  }
+
+  @Setup
+  public void setup() throws Exception {
+    super.setup();
+    mutations = new ArrayList<>();
+    batchSizeBytes = 0;
+  }
+
+  @ProcessElement
+  public void processElement(ProcessContext c) throws Exception {
+    MutationGroup m = c.element();
+    mutations.add(m);
+    batchSizeBytes += MutationSizeEstimator.sizeOf(m);
+    if (batchSizeBytes >= spec.getBatchSizeBytes()) {
+      flushBatch();
+    }
+  }
+
+  @FinishBundle
+  public void finishBundle() throws Exception {
+    if (!mutations.isEmpty()) {
+      flushBatch();
+    }
+  }
+
+  /**
+   * Writes a batch of mutations to Cloud Spanner.
+   *
+   * <p>If a commit fails, it will be retried up to {@link #MAX_RETRIES} times. If
the retry limit
+   * is exceeded, the last exception from Cloud Spanner will be thrown.
+   *
+   * @throws AbortedException if the commit fails or IOException or InterruptedException
if
+   *     backing off between retries fails.
+   */
+  private void flushBatch() throws AbortedException, IOException, InterruptedException {
+    LOG.debug("Writing batch of {} mutations", mutations.size());
+    Sleeper sleeper = Sleeper.DEFAULT;
+    BackOff backoff = BUNDLE_WRITE_BACKOFF.backoff();
+
+    while (true) {
+      // Batch upsert rows.
+      try {
+        databaseClient().writeAtLeastOnce(Iterables.concat(mutations));
+
+        // Break if the commit threw no exception.
+        break;
+      } catch (AbortedException exception) {
+        // Only log the code and message for potentially-transient errors. The entire exception
+        // will be propagated upon the last retry.
+        LOG.error(
+            "Error writing to Spanner ({}): {}", exception.getCode(), exception.getMessage());
+        if (!BackOffUtils.next(sleeper, backoff)) {
+          LOG.error("Aborting after {} retries.", MAX_RETRIES);
+          throw exception;
+        }
+      }
+    }
+    LOG.debug("Successfully wrote {} mutations", mutations.size());
+    mutations = new ArrayList<>();
+    batchSizeBytes = 0;
+  }
+
+  @Override
+  public void populateDisplayData(DisplayData.Builder builder) {
+    super.populateDisplayData(builder);
+    spec.populateDisplayData(builder);
+  }
+}

http://git-wip-us.apache.org/repos/asf/beam/blob/454f1c42/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOTest.java
index 0cc08bf..abeac0a 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIOTest.java
@@ -149,7 +149,7 @@ public class SpannerIOTest implements Serializable {
             .withDatabaseId("test-database")
             .withBatchSizeBytes(1000000000)
             .withServiceFactory(serviceFactory);
-    SpannerIO.SpannerWriteGroupFn writerFn = new SpannerIO.SpannerWriteGroupFn(write);
+    SpannerWriteGroupFn writerFn = new SpannerWriteGroupFn(write);
     DoFnTester<MutationGroup, Void> fnTester = DoFnTester.of(writerFn);
     fnTester.processBundle(Arrays.asList(one, two));
 
@@ -175,7 +175,7 @@ public class SpannerIOTest implements Serializable {
             .withDatabaseId("test-database")
             .withBatchSizeBytes(batchSize)
             .withServiceFactory(serviceFactory);
-    SpannerIO.SpannerWriteGroupFn writerFn = new SpannerIO.SpannerWriteGroupFn(write);
+    SpannerWriteGroupFn writerFn = new SpannerWriteGroupFn(write);
     DoFnTester<MutationGroup, Void> fnTester = DoFnTester.of(writerFn);
     fnTester.processBundle(Arrays.asList(one, two, three));
 
@@ -198,7 +198,7 @@ public class SpannerIOTest implements Serializable {
             .withDatabaseId("test-database")
             .withBatchSizeBytes(0) // turn off batching.
             .withServiceFactory(serviceFactory);
-    SpannerIO.SpannerWriteGroupFn writerFn = new SpannerIO.SpannerWriteGroupFn(write);
+    SpannerWriteGroupFn writerFn = new SpannerWriteGroupFn(write);
     DoFnTester<MutationGroup, Void> fnTester = DoFnTester.of(writerFn);
     fnTester.processBundle(Arrays.asList(one, two));
 
@@ -224,7 +224,7 @@ public class SpannerIOTest implements Serializable {
             .withDatabaseId("test-database")
             .withBatchSizeBytes(batchSize)
             .withServiceFactory(serviceFactory);
-    SpannerIO.SpannerWriteGroupFn writerFn = new SpannerIO.SpannerWriteGroupFn(write);
+    SpannerWriteGroupFn writerFn = new SpannerWriteGroupFn(write);
     DoFnTester<MutationGroup, Void> fnTester = DoFnTester.of(writerFn);
     fnTester.processBundle(Arrays.asList(g(one, two, three)));
 


Mime
View raw message