Fix javadoc throughout Beam

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

@@ -100,7 +100,8 @@ 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 larger than
+ * {@code (mean * SCORE_WEIGHT)}.
  */
  // [START DocInclude_AbuseDetect]
  public static class CalculateSpammyUsers

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

@@ -73,7 +73,7 @@
  * results, e.g. for 'team prizes'. We're now outputting 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

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

@@ -68,7 +68,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 {

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

@@ -53,11 +53,9 @@
 
 /**
  * To run the example, first open a socket on a terminal by executing the command:
- * <li>
- * <li>
- * <code>nc -lk 9999</code>
- * </li>
- * </li>
+ * <ul>
+ * <li><code>nc -lk 9999</code>
+ * </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

@@ -39,14 +39,10 @@
 
 /**
  * To run the example, first open two sockets on two terminals by executing the commands:
- * <li>
- * <li>
- * <code>nc -lk 9999</code>, and
- * </li>
- * <li>
- * <code>nc -lk 9998</code>
- * </li>
- * </li>
+ * <ul>
+ * <li><code>nc -lk 9999</code>, and
+ * <li><code>nc -lk 9998</code>
+ * </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

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

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

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

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineJob.java

@@ -199,7 +199,6 @@ public State waitUntilFinish(Duration duration) {
  * thread is interrupted.
  * @throws IOException If there is a persistent problem getting job
  * information.
- * @throws InterruptedException
  */
  @Nullable
  @VisibleForTesting

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/internal/IsmFormat.java

@@ -167,11 +167,12 @@ public byte[] getMetadata() {
  }
  }
 
- /** A {@link Coder} for {@link IsmRecord}s.
+ /**
+ * A {@link Coder} for {@link IsmRecord}s.
  *
  * <p>Note that this coder standalone will not produce an Ism file. This coder can be used
  * to materialize a {@link PCollection} of {@link IsmRecord}s. Only when this coder
- * is combined with an {@link IsmSink} will one produce an Ism file.
+ * is combined with an {@code IsmSink} will one produce an Ism file.
  *
  * <p>The {@link IsmRecord} encoded format is:
  * <ul>

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/MonitoringUtil.java

@@ -149,7 +149,6 @@ public int compare(JobMessage o1, JobMessage o2) {
  * @param startTimestampMs Return only those messages with a
  * timestamp greater than this value.
  * @return collection of messages
- * @throws IOException
  */
  public ArrayList<JobMessage> getJobMessages(
  String jobId, long startTimestampMs) throws IOException {

runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/RandomAccessData.java

@@ -121,7 +121,7 @@ protected long getEncodedElementByteSize(RandomAccessData value, Coder.Context c
  * A {@link Comparator} that compares two byte arrays lexicographically. It compares
  * values as a list of unsigned bytes. The first pair of values that follow any common prefix,
  * or when one array is a prefix of the other, treats the shorter array as the lesser.
- * For example, [] < [0x01] < [0x01, 0x7F] < [0x01, 0x80] < [0x02] < POSITIVE INFINITY.
+ * For example, {@code [] < [0x01] < [0x01, 0x7F] < [0x01, 0x80] < [0x02] < POSITIVE INFINITY}.
  *
  * <p>Note that a token type of positive infinity is supported and is greater than
  * all other {@link RandomAccessData}.

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

@@ -117,10 +117,10 @@
  * } </pre>
  *
  * <p>By default, {@link AvroIO.Write} produces output files that are compressed using the
- * {@link org.apache.avro.file.DeflateCodec CodecFactory.deflateCodec(6)}. This default can
+ * {@link org.apache.avro.file.Codec CodecFactory.deflateCodec(6)}. This default can
  * be changed or overridden using {@link AvroIO.Write#withCodec}.
  *
- * <p><h3>Permissions</h3>
+ * <h3>Permissions</h3>
  * Permission requirements depend on the {@link PipelineRunner} that is used to execute the
  * Dataflow job. Please refer to the documentation of corresponding {@link PipelineRunner}s for
  * more details.

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

@@ -50,9 +50,6 @@
  * </li>
  * </ul>
  *
- * <p>To use this class for supporting your custom input type, derive your class
- * class from it, and override the abstract methods. For an example, see {@link DatastoreIO}.
- *
  * @param <T> Type of records read by the source.
  */
 public abstract class BoundedSource<T> extends Source<T> {
@@ -135,7 +132,7 @@ public abstract static class BoundedReader<T> extends Source.Reader<T> {
  *
  * <p>By default, returns null to indicate that this cannot be estimated.
  *
- * <h5>Thread safety</h5>
+ * <h3>Thread safety</h3>
  * If {@link #splitAtFraction} is implemented, this method can be called concurrently to other
  * methods (including itself), and it is therefore critical for it to be implemented
  * in a thread-safe way.
@@ -345,7 +342,7 @@ public long getSplitPointsRemaining() {
  *
  * <p>Returns a {@code BoundedSource} representing the remainder.
  *
- * <h5>Detailed description</h5>
+ * <h3>Detailed description</h3>
  * Assuming the following sequence of calls:
  * <pre>{@code
  * BoundedSource<T> initial = reader.getCurrentSource();
@@ -371,11 +368,11 @@ public long getSplitPointsRemaining() {
  * corresponding to the given fraction. In this case, the method MUST have no effect
  * (the reader must behave as if the method hadn't been called at all).
  *
- * <h5>Statefulness</h5>
+ * <h3>Statefulness</h3>
  * Since this method (if successful) affects the reader's source, in subsequent invocations
  * "fraction" should be interpreted relative to the new current source.
  *
- * <h5>Thread safety and blocking</h5>
+ * <h3>Thread safety and blocking</h3>
  * This method will be called concurrently to other methods (however there will not be multiple
  * concurrent invocations of this method itself), and it is critical for it to be implemented
  * in a thread-safe way (otherwise data loss is possible).

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

@@ -74,7 +74,6 @@ public class CompressedSource<T> extends FileBasedSource<T> {
  public interface DecompressingChannelFactory extends Serializable {
  /**
  * Given a channel, create a channel that decompresses the content read from the channel.
- * @throws IOException
  */
  ReadableByteChannel createDecompressingChannel(ReadableByteChannel channel)
  throws IOException;
@@ -88,7 +87,6 @@ private interface FileNameBasedDecompressingChannelFactory
  extends DecompressingChannelFactory {
  /**
  * Given a channel, create a channel that decompresses the content read from the channel.
- * @throws IOException
  */
  ReadableByteChannel createDecompressingChannel(String fileName, ReadableByteChannel channel)
  throws IOException;

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

@@ -133,7 +133,8 @@ public String getMimeType() {
 
  /**
  * The {@link WritableByteChannelFactory} that is used to wrap the raw data output to the
- * underlying channel. The default is to not compress the output using {@link #UNCOMPRESSED}.
+ * underlying channel. The default is to not compress the output using
+ * {@link CompressionType#UNCOMPRESSED}.
  */
  protected final WritableByteChannelFactory writableByteChannelFactory;
 
@@ -206,11 +207,6 @@ public String getBaseOutputFilename() {
  return baseOutputFilename;
  }
 
- /**
- * Perform pipeline-construction-time validation. The default implementation is a no-op.
- * Subclasses should override to ensure the sink is valid and can be written to. It is recommended
- * to use {@link Preconditions#checkState(boolean)} in the implementation of this method.
- */
  @Override
  public void validate(PipelineOptions options) {}
 
@@ -802,7 +798,6 @@ public interface WritableByteChannelFactory extends Serializable {
  /**
  * @param channel the {@link WritableByteChannel} to wrap
  * @return the {@link WritableByteChannel} to be used during output
- * @throws IOException
  */
  WritableByteChannel create(WritableByteChannel channel) throws IOException;
 

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

@@ -108,7 +108,7 @@
  * <p>These dependencies have been declared as optional in the sdks/java/core/pom.xml file of
  * Apache Beam.
  *
- * <p><h3>Permissions</h3>
+ * <h3>Permissions</h3>
  * Permission requirements depend on the
  * {@link org.apache.beam.sdk.runners.PipelineRunner PipelineRunner} that is
  * used to execute the Beam pipeline. Please refer to the documentation of corresponding

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

@@ -133,7 +133,7 @@ public Boolean overlaps(ByteKeyRange other) {
  * Specifically, if this range is unsplittable (e.g., because the start and end keys are equal
  * up to padding by zero bytes), the list returned will only contain the start and end key.
  *
- * @throws IllegalArgumentException if the specified number of splits is < 1
+ * @throws IllegalArgumentException if the specified number of splits is less than 1
  * @see ByteKeyRange the ByteKeyRange class Javadoc for more information about split semantics.
  */
  public List<ByteKey> split(int numSplits) {

sdks/java/core/src/main/java/org/apache/beam/sdk/options/ValueProvider.java

@@ -133,7 +133,7 @@ public boolean isAccessible() {
  * construction time.
  *
  * <p>To enforce this contract, if there is no default, users must only call
- * {@link #get()} at execution time (after a call to {@link Pipeline#run}),
+ * {@link #get()} at execution time (after a call to {@link org.apache.beam.sdk.Pipeline#run}),
  * which will provide the value of {@code optionsMap}.
  */
  class RuntimeValueProvider<T> implements ValueProvider<T>, Serializable {

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Aggregator.java

@@ -29,26 +29,24 @@
  * typically from the {@link DoFn} constructor. Elements can be added to the
  * {@code Aggregator} by calling {@link Aggregator#addValue}.
  *
- * <p>Aggregators are visible in the monitoring UI, when the pipeline is run using
- * {@link DataflowRunner} along with their current value.
- * Aggregators may not become visible until the system begins executing the ParDo transform
- * that created them and/or their initial value is changed.
+ * <p>It is runner-dependent whether aggregators are accessible during pipeline execution or only
+ * after jobs have completed.
  *
  * <p>Example:
- * <pre> {@code
+ * <pre>{@code
  * class MyDoFn extends DoFn<String, String> {
  * private Aggregator<Integer, Integer> myAggregator;
  *
  * public MyDoFn() {
  * myAggregator = createAggregator("myAggregator", new Sum.SumIntegerFn());
  * }
  *
- * @ProcessElement
+ * {@literal @}ProcessElement
  * public void processElement(ProcessContext c) {
  * myAggregator.addValue(1);
  * }
  * }
- * } </pre>
+ * }</pre>
  *
  * @param <InputT> the type of input values
  * @param <OutputT> the type of output values

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/CombineFns.java

@@ -106,15 +106,15 @@ public static ComposeKeyedCombineFnBuilder composeKeyed() {
  * <p>The same {@link TupleTag} cannot be used in a composition multiple times.
  *
  * <p>Example:
- * <pre><{@code
+ * <pre>{@code
  * PCollection<Integer> globalLatencies = ...;
  *
  * TupleTag<Integer> maxLatencyTag = new TupleTag<Integer>();
  * TupleTag<Double> meanLatencyTag = new TupleTag<Double>();
  *
  * SimpleFunction<Integer, Integer> identityFn =
  * new SimpleFunction<Integer, Integer>() {
- * @Override
+ * {@literal @}Override
  * public Integer apply(Integer input) {
  * return input;
  * }};

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java

@@ -76,15 +76,15 @@
  *
  * <p>Example usage:
  *
- * <pre> {@code
+ * <pre>{@code
  * PCollection<String> lines = ... ;
  * PCollection<String> words =
  * lines.apply(ParDo.of(new DoFn<String, String>() {
- * @ProcessElement
+ * {@literal @}ProcessElement
  * public void processElement(ProcessContext c, BoundedWindow window) {
  *
  * }}));
- * } </pre>
+ * }</pre>
  *
  * @param <InputT> the type of the (main) input elements
  * @param <OutputT> the type of the (main) output elements
@@ -436,14 +436,14 @@ public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker() {
  *
  * <pre>{@code
  * new DoFn<KV<Key, Foo>, Baz>() {
- * @StateId("my-state-id")
+ * {@literal @}StateId("my-state-id")
  * private final StateSpec<K, ValueState<MyState>> myStateSpec =
  * StateSpecs.value(new MyStateCoder());
  *
- * @ProcessElement
+ * {@literal @}ProcessElement
  * public void processElement(
  * ProcessContext c,
- * @StateId("my-state-id") ValueState<MyState> myState) {
+ * {@literal @}StateId("my-state-id") ValueState<MyState> myState) {
  * myState.read();
  * myState.write(...);
  * }
@@ -480,17 +480,17 @@ public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker() {
  *
  * <pre>{@code
  * new DoFn<KV<Key, Foo>, Baz>() {
- * @TimerId("my-timer-id")
+ * {@literal @}TimerId("my-timer-id")
  * private final TimerSpec myTimer = TimerSpecs.timerForDomain(TimeDomain.EVENT_TIME);
  *
- * @ProcessElement
+ * {@literal @}ProcessElement
  * public void processElement(
  * ProcessContext c,
- * @TimerId("my-timer-id") Timer myTimer) {
+ * {@literal @}TimerId("my-timer-id") Timer myTimer) {
  * myTimer.setForNowPlus(Duration.standardSeconds(...));
  * }
  *
- * @OnTimer("my-timer-id")
+ * {@literal @}OnTimer("my-timer-id")
  * public void onMyTimer() {
  * ...
  * }
@@ -578,7 +578,7 @@ public <RestrictionT> RestrictionTracker<RestrictionT> restrictionTracker() {
  * <ul>
  * <li>Its first argument must be a {@link DoFn.ProcessContext}.
  * <li>If one of its arguments is a subtype of {@link RestrictionTracker}, then it is a <a
- * href="https://s.apache.org/splittable-do-fn>splittable</a> {@link DoFn} subject to the
+ * href="https://s.apache.org/splittable-do-fn">splittable</a> {@link DoFn} subject to the
  * separate requirements described below. Items below are assuming this is not a splittable
  * {@link DoFn}.
  * <li>If one of its arguments is {@link BoundedWindow}, this argument corresponds to the window

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java

@@ -96,7 +96,7 @@
  *
  * <p>For example:
  *
- * <pre><{@code
+ * <pre>{@code
  * PCollection<String> lines = ...;
  * PCollection<String> words =
  * lines.apply(ParDo.of(new DoFn<String, String>() {

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/View.java

@@ -213,7 +213,7 @@ public static <K, V> AsMap<K, V> asMap() {
 
  /**
  * Returns a {@link View.AsMultimap} transform that takes a
- * {@link PCollection PCollection&lt;KV&ltK, V&gt;&gt;}
+ * {@link PCollection PCollection&lt;KV&lt;K, V&gt;&gt;}
  * as input and produces a {@link PCollectionView} mapping
  * each window to its contents as a {@link Map Map&lt;K, Iterable&lt;V&gt;&gt;}
  * for use as a side input.