[BEAM-289] Examples Use TypeDescriptors

examples/java8/pom.xml

@@ -39,6 +39,19 @@
 
   <build>
     <plugins>
+      <!-- Disable javadoc for now.
+           TODO: this section should be removed as soon as possible. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-javadoc-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>attach-javadocs</id>
+            <phase/>
+          </execution>
+        </executions>
+      </plugin>
+
       <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>

examples/java8/src/main/java/org/apache/beam/examples/MinimalWordCountJava8.java

@@ -27,7 +27,7 @@
 import org.apache.beam.sdk.transforms.FlatMapElements;
 import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.values.KV;
-import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
 
 import java.util.Arrays;
 
@@ -54,12 +54,12 @@ public static void main(String[] args) {
 
     p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
      .apply(FlatMapElements.via((String word) -> Arrays.asList(word.split("[^a-zA-Z']+")))
-         .withOutputType(new TypeDescriptor<String>() {}))
+         .withOutputType(TypeDescriptors.strings()))
      .apply(Filter.byPredicate((String word) -> !word.isEmpty()))
      .apply(Count.<String>perElement())
      .apply(MapElements
          .via((KV<String, Long> wordCount) -> wordCount.getKey() + ": " + wordCount.getValue())
-         .withOutputType(new TypeDescriptor<String>() {}))
+         .withOutputType(TypeDescriptors.strings()))
 
      // CHANGE 3 of 3: The Google Cloud Storage path is required for outputting the results to.
      .apply(TextIO.Write.to("gs://YOUR_OUTPUT_BUCKET/AND_OUTPUT_PREFIX"));

examples/java8/src/main/java/org/apache/beam/examples/complete/game/GameStats.java

@@ -44,7 +44,7 @@
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionView;
-import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
 
 import org.joda.time.DateTimeZone;
 import org.joda.time.Duration;
@@ -255,7 +255,8 @@ public static void main(String[] args) throws Exception {
     PCollection<KV<String, Integer>> userEvents =
         rawEvents.apply("ExtractUserScore",
           MapElements.via((GameActionInfo gInfo) -> KV.of(gInfo.getUser(), gInfo.getScore()))
-            .withOutputType(new TypeDescriptor<KV<String, Integer>>() {}));
+            .withOutputType(
+                TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers())));
 
     // Calculate the total score per user over fixed windows, and
     // cumulative updates for late data.

examples/java8/src/main/java/org/apache/beam/examples/complete/game/UserScore.java

@@ -35,7 +35,7 @@
 import org.apache.beam.sdk.transforms.Sum;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
 
 import org.apache.avro.reflect.Nullable;
 import org.slf4j.Logger;
@@ -168,7 +168,8 @@ public PCollection<KV<String, Integer>> apply(
       return gameInfo
         .apply(MapElements
             .via((GameActionInfo gInfo) -> KV.of(gInfo.getKey(field), gInfo.getScore()))
-            .withOutputType(new TypeDescriptor<KV<String, Integer>>() {}))
+            .withOutputType(
+                TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers())))
         .apply(Sum.<String>integersPerKey());
     }
   }

examples/java8/src/test/java/org/apache/beam/examples/MinimalWordCountJava8Test.java

@@ -28,7 +28,7 @@
 import org.apache.beam.sdk.util.GcsUtil;
 import org.apache.beam.sdk.util.gcsfs.GcsPath;
 import org.apache.beam.sdk.values.KV;
-import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
 
 import com.google.common.collect.ImmutableList;
 
@@ -65,12 +65,12 @@ public void testMinimalWordCountJava8() throws Exception {
 
     p.apply(TextIO.Read.from("gs://dataflow-samples/shakespeare/*"))
      .apply(FlatMapElements.via((String word) -> Arrays.asList(word.split("[^a-zA-Z']+")))
-         .withOutputType(new TypeDescriptor<String>() {}))
+         .withOutputType(TypeDescriptors.strings()))
      .apply(Filter.byPredicate((String word) -> !word.isEmpty()))
      .apply(Count.<String>perElement())
      .apply(MapElements
          .via((KV<String, Long> wordCount) -> wordCount.getKey() + ": " + wordCount.getValue())
-         .withOutputType(new TypeDescriptor<String>() {}))
+         .withOutputType(TypeDescriptors.strings()))
      .apply(TextIO.Write.to("gs://YOUR_OUTPUT_BUCKET/AND_OUTPUT_PREFIX"));
   }
 

examples/java8/src/test/java/org/apache/beam/examples/complete/game/HourlyTeamScoreTest.java

@@ -30,7 +30,7 @@
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
 
 import org.joda.time.Instant;
 import org.junit.Test;
@@ -102,7 +102,8 @@ public void testUserScoresFilter() throws Exception {
       // run a map to access the fields in the result.
       .apply(MapElements
           .via((GameActionInfo gInfo) -> KV.of(gInfo.getUser(), gInfo.getScore()))
-          .withOutputType(new TypeDescriptor<KV<String, Integer>>() {}));
+          .withOutputType(
+              TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers())));
 
       PAssert.that(output).containsInAnyOrder(FILTERED_EVENTS);
 

examples/java8/src/test/java/org/apache/beam/examples/complete/game/UserScoreTest.java

@@ -31,7 +31,7 @@
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.TypeDescriptor;
+import org.apache.beam.sdk.values.TypeDescriptors;
 
 import org.junit.Assert;
 import org.junit.Test;
@@ -146,7 +146,8 @@ public void testUserScoresBadInput() throws Exception {
       .apply(ParDo.of(new ParseEventFn()))
       .apply(
           MapElements.via((GameActionInfo gInfo) -> KV.of(gInfo.getUser(), gInfo.getScore()))
-          .withOutputType(new TypeDescriptor<KV<String, Integer>>() {}));
+          .withOutputType(
+              TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptors.integers())));
 
     PAssert.that(extract).empty();
 

runners/flink/examples/pom.xml

@@ -86,6 +86,19 @@
       </plugin>
       -->
 
+      <!-- Disable javadoc for now.
+           TODO: this section should be removed as soon as possible. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-javadoc-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>attach-javadocs</id>
+            <phase/>
+          </execution>
+        </executions>
+      </plugin>
+
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>exec-maven-plugin</artifactId>

runners/flink/runner/pom.xml

@@ -63,7 +63,7 @@
       <version>${flink.version}</version>
     </dependency>
 
-    <!--- Beam -->
+    <!-- Beam -->
     <dependency>
       <groupId>org.apache.beam</groupId>
       <artifactId>java-sdk-all</artifactId>
@@ -159,6 +159,19 @@
       </plugin>
       -->
 
+      <!-- Disable javadoc for now.
+           TODO: this section should be removed as soon as possible. -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-javadoc-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>attach-javadocs</id>
+            <phase/>
+          </execution>
+        </executions>
+      </plugin>
+
       <!-- Integration Tests -->
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>

sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubIO.java

@@ -32,13 +32,13 @@
 import org.apache.beam.sdk.transforms.display.DisplayData;
 import org.apache.beam.sdk.transforms.windowing.AfterWatermark;
 import org.apache.beam.sdk.util.CoderUtils;
-import org.apache.beam.sdk.util.PubsubApiaryClient;
 import org.apache.beam.sdk.util.PubsubClient;
 import org.apache.beam.sdk.util.PubsubClient.IncomingMessage;
 import org.apache.beam.sdk.util.PubsubClient.OutgoingMessage;
 import org.apache.beam.sdk.util.PubsubClient.ProjectPath;
 import org.apache.beam.sdk.util.PubsubClient.SubscriptionPath;
 import org.apache.beam.sdk.util.PubsubClient.TopicPath;
+import org.apache.beam.sdk.util.PubsubJsonClient;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PDone;
 import org.apache.beam.sdk.values.PInput;
@@ -71,7 +71,7 @@ public class PubsubIO {
   private static final Logger LOG = LoggerFactory.getLogger(PubsubIO.class);
 
   /** Factory for creating pubsub client to manage transport. */
-  private static final PubsubClient.PubsubClientFactory FACTORY = PubsubApiaryClient.FACTORY;
+  private static final PubsubClient.PubsubClientFactory FACTORY = PubsubJsonClient.FACTORY;
 
   /** The default {@link Coder} used to translate to/from Cloud Pub/Sub messages. */
   public static final Coder<String> DEFAULT_PUBSUB_CODER = StringUtf8Coder.of();
@@ -646,7 +646,8 @@ public PCollection<T> apply(PInput input) {
         if (boundedOutput) {
           return input.getPipeline().begin()
                       .apply(Create.of((Void) null)).setCoder(VoidCoder.of())
-                      .apply(ParDo.of(new PubsubBoundedReader())).setCoder(coder);
+                      .apply(ParDo.of(new PubsubBoundedReader()))
+                      .setCoder(coder);
         } else {
           @Nullable ProjectPath projectPath =
               topic == null ? null : PubsubClient.projectPathFromId(topic.project);
@@ -655,8 +656,8 @@ public PCollection<T> apply(PInput input) {
           @Nullable SubscriptionPath subscriptionPath =
               subscription == null
                   ? null
-                  : PubsubClient
-                      .subscriptionPathFromName(subscription.project, subscription.subscription);
+                  : PubsubClient.subscriptionPathFromName(
+                      subscription.project, subscription.subscription);
           return input.getPipeline().begin()
                       .apply(new PubsubUnboundedSource<T>(
                           FACTORY, projectPath, topicPath, subscriptionPath,

sdks/java/core/src/main/java/org/apache/beam/sdk/io/PubsubUnboundedSink.java

@@ -57,8 +57,6 @@
 import com.google.common.hash.Hashing;
 
 import org.joda.time.Duration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -89,8 +87,6 @@
  * <p>NOTE: This is not the implementation used when running on the Google Cloud Dataflow service.
  */
 public class PubsubUnboundedSink<T> extends PTransform<PCollection<T>, PDone> {
-  private static final Logger LOG = LoggerFactory.getLogger(PubsubUnboundedSink.class);
-
   /**
    * Default maximum number of messages per publish.
    */
@@ -249,9 +245,8 @@ private static class WriterFn
      */
     private void publishBatch(List<OutgoingMessage> messages, int bytes)
         throws IOException {
-      long nowMsSinceEpoch = System.currentTimeMillis();
       int n = pubsubClient.publish(topic, messages);
-      checkState(n == messages.size(), "Attempted to publish %d messages but %d were successful",
+      checkState(n == messages.size(), "Attempted to publish %s messages but %s were successful",
                  messages.size(), n);
       batchCounter.addValue(1L);
       elementCounter.addValue((long) messages.size());

sdks/java/core/src/main/java/org/apache/beam/sdk/util/MovingFunction.java

@@ -29,11 +29,6 @@
  * {@link #sampleUpdateMs}.
  */
 public class MovingFunction {
-  /**
-   * How far back to retain samples, in ms.
-   */
-  private final long samplePeriodMs;
-
   /**
    * How frequently to update the moving function, in ms.
    */
@@ -77,7 +72,6 @@ public class MovingFunction {
   public MovingFunction(long samplePeriodMs, long sampleUpdateMs,
                         int numSignificantBuckets, int numSignificantSamples,
                         Combine.BinaryCombineLongFn function) {
-    this.samplePeriodMs = samplePeriodMs;
     this.sampleUpdateMs = sampleUpdateMs;
     this.numSignificantBuckets = numSignificantBuckets;
     this.numSignificantSamples = numSignificantSamples;
@@ -123,7 +117,7 @@ public void add(long nowMsSinceEpoch, long value) {
   }
 
   /**
-   * Return the minimum/maximum/sum of all retained values within {@link #samplePeriodMs}
+   * Return the minimum/maximum/sum of all retained values within samplePeriodMs
    * of {@code nowMsSinceEpoch}.
    */
   public long get(long nowMsSinceEpoch) {

sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubClient.java

@@ -57,7 +57,7 @@ PubsubClient newClient(
         PubsubOptions options) throws IOException;
 
     /**
-     * Return the display name for this factory. Eg "Apiary", "gRPC".
+     * Return the display name for this factory. Eg "Json", "gRPC".
      */
     String getKind();
   }

sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubApiaryClient.java → sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubJsonClient.java

@@ -22,6 +22,8 @@
 
 import org.apache.beam.sdk.options.PubsubOptions;
 
+import com.google.api.client.auth.oauth2.Credential;
+import com.google.api.client.http.HttpRequestInitializer;
 import com.google.api.services.pubsub.Pubsub;
 import com.google.api.services.pubsub.Pubsub.Builder;
 import com.google.api.services.pubsub.model.AcknowledgeRequest;
@@ -50,39 +52,48 @@
 import javax.annotation.Nullable;
 
 /**
- * A Pubsub client using Apiary.
+ * A Pubsub client using JSON transport.
  */
-public class PubsubApiaryClient extends PubsubClient {
+public class PubsubJsonClient extends PubsubClient {
+
+  private static class PubsubJsonClientFactory implements PubsubClientFactory {
+    private static HttpRequestInitializer chainHttpRequestInitializer(
+        Credential credential, HttpRequestInitializer httpRequestInitializer) {
+      if (credential == null) {
+        return httpRequestInitializer;
+      } else {
+        return new ChainingHttpRequestInitializer(credential, httpRequestInitializer);
+      }
+    }
 
-  private static class PubsubApiaryClientFactory implements PubsubClientFactory {
     @Override
     public PubsubClient newClient(
         @Nullable String timestampLabel, @Nullable String idLabel, PubsubOptions options)
         throws IOException {
       Pubsub pubsub = new Builder(
           Transport.getTransport(),
           Transport.getJsonFactory(),
-          new ChainingHttpRequestInitializer(
+          chainHttpRequestInitializer(
               options.getGcpCredential(),
               // Do not log 404. It clutters the output and is possibly even required by the caller.
               new RetryHttpRequestInitializer(ImmutableList.of(404))))
           .setRootUrl(options.getPubsubRootUrl())
           .setApplicationName(options.getAppName())
           .setGoogleClientRequestInitializer(options.getGoogleApiTrace())
           .build();
-      return new PubsubApiaryClient(timestampLabel, idLabel, pubsub);
+      return new PubsubJsonClient(timestampLabel, idLabel, pubsub);
     }
 
     @Override
     public String getKind() {
-      return "Apiary";
+      return "Json";
     }
   }
 
   /**
-   * Factory for creating Pubsub clients using Apiary transport.
+   * Factory for creating Pubsub clients using Json transport.
    */
-  public static final PubsubClientFactory FACTORY = new PubsubApiaryClientFactory();
+  public static final PubsubClientFactory FACTORY = new PubsubJsonClientFactory();
 
   /**
    * Label to use for custom timestamps, or {@literal null} if should use Pubsub publish time
@@ -98,12 +109,12 @@ public String getKind() {
   private final String idLabel;
 
   /**
-   * Underlying Apiary client.
+   * Underlying JSON transport.
    */
   private Pubsub pubsub;
 
   @VisibleForTesting
-  PubsubApiaryClient(
+  PubsubJsonClient(
       @Nullable String timestampLabel,
       @Nullable String idLabel,
       Pubsub pubsub) {