[BEAM-242] Fix javadoc on Flink runner.

examples/java8/pom.xml

@@ -39,16 +39,6 @@
 
   <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>
-        <configuration>
-          <skip>true</skip>
-        </configuration>
-      </plugin>
-
       <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>

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

@@ -102,7 +102,7 @@ public class GameStats extends LeaderBoard {
   /**
    * Filter out all but those users with a high clickrate, which we will consider as 'spammy' uesrs.
    * We do this by finding the mean total score per user, then using that information as a side
-   * input to filter out all but those user scores that are > (mean * SCORE_WEIGHT)
+   * input to filter out all but those user scores that are greater (mean * SCORE_WEIGHT).
    */
   // [START DocInclude_AbuseDetect]
   public static class CalculateSpammyUsers
@@ -193,6 +193,8 @@ static interface Options extends LeaderBoard.Options {
   /**
    * Create a map of information that describes how to write pipeline output to BigQuery. This map
    * is used to write information about team score sums.
+   *
+   * @return The map of fields to write on big query.
    */
   protected static Map<String, WriteWindowedToBigQuery.FieldInfo<KV<String, Integer>>>
       configureWindowedWrite() {
@@ -219,6 +221,8 @@ static interface Options extends LeaderBoard.Options {
   /**
    * Create a map of information that describes how to write pipeline output to BigQuery. This map
    * is used to write information about mean user session time.
+   *
+   * @return The map of fields to write on big query.
    */
   protected static Map<String, WriteWindowedToBigQuery.FieldInfo<Double>>
       configureSessionWindowWrite() {

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

@@ -119,6 +119,8 @@ static interface Options extends UserScore.Options {
    * Create a map of information that describes how to write pipeline output to BigQuery. This map
    * is passed to the {@link WriteWindowedToBigQuery} constructor to write team score sums and
    * includes information about window start time.
+   *
+   * @return The map of field to write on big querey.
    */
   protected static Map<String, WriteWindowedToBigQuery.FieldInfo<KV<String, Integer>>>
       configureWindowedTableWrite() {
@@ -142,6 +144,9 @@ static interface Options extends UserScore.Options {
 
   /**
    * Run a batch pipeline to do windowed analysis of the data.
+   *
+   * @param args The main arguments.
+   * @throws Exception in case of execution failure.
    */
   // [START DocInclude_HTSMain]
   public static void main(String[] args) throws Exception {

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

@@ -72,7 +72,7 @@
  * results, e.g. for 'team prizes'. We're now outputing window results as they're
  * calculated, giving us much lower latency than with the previous batch examples.
  *
- * <p> Run {@link injector.Injector} to generate pubsub data for this pipeline.  The Injector
+ * <p> Run {@code injector.Injector} to generate pubsub data for this pipeline.  The Injector
  * documentation provides more detail on how to do this.
  *
  * <p> To execute this pipeline using the Dataflow service, specify the pipeline configuration
@@ -128,6 +128,8 @@ static interface Options extends HourlyTeamScore.Options, DataflowExampleOptions
   /**
    * Create a map of information that describes how to write pipeline output to BigQuery. This map
    * is used to write team score sums and includes event timing information.
+   *
+   * @return The map of fields to write on big query.
    */
   protected static Map<String, WriteWindowedToBigQuery.FieldInfo<KV<String, Integer>>>
       configureWindowedTableWrite() {
@@ -158,6 +160,8 @@ static interface Options extends HourlyTeamScore.Options, DataflowExampleOptions
   /**
    * Create a map of information that describes how to write pipeline output to BigQuery. This map
    * is used to write user score sums.
+   *
+   * @return the Map of field to write on big query.
    */
   protected static Map<String, WriteToBigQuery.FieldInfo<KV<String, Integer>>>
       configureGlobalWindowBigQueryWrite() {

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

@@ -70,7 +70,7 @@
  * where the BigQuery dataset you specify must already exist.
  *
  * <p> Optionally include the --input argument to specify a batch input file.
- * See the --input default value for example batch data file, or use {@link injector.Injector} to
+ * See the --input default value for example batch data file, or use {@code injector.Injector} to
  * generate your own batch data.
   */
 public class UserScore {
@@ -201,6 +201,8 @@ public static interface Options extends PipelineOptions {
   /**
    * Create a map of information that describes how to write pipeline output to BigQuery. This map
    * is passed to the {@link WriteToBigQuery} constructor to write user score sums.
+   *
+   * @return A map to fields written on big query.
    */
   protected static Map<String, WriteToBigQuery.FieldInfo<KV<String, Integer>>>
     configureBigQueryWrite() {
@@ -216,6 +218,9 @@ public static interface Options extends PipelineOptions {
 
   /**
    * Run a batch pipeline.
+   *
+   * @param args The main arguments.
+   * @throws Exception in case of execution failure.
    */
  // [START DocInclude_USMain]
   public static void main(String[] args) throws Exception {

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

@@ -100,7 +100,10 @@ public void processElement(ProcessContext c) {
     }
   }
 
-  /** Build the output table schema. */
+  /** Build the output table schema.
+   *
+   * @return The {@link TableSchema}.
+   */
   protected TableSchema getSchema() {
     List<TableFieldSchema> fields = new ArrayList<>();
     for (Map.Entry<String, FieldInfo<T>> entry : fieldInfo.entrySet()) {

runners/flink/examples/pom.xml

@@ -86,16 +86,6 @@
       </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>
-        <configuration>
-          <skip>true</skip>
-        </configuration>
-      </plugin>
-
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>exec-maven-plugin</artifactId>

runners/flink/examples/src/main/java/org/apache/beam/runners/flink/examples/TFIDF.java

@@ -110,6 +110,10 @@ private interface Options extends PipelineOptions, FlinkPipelineOptions {
 
   /**
    * Lists documents contained beneath the {@code options.input} prefix/directory.
+   *
+   * @param options the pipeline options.
+   * @throws URISyntaxException in case of malformed URI
+   * @throws IOException in case of URI loading issue.
    */
   public static Set<URI> listInputDocuments(Options options)
       throws URISyntaxException, IOException {

runners/flink/examples/src/main/java/org/apache/beam/runners/flink/examples/streaming/AutoComplete.java

@@ -55,11 +55,11 @@
 
 /**
  * To run the example, first open a socket on a terminal by executing the command:
- * <li>
+ * <ul>
  *     <li>
  *     <code>nc -lk 9999</code>
  *     </li>
- * </li>
+ * </ul>
  * and then launch the example. Now whatever you type in the terminal is going to be
  * the input to the program.
  * */

runners/flink/examples/src/main/java/org/apache/beam/runners/flink/examples/streaming/JoinExamples.java

@@ -42,14 +42,14 @@
 
 /**
  * To run the example, first open two sockets on two terminals by executing the commands:
- * <li>
+ * <ul>
  *     <li>
  *         <code>nc -lk 9999</code>, and
  *     </li>
  *     <li>
  *         <code>nc -lk 9998</code>
  *     </li>
- * </li>
+ * </ul>
  * and then launch the example. Now whatever you type in the terminal is going to be
  * the input to the program.
  * */

runners/flink/examples/src/main/java/org/apache/beam/runners/flink/examples/streaming/KafkaIOExamples.java

@@ -171,7 +171,8 @@ public static void main(String[] args) {
 
     /**
      * Serialiation/Deserialiation schema for Avro types
-     * @param <T>
+     * 
+     * @param <T> serialization type.
      */
     static class AvroSerializationDeserializationSchema<T>
         implements SerializationSchema<T>, DeserializationSchema<T> {

runners/flink/examples/src/main/java/org/apache/beam/runners/flink/examples/streaming/WindowedWordCount.java

@@ -44,11 +44,11 @@
 
 /**
  * To run the example, first open a socket on a terminal by executing the command:
- * <li>
+ * <ul>
  *     <li>
  *     <code>nc -lk 9999</code>
  *     </li>
- * </li>
+ * </ul>
  * and then launch the example. Now whatever you type in the terminal is going to be
  * the input to the program.
  * */

runners/flink/runner/pom.xml

@@ -159,16 +159,6 @@
       </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>
-        <configuration>
-          <skip>true</skip>
-        </configuration>
-      </plugin>
-
       <!-- Integration Tests -->
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineExecutionEnvironment.java

@@ -120,6 +120,8 @@ private void createPipelineTranslator() {
    * the {@link org.apache.beam.sdk.values.PCollection} program into
    * a {@link org.apache.flink.api.java.DataSet} or {@link org.apache.flink.streaming.api.datastream.DataStream}
    * one.
+   *
+   * @param pipeline The pipeline to translate.
    * */
   public void translate(Pipeline pipeline) {
     checkInitializationState();
@@ -134,7 +136,10 @@ public void translate(Pipeline pipeline) {
 
   /**
    * Launches the program execution.
-   * */
+   *
+   * @return The job execution result.
+   * @throws Exception In case of error during pipeline execution.
+   */
   public JobExecutionResult executePipeline() throws Exception {
     if (options.isStreaming()) {
       if (this.flinkStreamEnv == null) {

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java

@@ -39,11 +39,17 @@
 public interface FlinkPipelineOptions extends PipelineOptions, ApplicationNameOptions, StreamingOptions {
 
   /**
+   * <p>
    * List of local files to make available to workers.
+   * </p>
    * <p>
    * Jars are placed on the worker's classpath.
+   * </p>
    * <p>
    * The default value is the list of jars from the main program's classpath.
+   * </p>
+   *
+   * @return The list of files to stage.
    */
   @Description("Jar-Files to send to all workers and put on the classpath. " +
       "The default value is all files from the classpath.")
@@ -53,6 +59,8 @@ public interface FlinkPipelineOptions extends PipelineOptions, ApplicationNameOp
 
   /**
    * The job name is used to identify jobs running on a Flink cluster.
+   *
+   * @return The Flink job name.
    */
   @Description("Flink job name, to uniquely identify active jobs. "
       + "Defaults to using the ApplicationName-UserName-Date.")
@@ -66,6 +74,8 @@ public interface FlinkPipelineOptions extends PipelineOptions, ApplicationNameOp
    * Strings "[local]", "[collection]" or "[auto]". "[local]" will start a local Flink
    * Cluster in the JVM, "[collection]" will execute the pipeline on Java Collections while
    * "[auto]" will let the system decide where to execute the pipeline based on the environment.
+   *
+   * @return The Flink master URL.
    */
   @Description("Address of the Flink Master where the Pipeline should be executed. Can" +
       " either be of the form \"host:port\" or one of the special values [local], " +

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/FlinkPipelineRunner.java

@@ -41,10 +41,11 @@
 import java.util.Map;
 
 /**
+ * <p>
  * A {@link PipelineRunner} that executes the operations in the
  * pipeline by first translating them to a Flink Plan and then executing them either locally
  * or on a Flink cluster, depending on the configuration.
- * <p>
+ * </p>
  */
 public class FlinkPipelineRunner extends PipelineRunner<FlinkRunnerResult> {
 
@@ -133,6 +134,8 @@ public FlinkRunnerResult run(Pipeline pipeline) {
 
   /**
    * For testing.
+   *
+   * @return The {@link FlinkPipelineOptions}.
    */
   public FlinkPipelineOptions getPipelineOptions() {
     return options;

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkBatchTranslationContext.java

@@ -98,7 +98,8 @@ public <T> void setOutputDataSet(PValue value, DataSet<WindowedValue<T>> set) {
 
   /**
    * Sets the AppliedPTransform which carries input/output.
-   * @param currentTransform
+   *
+   * @param currentTransform The current applied {@link PTransform}.
    */
   public void setCurrentTransform(AppliedPTransform<?, ?, ?> currentTransform) {
     this.currentTransform = currentTransform;

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingPipelineTranslator.java

@@ -32,7 +32,7 @@
  * {@link org.apache.beam.sdk.values.PCollection}-based job into a
  * {@link org.apache.flink.streaming.api.datastream.DataStream} one.
  *
- * This is based on {@link org.apache.beam.runners.dataflow.DataflowPipelineTranslator}
+ * This is based on {@code org.apache.beam.runners.dataflow.DataflowPipelineTranslator}
  * */
 public class FlinkStreamingPipelineTranslator extends FlinkPipelineTranslator {
 

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTranslationContext.java

@@ -82,7 +82,8 @@ public void setOutputDataStream(PValue value, DataStream<?> set) {
 
   /**
    * Sets the AppliedPTransform which carries input/output.
-   * @param currentTransform
+   *
+   * @param currentTransform the current applied {@link PTransform}.
    */
   public void setCurrentTransform(AppliedPTransform<?, ?, ?> currentTransform) {
     this.currentTransform = currentTransform;

runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/types/InspectableByteArrayOutputStream.java

@@ -27,6 +27,8 @@ public class InspectableByteArrayOutputStream extends ByteArrayOutputStream {
 
   /**
    * Get the underlying byte array.
+   *
+   * @return a buffer as {@code byte[]}.
    */
   public byte[] getBuffer() {
     return buf;