From 6a7d94ee032bae2d7f8c5f3d3471be3049b16b61 Mon Sep 17 00:00:00 2001
From: "Matthias J. Sax" <matthias@confluent.io>
Date: Fri, 27 Jan 2017 17:20:36 -0800
Subject: [PATCH] MINOR: JavaDoc markup cleanup

---
 .../apache/kafka/streams/KafkaStreams.java    |  21 +-
 .../kafka/streams/kstream/Aggregator.java     |   6 +-
 .../kafka/streams/kstream/ForeachAction.java  |   2 +-
 .../kafka/streams/kstream/GlobalKTable.java   |  18 +-
 .../kafka/streams/kstream/Initializer.java    |   4 +-
 .../kafka/streams/kstream/JoinWindows.java    |  10 +-
 .../kafka/streams/kstream/KGroupedStream.java |   4 +-
 .../kafka/streams/kstream/KGroupedTable.java  |   4 +-
 .../apache/kafka/streams/kstream/KStream.java | 274 +++++++++---------
 .../kafka/streams/kstream/KStreamBuilder.java |   2 +-
 .../apache/kafka/streams/kstream/KTable.java  | 250 ++++++++--------
 .../kafka/streams/kstream/KeyValueMapper.java |   4 +-
 .../kafka/streams/kstream/Predicate.java      |   2 +-
 .../apache/kafka/streams/kstream/Reducer.java |   4 +-
 .../kafka/streams/kstream/SessionWindows.java |   6 +-
 .../kafka/streams/kstream/TimeWindows.java    |   9 +-
 .../kafka/streams/kstream/Transformer.java    |   4 +-
 .../streams/kstream/TransformerSupplier.java  |   2 +-
 .../streams/kstream/UnlimitedWindows.java     |   4 +-
 .../kafka/streams/kstream/ValueJoiner.java    |   2 +-
 .../kafka/streams/kstream/ValueMapper.java    |   4 +-
 .../streams/kstream/ValueTransformer.java     |   6 +-
 .../kstream/ValueTransformerSupplier.java     |   2 +-
 .../apache/kafka/streams/kstream/Window.java  |   6 +-
 .../kafka/streams/kstream/Windowed.java       |   2 +-
 25 files changed, 324 insertions(+), 328 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index ef0e72dd3a8ef..c57804e532510 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -92,7 +92,7 @@
  * If instances are added or fail, all (remaining) instances will rebalance the partition assignment among themselves
  * to balance processing load and ensure that all input topic partitions are processed.
  * <p>
- * Internally a {@link KafkaStreams} instance contains a normal {@link KafkaProducer} and {@link KafkaConsumer} instance
+ * Internally a {@code KafkaStreams} instance contains a normal {@link KafkaProducer} and {@link KafkaConsumer} instance
  * that is used for reading input and writing output.
  * <p>
  * A simple example might look like this:
@@ -187,7 +187,7 @@ public boolean isValidTransition(final State newState) {
             return validTransitions.contains(newState.ordinal());
         }
     }
-    private volatile State state = KafkaStreams.State.CREATED;
+    private volatile State state = State.CREATED;
     private StateListener stateListener = null;
 
 
@@ -412,9 +412,9 @@ private void checkBrokerVersionCompatibility() throws StreamsException {
     public synchronized void start() throws IllegalStateException, StreamsException {
         log.debug("Starting Kafka Stream process.");
 
-        if (state == KafkaStreams.State.CREATED) {
+        if (state == State.CREATED) {
             checkBrokerVersionCompatibility();
-            setState(KafkaStreams.State.RUNNING);
+            setState(State.RUNNING);
 
             if (globalStreamThread != null) {
                 globalStreamThread.start();
@@ -451,7 +451,7 @@ public void close() {
     public synchronized boolean close(final long timeout, final TimeUnit timeUnit) {
         log.debug("Stopping Kafka Stream process.");
         if (state.isCreatedOrRunning()) {
-            setState(KafkaStreams.State.PENDING_SHUTDOWN);
+            setState(State.PENDING_SHUTDOWN);
             // save the current thread so that if it is a stream thread
             // we don't attempt to join it and cause a deadlock
             final Thread shutdown = new Thread(new Runnable() {
@@ -495,7 +495,7 @@ public void run() {
             } catch (final InterruptedException e) {
                 Thread.interrupted();
             }
-            setState(KafkaStreams.State.NOT_RUNNING);
+            setState(State.NOT_RUNNING);
             return !shutdown.isAlive();
         }
         return true;
@@ -539,11 +539,10 @@ public String toString(final String indent) {
      * Do a clean up of the local {@link StateStore} directory ({@link StreamsConfig#STATE_DIR_CONFIG}) by deleting all
      * data with regard to the {@link StreamsConfig#APPLICATION_ID_CONFIG application ID}.
      * <p>
-     * May only be called either before this {@code KafkaStreams} instance is {@link KafkaStreams#start() started} or
-     * after the instance is {@link KafkaStreams#close() closed}.
+     * May only be called either before this {@code KafkaStreams} instance is {@link #start() started} or after the
+     * instance is {@link #close() closed}.
      * <p>
-     * Calling this method triggers a restore of local {@link StateStore}s on the next {@link KafkaStreams#start()
-     * application start}.
+     * Calling this method triggers a restore of local {@link StateStore}s on the next {@link #start() application start}.
      *
      * @throws IllegalStateException if the instance is currently running
      */
@@ -629,7 +628,7 @@ public Collection<StreamsMetadata> allMetadataForStore(final String storeName) {
      * {@link ProducerConfig#PARTITIONER_CLASS_CONFIG configured} via {@link StreamsConfig},
      * {@link KStream#through(StreamPartitioner, String)}, or {@link KTable#through(StreamPartitioner, String, String)},
      * or if the original {@link KTable}'s input {@link KStreamBuilder#table(String, String) topic} is partitioned
-     * differently, please use {@link KafkaStreams#metadataForKey(String, Object, StreamPartitioner)}.
+     * differently, please use {@link #metadataForKey(String, Object, StreamPartitioner)}.
      * <p>
      * Note:
      * <ul>
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java
index e433ea7334558..786cfa2b727d1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java
@@ -20,12 +20,12 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * The {@link Aggregator} interface for aggregating values of the given key.
+ * The {@code Aggregator} interface for aggregating values of the given key.
  * This is a generalization of {@link Reducer} and allows to have different types for input value and aggregation
  * result.
- * {@link Aggregator} is used in combination with {@link Initializer} that provides an initial aggregation value.
+ * {@code Aggregator} is used in combination with {@link Initializer} that provides an initial aggregation value.
  * <p>
- * {@link Aggregator} can be used to implement aggregation functions like count.
+ * {@code Aggregator} can be used to implement aggregation functions like count.
 
  * @param <K> key type
  * @param <V> input value type
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java
index e68bf8dbe75e5..92fdf99434ab2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java
@@ -19,7 +19,7 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * The {@link ForeachAction} interface for performing an action on a {@link org.apache.kafka.streams.KeyValue key-value
+ * The {@code ForeachAction} interface for performing an action on a {@link org.apache.kafka.streams.KeyValue key-value
  * pair}.
  * This is a stateless record-by-record operation, i.e, {@link #apply(Object, Object)} is invoked individually for each
  * record of a stream.
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java
index 56f7bc8638ee5..fa1a5f29be5df 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java
@@ -23,24 +23,24 @@
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 
 /**
- * {@link GlobalKTable} is an abstraction of a <i>changelog stream</i> from a primary-keyed table.
+ * {@code GlobalKTable} is an abstraction of a <i>changelog stream</i> from a primary-keyed table.
  * Each record in this changelog stream is an update on the primary-keyed table with the record key as the primary key.
  * <p>
- * {@link GlobalKTable} can only be used as right-hand side input for {@link KStream stream}-table joins.
+ * {@code GlobalKTable} can only be used as right-hand side input for {@link KStream stream}-table joins.
  * <p>
- * In contrast to a {@link KTable} that is partitioned over all {@link KafkaStreams} instances, a {@link GlobalKTable}
+ * In contrast to a {@link KTable} that is partitioned over all {@link KafkaStreams} instances, a {@code GlobalKTable}
  * is fully replicated per {@link KafkaStreams} instance.
- * Every partition of the underlying topic is consumed by each {@link GlobalKTable}, such that the full set of data is
+ * Every partition of the underlying topic is consumed by each {@code GlobalKTable}, such that the full set of data is
  * available in every {@link KafkaStreams} instance.
  * This provides the ability to perform joins with {@link KStream} without having to repartition the input stream.
- * All joins with the {@link GlobalKTable} require that a {@link KeyValueMapper} is provided that can map from the
- * {@link KeyValue} of the left hand side {@link KStream} to the key of the right hand side {@link GlobalKTable}.
+ * All joins with the {@code GlobalKTable} require that a {@link KeyValueMapper} is provided that can map from the
+ * {@link KeyValue} of the left hand side {@link KStream} to the key of the right hand side {@code GlobalKTable}.
  * <p>
- * A {@link GlobalKTable} is created via a {@link KStreamBuilder}. For example:
+ * A {@code GlobalKTable} is created via a {@link KStreamBuilder}. For example:
  * <pre>{@code
  * builder.globalTable("topic-name", "queryable-store-name");
  * }</pre>
- * all {@link GlobalKTable}s are backed by a {@link ReadOnlyKeyValueStore} and are therefore queryable via the
+ * all {@code GlobalKTable}s are backed by a {@link ReadOnlyKeyValueStore} and are therefore queryable via the
  * interactive queries API.
  * For example:
  * <pre>{@code
@@ -53,7 +53,7 @@
  * ReadOnlyKeyValueStore view = streams.store("g1-store", QueryableStoreTypes.keyValueStore());
  * view.get(key); // can be done on any key, as all keys are present
  *}</pre>
- * Note that in contrast to {@link KTable} a {@link GlobalKTable}'s state holds a full copy if the underlying topic and
+ * Note that in contrast to {@link KTable} a {@code GlobalKTable}'s state holds a full copy if the underlying topic and
  * thus, all keys can be queried locally.
  *
  * @param <K> Type of primary keys
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java
index 96a6995605f8f..bcd16b1321a1f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java
@@ -20,8 +20,8 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * The {@link Initializer} interface for creating an initial value in aggregations.
- * {@link Initializer} is used in combination with {@link Aggregator}.
+ * The {@code Initializer} interface for creating an initial value in aggregations.
+ * {@code Initializer} is used in combination with {@link Aggregator}.
  *
  * @param <VA> aggregate value type
  * @see Aggregator
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
index 720efb298f1df..3869721bdaf9d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
@@ -25,7 +25,7 @@
 /**
  * The window specifications used for joins.
  * <p>
- * A {@link JoinWindows} instance defines a maximum time difference for a {@link KStream#join(KStream, ValueJoiner,
+ * A {@code JoinWindows} instance defines a maximum time difference for a {@link KStream#join(KStream, ValueJoiner,
  * JoinWindows) join over two streams} on the same key.
  * In SQL-style you would express this join as
  * <pre>{@code
@@ -47,7 +47,7 @@
  * Both values (before and after) must not result in an "inverse" window, i.e., upper-interval bound cannot be smaller
  * than lower-interval bound.
  * <p>
- * {@link JoinWindows} are sliding windows, thus, they are aligned to the actual record timestamps.
+ * {@code JoinWindows} are sliding windows, thus, they are aligned to the actual record timestamps.
  * This implies, that each input record defines its own window with start and end time being relative to the record's
  * timestamp.
  * <p>
@@ -65,7 +65,7 @@
  * @see TimestampExtractor
  */
 @InterfaceStability.Unstable
-public class JoinWindows extends Windows<Window> {
+public final class JoinWindows extends Windows<Window> {
 
     /** Maximum time difference for tuples that are before the join tuple. */
     public final long beforeMs;
@@ -121,7 +121,7 @@ public JoinWindows after(final long timeDifferenceMs) throws IllegalArgumentExce
     }
 
     /**
-     * Not supported by {@link JoinWindows}.
+     * Not supported by {@code JoinWindows}.
      * Throws {@link UnsupportedOperationException}.
      *
      * @throws UnsupportedOperationException at every invocation
@@ -137,8 +137,6 @@ public long size() {
     }
 
     /**
-     * {@inheritDoc}
-     *
      * @param durationMs the window retention time in milliseconds
      * @return itself
      * @throws IllegalArgumentException if {@code durationMs} is smaller than the window size
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java
index f1f13381abb6c..d767f0be62087 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java
@@ -26,14 +26,14 @@
 import org.apache.kafka.streams.state.WindowStore;
 
 /**
- * {@link KGroupedStream} is an abstraction of a <i>grouped</i> record stream of {@link KeyValue} pairs.
+ * {@code KGroupedStream} is an abstraction of a <i>grouped</i> record stream of {@link KeyValue} pairs.
  * It is an intermediate representation of a {@link KStream} in order to apply an aggregation operation on the original
  * {@link KStream} records.
  * <p>
  * It is an intermediate representation after a grouping of a {@link KStream} before an aggregation is applied to the
  * new partitions resulting in a {@link KTable}.
  * <p>
- * A {@link KGroupedStream} must be obtained from a {@link KStream} via {@link KStream#groupByKey() groupByKey()} or
+ * A {@code KGroupedStream} must be obtained from a {@link KStream} via {@link KStream#groupByKey() groupByKey()} or
  * {@link KStream#groupBy(KeyValueMapper) groupBy(...)}.
  *
  * @param <K> Type of keys
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java
index c736a8567a6eb..53c0a127fc409 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java
@@ -26,13 +26,13 @@
 import org.apache.kafka.streams.state.QueryableStoreType;
 
 /**
- * {@link KGroupedTable} is an abstraction of a <i>re-grouped changelog stream</i> from a primary-keyed table,
+ * {@code KGroupedTable} is an abstraction of a <i>re-grouped changelog stream</i> from a primary-keyed table,
  * usually on a different grouping key than the original primary key.
  * <p>
  * It is an intermediate representation after a re-grouping of a {@link KTable} before an aggregation is applied to the
  * new partitions resulting in a new {@link KTable}.
  * <p>
- * A {@link KGroupedTable} must be obtained from a {@link KTable} via {@link KTable#groupBy(KeyValueMapper)
+ * A {@code KGroupedTable} must be obtained from a {@link KTable} via {@link KTable#groupBy(KeyValueMapper)
  * groupBy(...)}.
  *
  * @param <K> Type of keys
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
index 15eb05820c96d..21135fb23b442 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
@@ -31,16 +31,16 @@
 import org.apache.kafka.streams.processor.TopologyBuilder;
 
 /**
- * {@link KStream} is an abstraction of a <i>record stream</i> of {@link KeyValue} pairs, i.e., each record is an
+ * {@code KStream} is an abstraction of a <i>record stream</i> of {@link KeyValue} pairs, i.e., each record is an
  * independent entity/event in the real world.
  * For example a user X might buy two items I1 and I2, and thus there might be two records {@code <K:I1>, <K:I2>}
  * in the stream.
  * <p>
- * A {@link KStream} is either {@link KStreamBuilder#stream(String...) defined from one or multiple Kafka topics} that
- * are consumed message by message or the result of a {@link KStream} transformation.
- * A {@link KTable} can also be {@link KTable#toStream() converted} into a {@link KStream}.
+ * A {@code KStream} is either {@link KStreamBuilder#stream(String...) defined from one or multiple Kafka topics} that
+ * are consumed message by message or the result of a {@code KStream} transformation.
+ * A {@link KTable} can also be {@link KTable#toStream() converted} into a {@code KStream}.
  * <p>
- * A {@link KStream} can be transformed record by record, joined with another {@link KStream}, {@link KTable},
+ * A {@code KStream} can be transformed record by record, joined with another {@code KStream}, {@link KTable},
  * {@link GlobalKTable}, or can be aggregated into a {@link KTable}.
  * Kafka Streams DSL can be mixed-and-matched with Processor API (PAPI) (c.f. {@link TopologyBuilder}) via
  * {@link #process(ProcessorSupplier, String...) process(...)},
@@ -58,24 +58,24 @@
 public interface KStream<K, V> {
 
     /**
-     * Create a new {@link KStream} that consists of all records of this stream which satisfy the given predicate.
+     * Create a new {@code KStream} that consists of all records of this stream which satisfy the given predicate.
      * All records that do not satisfy the predicate are dropped.
      * This is a stateless record-by-record operation.
      *
      * @param predicate a filter {@link Predicate} that is applied to each record
-     * @return a {@link KStream} that contains only those records that satisfy the given predicate
+     * @return a {@code KStream} that contains only those records that satisfy the given predicate
      * @see #filterNot(Predicate)
      */
     KStream<K, V> filter(Predicate<? super K, ? super V> predicate);
 
     /**
-     * Create a new {@link KStream} that consists all records of this stream which do <em>not</em> satisfy the given
+     * Create a new {@code KStream} that consists all records of this stream which do <em>not</em> satisfy the given
      * predicate.
      * All records that <em>do</em> satisfy the predicate are dropped.
      * This is a stateless record-by-record operation.
      *
      * @param predicate a filter {@link Predicate} that is applied to each record
-     * @return a {@link KStream} that contains only those records that do <em>not</em> satisfy the given predicate
+     * @return a {@code KStream} that contains only those records that do <em>not</em> satisfy the given predicate
      * @see #filter(Predicate)
      */
     KStream<K, V> filterNot(Predicate<? super K, ? super V> predicate);
@@ -99,11 +99,11 @@ public interface KStream<K, V> {
      * }</pre>
      * <p>
      * Setting a new key might result in an internal data redistribution if a key based operator (like an aggregation or
-     * join) is applied to the result {@link KStream}.
+     * join) is applied to the result {@code KStream}.
      *
      * @param mapper a {@link KeyValueMapper} that computes a new key for each record
      * @param <KR>   the new key type of the result stream
-     * @return a {@link KStream} that contains records with new key (possibly of different type) and unmodified value
+     * @return a {@code KStream} that contains records with new key (possibly of different type) and unmodified value
      * @see #map(KeyValueMapper)
      * @see #flatMap(KeyValueMapper)
      * @see #mapValues(ValueMapper)
@@ -132,12 +132,12 @@ public interface KStream<K, V> {
      * The provided {@link KeyValueMapper} must return a {@link KeyValue} type and must not return {@code null}.
      * <p>
      * Mapping records might result in an internal data redistribution if a key based operator (like an aggregation or
-     * join) is applied to the result {@link KStream}. (cf. {@link #mapValues(ValueMapper)})
+     * join) is applied to the result {@code KStream}. (cf. {@link #mapValues(ValueMapper)})
      *
      * @param mapper a {@link KeyValueMapper} that computes a new output record
      * @param <KR>   the key type of the result stream
      * @param <VR>   the value type of the result stream
-     * @return a {@link KStream} that contains records with new key and value (possibly both of different type)
+     * @return a {@code KStream} that contains records with new key and value (possibly both of different type)
      * @see #selectKey(KeyValueMapper)
      * @see #flatMap(KeyValueMapper)
      * @see #mapValues(ValueMapper)
@@ -166,11 +166,11 @@ public interface KStream<K, V> {
      * <p>
      * Setting a new value preserves data co-location with respect to the key.
      * Thus, <em>no</em> internal data redistribution is required if a key based operator (like an aggregation or join)
-     * is applied to the result {@link KStream}. (cf. {@link #map(KeyValueMapper)})
+     * is applied to the result {@code KStream}. (cf. {@link #map(KeyValueMapper)})
      *
      * @param mapper a {@link ValueMapper} that computes a new output value
      * @param <VR>   the value type of the result stream
-     * @return a {@link KStream} that contains records with unmodified key and new values (possibly of different type)
+     * @return a {@code KStream} that contains records with unmodified key and new values (possibly of different type)
      * @see #selectKey(KeyValueMapper)
      * @see #map(KeyValueMapper)
      * @see #flatMap(KeyValueMapper)
@@ -210,12 +210,12 @@ public interface KStream<K, V> {
      * and the return value must not be {@code null}.
      * <p>
      * Flat-mapping records might result in an internal data redistribution if a key based operator (like an aggregation
-     * or join) is applied to the result {@link KStream}. (cf. {@link #flatMapValues(ValueMapper)})
+     * or join) is applied to the result {@code KStream}. (cf. {@link #flatMapValues(ValueMapper)})
      *
      * @param mapper a {@link KeyValueMapper} that computes the new output records
      * @param <KR>   the key type of the result stream
      * @param <VR>   the value type of the result stream
-     * @return a {@link KStream} that contains more or less records with new key and value (possibly of different type)
+     * @return a {@code KStream} that contains more or less records with new key and value (possibly of different type)
      * @see #selectKey(KeyValueMapper)
      * @see #map(KeyValueMapper)
      * @see #mapValues(ValueMapper)
@@ -226,7 +226,7 @@ public interface KStream<K, V> {
     <KR, VR> KStream<KR, VR> flatMap(final KeyValueMapper<? super K, ? super V, ? extends Iterable<? extends KeyValue<? extends KR, ? extends VR>>> mapper);
 
     /**
-     * Create a new {@link KStream} by transforming the value of each record in this stream into zero or more values
+     * Create a new {@code KStream} by transforming the value of each record in this stream into zero or more values
      * with the same key in the new stream.
      * Transform the value of each input record into zero or more records with the same (unmodified) key in the output
      * stream (value type can be altered arbitrarily).
@@ -250,11 +250,11 @@ public interface KStream<K, V> {
      * <p>
      * Splitting a record into multiple records with the same key preserves data co-location with respect to the key.
      * Thus, <em>no</em> internal data redistribution is required if a key based operator (like an aggregation or join)
-     * is applied to the result {@link KStream}. (cf. {@link #flatMap(KeyValueMapper)})
+     * is applied to the result {@code KStream}. (cf. {@link #flatMap(KeyValueMapper)})
      *
      * @param processor a {@link ValueMapper} the computes the new output values
      * @param <VR>      the value type of the result stream
-     * @return a {@link KStream} that contains more or less records with unmodified keys and new values of different type
+     * @return a {@code KStream} that contains more or less records with unmodified keys and new values of different type
      * @see #selectKey(KeyValueMapper)
      * @see #map(KeyValueMapper)
      * @see #flatMap(KeyValueMapper)
@@ -396,7 +396,7 @@ void writeAsText(final String filePath,
                      final Serde<V> valSerde);
 
     /**
-     * Perform an action on each record of {@link KStream}.
+     * Perform an action on each record of {@code KStream}.
      * This is a stateless record-by-record operation (cf. {@link #process(ProcessorSupplier, String...)}).
      * Note that this is a terminal operation that returns void.
      *
@@ -406,7 +406,7 @@ void writeAsText(final String filePath,
     void foreach(final ForeachAction<? super K, ? super V> action);
 
     /**
-     * Creates an array of {@link KStream} from this stream by branching the records in the original stream based on
+     * Creates an array of {@code KStream} from this stream by branching the records in the original stream based on
      * the supplied predicates.
      * Each record is evaluated against the supplied predicates, and predicates are evaluated in order.
      * Each stream in the result array corresponds position-wise (index) to the predicate in the supplied predicates.
@@ -416,13 +416,13 @@ void writeAsText(final String filePath,
      * This is a stateless record-by-record operation.
      *
      * @param predicates the ordered list of {@link Predicate} instances
-     * @return multiple distinct substreams of this {@link KStream}
+     * @return multiple distinct substreams of this {@code KStream}
      */
     @SuppressWarnings("unchecked")
     KStream<K, V>[] branch(final Predicate<? super K, ? super V>... predicates);
 
     /**
-     * Materialize this stream to a topic and creates a new {@link KStream} from the topic using default serializers and
+     * Materialize this stream to a topic and creates a new {@code KStream} from the topic using default serializers and
      * deserializers and producer's {@link DefaultPartitioner}.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
@@ -431,12 +431,12 @@ void writeAsText(final String filePath,
      * KStreamBuilder#stream(someTopicName)}.
      *
      * @param topic the topic name
-     * @return a {@link KStream} that contains the exact same (and potentially repartitioned) records as this {@link KStream}
+     * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
      */
     KStream<K, V> through(final String topic);
 
     /**
-     * Materialize this stream to a topic and creates a new {@link KStream} from the topic using default serializers and
+     * Materialize this stream to a topic and creates a new {@code KStream} from the topic using default serializers and
      * deserializers and a customizable {@link StreamPartitioner} to determine the distribution of records to partitions.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
@@ -447,13 +447,13 @@ void writeAsText(final String filePath,
      * @param partitioner the function used to determine how records are distributed among partitions of the topic,
      *                    if not specified producer's {@link DefaultPartitioner} will be used
      * @param topic       the topic name
-     * @return a {@link KStream} that contains the exact same (and potentially repartitioned) records as this {@link KStream}
+     * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
      */
     KStream<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                           final String topic);
 
     /**
-     * Materialize this stream to a topic, and creates a new {@link KStream} from the topic.
+     * Materialize this stream to a topic, and creates a new {@code KStream} from the topic.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
      * <p>
@@ -468,14 +468,14 @@ KStream<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
      * @param valSerde value serde used to send key-value pairs,
      *                 if not specified the default value serde defined in the configuration will be used
      * @param topic    the topic name
-     * @return a {@link KStream} that contains the exact same (and potentially repartitioned) records as this {@link KStream}
+     * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
      */
     KStream<K, V> through(final Serde<K> keySerde,
                           final Serde<V> valSerde,
                           final String topic);
 
     /**
-     * Materialize this stream to a topic and creates a new {@link KStream} from the topic using a customizable
+     * Materialize this stream to a topic and creates a new {@code KStream} from the topic using a customizable
      * {@link StreamPartitioner} to determine the distribution of records to partitions.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
@@ -493,7 +493,7 @@ KStream<K, V> through(final Serde<K> keySerde,
      *                    {@link WindowedStreamPartitioner} will be used&mdash;otherwise {@link DefaultPartitioner} will
      *                    be used
      * @param topic       the topic name
-     * @return a {@link KStream} that contains the exact same (and potentially repartitioned) records as this {@link KStream}
+     * @return a {@code KStream} that contains the exact same (and potentially repartitioned) records as this {@code KStream}
      */
     KStream<K, V> through(final Serde<K> keySerde,
                           final Serde<V> valSerde,
@@ -626,14 +626,14 @@ void to(final Serde<K> keySerde,
      * }</pre>
      * <p>
      * Transforming records might result in an internal data redistribution if a key based operator (like an aggregation
-     * or join) is applied to the result {@link KStream}.
+     * or join) is applied to the result {@code KStream}.
      * (cf. {@link #transformValues(ValueTransformerSupplier, String...)})
      *
      * @param transformerSupplier a instance of {@link TransformerSupplier} that generates a {@link Transformer}
      * @param stateStoreNames     the names of the state stores used by the processor
      * @param <K1>                the key type of the new stream
      * @param <V1>                the value type of the new stream
-     * @return a {@link KStream} that contains more or less records with new key and value (possibly of different type)
+     * @return a {@code KStream} that contains more or less records with new key and value (possibly of different type)
      * @see #flatMap(KeyValueMapper)
      * @see #transformValues(ValueTransformerSupplier, String...)
      * @see #process(ProcessorSupplier, String...)
@@ -703,13 +703,13 @@ <K1, V1> KStream<K1, V1> transform(final TransformerSupplier<? super K, ? super
      * <p>
      * Setting a new value preserves data co-location with respect to the key.
      * Thus, <em>no</em> internal data redistribution is required if a key based operator (like an aggregation or join)
-     * is applied to the result {@link KStream}. (cf. {@link #transform(TransformerSupplier, String...)})
+     * is applied to the result {@code KStream}. (cf. {@link #transform(TransformerSupplier, String...)})
      *
      * @param valueTransformerSupplier a instance of {@link ValueTransformerSupplier} that generates a
      *                                 {@link ValueTransformer}
      * @param stateStoreNames          the names of the state stores used by the processor
      * @param <VR>                     the value type of the result stream
-     * @return a {@link KStream} that contains records with unmodified key and new values (possibly of different type)
+     * @return a {@code KStream} that contains records with unmodified key and new values (possibly of different type)
      * @see #mapValues(ValueMapper)
      * @see #transform(TransformerSupplier, String...)
      */
@@ -800,7 +800,7 @@ void process(final ProcessorSupplier<? super K, ? super V> processorSupplier,
      * If the last key changing operator changed the key type, it is recommended to use
      * {@link #groupByKey(Serde, Serde)} instead.
      *
-     * @return a {@link KGroupedStream} that contains the grouped records of the original {@link KStream}
+     * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KStream}
      * @see #groupBy(KeyValueMapper)
      */
     KGroupedStream<K, V> groupByKey();
@@ -828,13 +828,13 @@ void process(final ProcessorSupplier<? super K, ? super V> processorSupplier,
      *                 if not specified the default serdes defined in the configs will be used
      * @param valSerde value serdes for materializing this stream,
      *                 if not specified the default serdes defined in the configs will be used
-     * @return a {@link KGroupedStream} that contains the grouped records of the original {@link KStream}
+     * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KStream}
      */
     KGroupedStream<K, V> groupByKey(final Serde<K> keySerde,
                                     final Serde<V> valSerde);
 
     /**
-     * Group the records of this {@link KStream} on a new key that is selected using the provided {@link KeyValueMapper}
+     * Group the records of this {@code KStream} on a new key that is selected using the provided {@link KeyValueMapper}
      * and default serializers and deserializers.
      * Grouping a stream on the record key is required before an aggregation operator can be applied to the data
      * (cf. {@link KGroupedStream}).
@@ -855,12 +855,12 @@ KGroupedStream<K, V> groupByKey(final Serde<K> keySerde,
      *
      * @param selector a {@link KeyValueMapper} that computes a new key for grouping
      * @param <KR>     the key type of the result {@link KGroupedStream}
-     * @return a {@link KGroupedStream} that contains the grouped records of the original {@link KStream}
+     * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KStream}
      */
     <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR> selector);
 
     /**
-     * Group the records of this {@link KStream} on a new key that is selected using the provided {@link KeyValueMapper}.
+     * Group the records of this {@code KStream} on a new key that is selected using the provided {@link KeyValueMapper}.
      * Grouping a stream on the record key is required before an aggregation operator can be applied to the data
      * (cf. {@link KGroupedStream}).
      * The {@link KeyValueMapper} selects a new key (with potentially different type) while preserving the original values.
@@ -884,7 +884,7 @@ KGroupedStream<K, V> groupByKey(final Serde<K> keySerde,
      * @param valSerde value serdes for materializing this stream,
      *                 if not specified the default serdes defined in the configs will be used
      * @param <KR>     the key type of the result {@link KGroupedStream}
-     * @return a {@link KGroupedStream} that contains the grouped records of the original {@link KStream}
+     * @return a {@link KGroupedStream} that contains the grouped records of the original {@code KStream}
      * @see #groupByKey()
      */
     <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR> selector,
@@ -892,7 +892,7 @@ <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR
                                        final Serde<V> valSerde);
 
     /**
-     * Join records of this stream with another {@link KStream}'s records using windowed inner equi join with default
+     * Join records of this stream with another {@code KStream}'s records using windowed inner equi join with default
      * serializers and deserializers.
      * The join is computed on the records' key with join attribute {@code thisKStream.key == otherKStream.key}.
      * Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
@@ -902,7 +902,7 @@ <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR
      * a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
      * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
-     * output record will be added to the resulting {@link KStream}.
+     * output record will be added to the resulting {@code KStream}.
      * <p>
      * Example (assuming all input records belong to the correct windows):
      * <table border='1'>
@@ -936,12 +936,12 @@ <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen for one or both of the joining {@link KStream}s.
+     * Repartitioning can happen for one or both of the joining {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      * <p>
-     * Both of the joining {@link KStream}s will be materialized in local state stores with auto-generated store names.
+     * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
      * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
      * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
      * in {@link StreamsConfig} via parameter
@@ -949,12 +949,12 @@ <KR> KGroupedStream<KR, V> groupBy(final KeyValueMapper<? super K, ? super V, KR
      * internally generated name, and "-changelog" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      *
-     * @param otherStream the {@link KStream} to be joined with this stream
+     * @param otherStream the {@code KStream} to be joined with this stream
      * @param joiner      a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param windows     the specification of the {@link JoinWindows}
      * @param <VO>        the value type of the other stream
      * @param <VR>        the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key and within the joining window intervals
      * @see #leftJoin(KStream, ValueJoiner, JoinWindows)
      * @see #outerJoin(KStream, ValueJoiner, JoinWindows)
@@ -964,7 +964,7 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
                                  final JoinWindows windows);
 
     /**
-     * Join records of this stream with another {@link KStream}'s records using windowed inner equi join.
+     * Join records of this stream with another {@code KStream}'s records using windowed inner equi join.
      * The join is computed on the records' key with join attribute {@code thisKStream.key == otherKStream.key}.
      * Furthermore, two records are only joined if their timestamps are close to each other as defined by the given
      * {@link JoinWindows}, i.e., the window defines an additional join predicate on the record timestamps.
@@ -973,7 +973,7 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
      * a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
      * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
-     * output record will be added to the resulting {@link KStream}.
+     * output record will be added to the resulting {@code KStream}.
      * <p>
      * Example (assuming all input records belong to the correct windows):
      * <table border='1'>
@@ -1007,19 +1007,19 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen for one or both of the joining {@link KStream}s.
+     * Repartitioning can happen for one or both of the joining {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      * <p>
-     * Both of the joining {@link KStream}s will be materialized in local state stores with auto-generated store names.
+     * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
      * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
      * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
      * in {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
      * "storeName" is an internally generated name, and "-changelog" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      *
-     * @param otherStream     the {@link KStream} to be joined with this stream
+     * @param otherStream     the {@code KStream} to be joined with this stream
      * @param joiner          a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param windows         the specification of the {@link JoinWindows}
      * @param keySerde        key serdes for materializing both streams,
@@ -1030,7 +1030,7 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
      *                        if not specified the default serdes defined in the configs will be used
      * @param <VO>            the value type of the other stream
      * @param <VR>            the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key and within the joining window intervals
      * @see #leftJoin(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde)
      * @see #outerJoin(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde)
@@ -1043,7 +1043,7 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
                                  final Serde<VO> otherValueSerde);
 
     /**
-     * Join records of this stream with another {@link KStream}'s records using windowed left equi join with default
+     * Join records of this stream with another {@code KStream}'s records using windowed left equi join with default
      * serializers and deserializers.
      * In contrast to {@link #join(KStream, ValueJoiner, JoinWindows) inner-join}, all records from this stream will
      * produce at least one output record (cf. below).
@@ -1054,10 +1054,10 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
      * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
      * a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
-     * Furthermore, for each input record of this {@link KStream} that does not satisfy the join predicate the provided
+     * Furthermore, for each input record of this {@code KStream} that does not satisfy the join predicate the provided
      * {@link ValueJoiner} will be called with a {@code null} value for the other stream.
      * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
-     * output record will be added to the resulting {@link KStream}.
+     * output record will be added to the resulting {@code KStream}.
      * <p>
      * Example (assuming all input records belong to the correct windows):
      * <table border='1'>
@@ -1091,26 +1091,26 @@ <VO, VR> KStream<K, VR> join(final KStream<K, VO> otherStream,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen for one or both of the joining {@link KStream}s.
+     * Repartitioning can happen for one or both of the joining {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      * <p>
-     * Both of the joining {@link KStream}s will be materialized in local state stores with auto-generated store names.
+     * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
      * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
      * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
      * in {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
      * "storeName" is an internally generated name, and "-changelog" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      *
-     * @param otherStream the {@link KStream} to be joined with this stream
+     * @param otherStream the {@code KStream} to be joined with this stream
      * @param joiner      a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param windows     the specification of the {@link JoinWindows}
      * @param <VO>        the value type of the other stream
      * @param <VR>        the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key plus one for each non-matching record of
-     * this {@link KStream} and within the joining window intervals
+     * this {@code KStream} and within the joining window intervals
      * @see #join(KStream, ValueJoiner, JoinWindows)
      * @see #outerJoin(KStream, ValueJoiner, JoinWindows)
      */
@@ -1119,7 +1119,7 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
                                      final JoinWindows windows);
 
     /**
-     * Join records of this stream with another {@link KStream}'s records using windowed left equi join.
+     * Join records of this stream with another {@code KStream}'s records using windowed left equi join.
      * In contrast to {@link #join(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde) inner-join}, all records from
      * this stream will produce at least one output record (cf. below).
      * The join is computed on the records' key with join attribute {@code thisKStream.key == otherKStream.key}.
@@ -1129,10 +1129,10 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
      * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
      * a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
-     * Furthermore, for each input record of this {@link KStream} that does not satisfy the join predicate the provided
+     * Furthermore, for each input record of this {@code KStream} that does not satisfy the join predicate the provided
      * {@link ValueJoiner} will be called with a {@code null} value for the other stream.
      * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
-     * output record will be added to the resulting {@link KStream}.
+     * output record will be added to the resulting {@code KStream}.
      * <p>
      * Example (assuming all input records belong to the correct windows):
      * <table border='1'>
@@ -1166,12 +1166,12 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen for one or both of the joining {@link KStream}s.
+     * Repartitioning can happen for one or both of the joining {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      * <p>
-     * Both of the joining {@link KStream}s will be materialized in local state stores with auto-generated store names.
+     * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
      * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
      * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
      * in {@link StreamsConfig} via parameter
@@ -1179,7 +1179,7 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
      * internally generated name, and "-changelog" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      *
-     * @param otherStream     the {@link KStream} to be joined with this stream
+     * @param otherStream     the {@code KStream} to be joined with this stream
      * @param joiner          a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param windows         the specification of the {@link JoinWindows}
      * @param keySerde        key serdes for materializing the other stream,
@@ -1190,9 +1190,9 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
      *                        if not specified the default serdes defined in the configs will be used
      * @param <VO>            the value type of the other stream
      * @param <VR>            the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key plus one for each non-matching record of
-     * this {@link KStream} and within the joining window intervals
+     * this {@code KStream} and within the joining window intervals
      * @see #join(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde)
      * @see #outerJoin(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde)
      */
@@ -1204,7 +1204,7 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
                                      final Serde<VO> otherValueSerde);
 
     /**
-     * Join records of this stream with another {@link KStream}'s records using windowed left equi join with default
+     * Join records of this stream with another {@code KStream}'s records using windowed left equi join with default
      * serializers and deserializers.
      * In contrast to {@link #join(KStream, ValueJoiner, JoinWindows) inner-join} or
      * {@link #leftJoin(KStream, ValueJoiner, JoinWindows) left-join}, all records from both streams will produce at
@@ -1216,10 +1216,10 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
      * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
      * a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
-     * Furthermore, for each input record of both {@link KStream}s that does not satisfy the join predicate the provided
+     * Furthermore, for each input record of both {@code KStream}s that does not satisfy the join predicate the provided
      * {@link ValueJoiner} will be called with a {@code null} value for the this/other stream, respectively.
      * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
-     * output record will be added to the resulting {@link KStream}.
+     * output record will be added to the resulting {@code KStream}.
      * <p>
      * Example (assuming all input records belong to the correct windows):
      * <table border='1'>
@@ -1253,26 +1253,26 @@ <VO, VR> KStream<K, VR> leftJoin(final KStream<K, VO> otherStream,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen for one or both of the joining {@link KStream}s.
+     * Repartitioning can happen for one or both of the joining {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      * <p>
-     * Both of the joining {@link KStream}s will be materialized in local state stores with auto-generated store names.
+     * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
      * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
      * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
      * in {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
      * "storeName" is an internally generated name, and "-changelog" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      *
-     * @param otherStream the {@link KStream} to be joined with this stream
+     * @param otherStream the {@code KStream} to be joined with this stream
      * @param joiner      a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param windows     the specification of the {@link JoinWindows}
      * @param <VO>        the value type of the other stream
      * @param <VR>        the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key plus one for each non-matching record of
-     * both {@link KStream} and within the joining window intervals
+     * both {@code KStream} and within the joining window intervals
      * @see #join(KStream, ValueJoiner, JoinWindows)
      * @see #leftJoin(KStream, ValueJoiner, JoinWindows)
      */
@@ -1281,7 +1281,7 @@ <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
                                       final JoinWindows windows);
 
     /**
-     * Join records of this stream with another {@link KStream}'s records using windowed left equi join.
+     * Join records of this stream with another {@code KStream}'s records using windowed left equi join.
      * In contrast to {@link #join(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde) inner-join} or
      * {@link #leftJoin(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde) left-join}, all records from both
      * streams will produce at least one output record (cf. below).
@@ -1292,10 +1292,10 @@ <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
      * For each pair of records meeting both join predicates the provided {@link ValueJoiner} will be called to compute
      * a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
-     * Furthermore, for each input record of both {@link KStream}s that does not satisfy the join predicate the provided
+     * Furthermore, for each input record of both {@code KStream}s that does not satisfy the join predicate the provided
      * {@link ValueJoiner} will be called with a {@code null} value for this/other stream, respectively.
      * If an input record key or value is {@code null} the record will not be included in the join operation and thus no
-     * output record will be added to the resulting {@link KStream}.
+     * output record will be added to the resulting {@code KStream}.
      * <p>
      * Example (assuming all input records belong to the correct windows):
      * <table border='1'>
@@ -1329,19 +1329,19 @@ <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen for one or both of the joining {@link KStream}s.
+     * Repartitioning can happen for one or both of the joining {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      * <p>
-     * Both of the joining {@link KStream}s will be materialized in local state stores with auto-generated store names.
+     * Both of the joining {@code KStream}s will be materialized in local state stores with auto-generated store names.
      * For failure and recovery each store will be backed by an internal changelog topic that will be created in Kafka.
      * The changelog topic will be named "${applicationId}-storeName-changelog", where "applicationId" is user-specified
      * in {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
      * "storeName" is an internally generated name, and "-changelog" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      *
-     * @param otherStream     the {@link KStream} to be joined with this stream
+     * @param otherStream     the {@code KStream} to be joined with this stream
      * @param joiner          a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param windows         the specification of the {@link JoinWindows}
      * @param keySerde        key serdes for materializing both streams,
@@ -1352,9 +1352,9 @@ <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
      *                        if not specified the default serdes defined in the configs will be used
      * @param <VO>            the value type of the other stream
      * @param <VR>            the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key plus one for each non-matching record of
-     * both {@link KStream}s and within the joining window intervals
+     * both {@code KStream}s and within the joining window intervals
      * @see #join(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde)
      * @see #leftJoin(KStream, ValueJoiner, JoinWindows, Serde, Serde, Serde)
      */
@@ -1369,17 +1369,17 @@ <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
      * Join records of this stream with {@link KTable}'s records using non-windowed inner equi join with default
      * serializers and deserializers.
      * The join is a primary key table lookup join with join attribute {@code stream.key == table.key}.
-     * "Table lookup join" means, that results are only computed if {@link KStream} records are processed.
+     * "Table lookup join" means, that results are only computed if {@code KStream} records are processed.
      * This is done by performing a lookup for matching records in the <em>current</em> (i.e., processing time) internal
      * {@link KTable} state.
      * In contrast, processing {@link KTable} input records will only update the internal {@link KTable} state and
      * will not produce any result records.
      * <p>
-     * For each {@link KStream} record that finds a corresponding record in {@link KTable} the provided
+     * For each {@code KStream} record that finds a corresponding record in {@link KTable} the provided
      * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
-     * If an {@link KStream} input record key or value is {@code null} the record will not be included in the join
-     * operation and thus no output record will be added to the resulting {@link KStream}.
+     * If an {@code KStream} input record key or value is {@code null} the record will not be included in the join
+     * operation and thus no output record will be added to the resulting {@code KStream}.
      * <p>
      * Example:
      * <table border='1'>
@@ -1418,16 +1418,16 @@ <VO, VR> KStream<K, VR> outerJoin(final KStream<K, VO> otherStream,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen only for this {@link KStream}s.
+     * Repartitioning can happen only for this {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      *
      * @param table  the {@link KTable} to be joined with this stream
      * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param <VT>   the value type of the table
      * @param <VR>   the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key
      * @see #leftJoin(KTable, ValueJoiner)
      * @see #join(GlobalKTable, KeyValueMapper, ValueJoiner)
@@ -1438,17 +1438,17 @@ <VT, VR> KStream<K, VR> join(final KTable<K, VT> table,
     /**
      * Join records of this stream with {@link KTable}'s records using non-windowed inner equi join.
      * The join is a primary key table lookup join with join attribute {@code stream.key == table.key}.
-     * "Table lookup join" means, that results are only computed if {@link KStream} records are processed.
+     * "Table lookup join" means, that results are only computed if {@code KStream} records are processed.
      * This is done by performing a lookup for matching records in the <em>current</em> (i.e., processing time) internal
      * {@link KTable} state.
      * In contrast, processing {@link KTable} input records will only update the internal {@link KTable} state and
      * will not produce any result records.
      * <p>
-     * For each {@link KStream} record that finds a corresponding record in {@link KTable} the provided
+     * For each {@code KStream} record that finds a corresponding record in {@link KTable} the provided
      * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
-     * If an {@link KStream} input record key or value is {@code null} the record will not be included in the join
-     * operation and thus no output record will be added to the resulting {@link KStream}.
+     * If an {@code KStream} input record key or value is {@code null} the record will not be included in the join
+     * operation and thus no output record will be added to the resulting {@code KStream}.
      * <p>
      * Example:
      * <table border='1'>
@@ -1487,9 +1487,9 @@ <VT, VR> KStream<K, VR> join(final KTable<K, VT> table,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen only for this {@link KStream}s.
+     * Repartitioning can happen only for this {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      *
      * @param table    the {@link KTable} to be joined with this stream
@@ -1500,7 +1500,7 @@ <VT, VR> KStream<K, VR> join(final KTable<K, VT> table,
      *                 if not specified the default serdes defined in the configs will be used
      * @param <VT>     the value type of the table
      * @param <VR>     the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key
      * @see #leftJoin(KTable, ValueJoiner, Serde, Serde)
      * @see #join(GlobalKTable, KeyValueMapper, ValueJoiner)
@@ -1516,18 +1516,18 @@ <VT, VR> KStream<K, VR> join(final KTable<K, VT> table,
      * In contrast to {@link #join(KTable, ValueJoiner) inner-join}, all records from this stream will produce an
      * output record (cf. below).
      * The join is a primary key table lookup join with join attribute {@code stream.key == table.key}.
-     * "Table lookup join" means, that results are only computed if {@link KStream} records are processed.
+     * "Table lookup join" means, that results are only computed if {@code KStream} records are processed.
      * This is done by performing a lookup for matching records in the <em>current</em> (i.e., processing time) internal
      * {@link KTable} state.
      * In contrast, processing {@link KTable} input records will only update the internal {@link KTable} state and
      * will not produce any result records.
      * <p>
-     * For each {@link KStream} record weather or not it finds a corresponding record in {@link KTable} the provided
+     * For each {@code KStream} record weather or not it finds a corresponding record in {@link KTable} the provided
      * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * If no {@link KTable} record was found during lookup, a {@code null} value will be provided to {@link ValueJoiner}.
      * The key of the result record is the same as for both joining input records.
-     * If an {@link KStream} input record key or value is {@code null} the record will not be included in the join
-     * operation and thus no output record will be added to the resulting {@link KStream}.
+     * If an {@code KStream} input record key or value is {@code null} the record will not be included in the join
+     * operation and thus no output record will be added to the resulting {@code KStream}.
      * <p>
      * Example:
      * <table border='1'>
@@ -1566,17 +1566,17 @@ <VT, VR> KStream<K, VR> join(final KTable<K, VT> table,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen only for this {@link KStream}s.
+     * Repartitioning can happen only for this {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      *
      * @param table  the {@link KTable} to be joined with this stream
      * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param <VT>   the value type of the table
      * @param <VR>   the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
-     * {@link ValueJoiner}, one output for each input {@link KStream} record
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
+     * {@link ValueJoiner}, one output for each input {@code KStream} record
      * @see #join(KTable, ValueJoiner)
      * @see #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
      */
@@ -1588,18 +1588,18 @@ <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,
      * In contrast to {@link #join(KTable, ValueJoiner) inner-join}, all records from this stream will produce an
      * output record (cf. below).
      * The join is a primary key table lookup join with join attribute {@code stream.key == table.key}.
-     * "Table lookup join" means, that results are only computed if {@link KStream} records are processed.
+     * "Table lookup join" means, that results are only computed if {@code KStream} records are processed.
      * This is done by performing a lookup for matching records in the <em>current</em> (i.e., processing time) internal
      * {@link KTable} state.
      * In contrast, processing {@link KTable} input records will only update the internal {@link KTable} state and
      * will not produce any result records.
      * <p>
-     * For each {@link KStream} record weather or not it finds a corresponding record in {@link KTable} the provided
+     * For each {@code KStream} record weather or not it finds a corresponding record in {@link KTable} the provided
      * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * If no {@link KTable} record was found during lookup, a {@code null} value will be provided to {@link ValueJoiner}.
      * The key of the result record is the same as for both joining input records.
-     * If an {@link KStream} input record key or value is {@code null} the record will not be included in the join
-     * operation and thus no output record will be added to the resulting {@link KStream}.
+     * If an {@code KStream} input record key or value is {@code null} the record will not be included in the join
+     * operation and thus no output record will be added to the resulting {@code KStream}.
      * <p>
      * Example:
      * <table border='1'>
@@ -1638,9 +1638,9 @@ <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen only for this {@link KStream}s.
+     * Repartitioning can happen only for this {@code KStream}s.
      * For this case, all data of the stream will be redistributed through the repartitioning topic by writing all
-     * records to it, and rereading all records from it, such that the join input {@link KStream} is partitioned
+     * records to it, and rereading all records from it, such that the join input {@code KStream} is partitioned
      * correctly on its key.
      *
      * @param table    the {@link KTable} to be joined with this stream
@@ -1651,8 +1651,8 @@ <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,
      *                 if not specified the default serdes defined in the configs will be used
      * @param <VT>     the value type of the table
      * @param <VR>     the value type of the result stream
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
-     * {@link ValueJoiner}, one output for each input {@link KStream} record
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
+     * {@link ValueJoiner}, one output for each input {@code KStream} record
      * @see #join(KTable, ValueJoiner, Serde, Serde)
      * @see #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
      */
@@ -1665,17 +1665,17 @@ <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,
      * Join records of this stream with {@link GlobalKTable}'s records using non-windowed inner equi join.
      * The join is a primary key table lookup join with join attribute
      * {@code keyValueMapper.map(stream.keyValue) == table.key}.
-     * "Table lookup join" means, that results are only computed if {@link KStream} records are processed.
+     * "Table lookup join" means, that results are only computed if {@code KStream} records are processed.
      * This is done by performing a lookup for matching records in the <em>current</em> internal {@link GlobalKTable}
      * state.
      * In contrast, processing {@link GlobalKTable} input records will only update the internal {@link GlobalKTable}
      * state and will not produce any result records.
      * <p>
-     * For each {@link KStream} record that finds a corresponding record in {@link GlobalKTable} the provided
+     * For each {@code KStream} record that finds a corresponding record in {@link GlobalKTable} the provided
      * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
-     * The key of the result record is the same as the key of this {@link KStream}.
-     * If an {@link KStream} input record key or value is {@code null} the record will not be included in the join
-     * operation and thus no output record will be added to the resulting {@link KStream}.
+     * The key of the result record is the same as the key of this {@code KStream}.
+     * If an {@code KStream} input record key or value is {@code null} the record will not be included in the join
+     * operation and thus no output record will be added to the resulting {@code KStream}.
      *
      * @param globalKTable   the {@link GlobalKTable} to be joined with this stream
      * @param keyValueMapper instance of {@link KeyValueMapper} used to map from the (key, value) of this stream
@@ -1683,9 +1683,9 @@ <VT, VR> KStream<K, VR> leftJoin(final KTable<K, VT> table,
      * @param joiner         a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param <GK>           the key type of {@link GlobalKTable}
      * @param <GV>           the value type of the {@link GlobalKTable}
-     * @param <RV>           the value type of the resulting {@link KStream}
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
-     * {@link ValueJoiner}, one output for each input {@link KStream} record
+     * @param <RV>           the value type of the resulting {@code KStream}
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
+     * {@link ValueJoiner}, one output for each input {@code KStream} record
      * @see #leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
      */
     <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalKTable,
@@ -1698,19 +1698,19 @@ <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalKTable,
      * will produce an output record (cf. below).
      * The join is a primary key table lookup join with join attribute
      * {@code keyValueMapper.map(stream.keyValue) == table.key}.
-     * "Table lookup join" means, that results are only computed if {@link KStream} records are processed.
+     * "Table lookup join" means, that results are only computed if {@code KStream} records are processed.
      * This is done by performing a lookup for matching records in the <em>current</em> internal {@link GlobalKTable}
      * state.
      * In contrast, processing {@link GlobalKTable} input records will only update the internal {@link GlobalKTable}
      * state and will not produce any result records.
      * <p>
-     * For each {@link KStream} record whether or not it finds a corresponding record in {@link GlobalKTable} the
+     * For each {@code KStream} record whether or not it finds a corresponding record in {@link GlobalKTable} the
      * provided {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * If no {@link GlobalKTable} record was found during lookup, a {@code null} value will be provided to
      * {@link ValueJoiner}.
-     * The key of the result record is the same as this {@link KStream}.
-     * If an {@link KStream} input record key or value is {@code null} the record will not be included in the join
-     * operation and thus no output record will be added to the resulting {@link KStream}.
+     * The key of the result record is the same as this {@code KStream}.
+     * If an {@code KStream} input record key or value is {@code null} the record will not be included in the join
+     * operation and thus no output record will be added to the resulting {@code KStream}.
      *
      * @param globalKTable   the {@link GlobalKTable} to be joined with this stream
      * @param keyValueMapper instance of {@link KeyValueMapper} used to map from the (key, value) of this stream
@@ -1718,9 +1718,9 @@ <GK, GV, RV> KStream<K, RV> join(final GlobalKTable<GK, GV> globalKTable,
      * @param valueJoiner    a {@link ValueJoiner} that computes the join result for a pair of matching records
      * @param <GK>           the key type of {@link GlobalKTable}
      * @param <GV>           the value type of the {@link GlobalKTable}
-     * @param <RV>           the value type of the resulting {@link KStream}
-     * @return a {@link KStream} that contains join-records for each key and values computed by the given
-     * {@link ValueJoiner}, one output for each input {@link KStream} record
+     * @param <RV>           the value type of the resulting {@code KStream}
+     * @return a {@code KStream} that contains join-records for each key and values computed by the given
+     * {@link ValueJoiner}, one output for each input {@code KStream} record
      * @see #join(GlobalKTable, KeyValueMapper, ValueJoiner)
      */
     <GK, GV, RV> KStream<K, RV> leftJoin(final GlobalKTable<GK, GV> globalKTable,
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
index 8f67d457156bb..7c64e6fc52e1a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
@@ -40,7 +40,7 @@
 import java.util.regex.Pattern;
 
 /**
- * {@link KStreamBuilder} provide the high-level Kafka Streams DSL to specify a Kafka Streams topology.
+ * {@code KStreamBuilder} provide the high-level Kafka Streams DSL to specify a Kafka Streams topology.
  *
  * @see TopologyBuilder
  * @see KStream
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
index bfb470a2d71b9..4b703979b5bf2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
@@ -29,17 +29,17 @@
 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
 
 /**
- * {@link KTable} is an abstraction of a <i>changelog stream</i> from a primary-keyed table.
+ * {@code KTable} is an abstraction of a <i>changelog stream</i> from a primary-keyed table.
  * Each record in this changelog stream is an update on the primary-keyed table with the record key as the primary key.
  * <p>
- * A {@link KTable} is either {@link KStreamBuilder#table(String, String) defined from a single Kafka topic} that is
- * consumed message by message or the result of a {@link KTable} transformation.
- * An aggregation of a {@link KStream} also yields a {@link KTable}.
+ * A {@code KTable} is either {@link KStreamBuilder#table(String, String) defined from a single Kafka topic} that is
+ * consumed message by message or the result of a {@code KTable} transformation.
+ * An aggregation of a {@link KStream} also yields a {@code KTable}.
  * <p>
- * A {@link KTable} can be transformed record by record, joined with another {@link KTable} or {@link KStream}, or
- * can be re-partitioned and aggregated into a new {@link KTable}.
+ * A {@code KTable} can be transformed record by record, joined with another {@code KTable} or {@link KStream}, or
+ * can be re-partitioned and aggregated into a new {@code KTable}.
  * <p>
- * Some {@link KTable}s have an internal state (a {@link ReadOnlyKeyValueStore}) and are therefore queryable via the
+ * Some {@code KTable}s have an internal state (a {@link ReadOnlyKeyValueStore}) and are therefore queryable via the
  * interactive queries API.
  * For example:
  * <pre>{@code
@@ -64,11 +64,11 @@
 public interface KTable<K, V> {
 
     /**
-     * Create a new {@link KTable} that consists of all records of this {@link KTable} which satisfy the given
+     * Create a new {@code KTable} that consists of all records of this {@code KTable} which satisfy the given
      * predicate.
      * All records that do not satisfy the predicate are dropped.
-     * For each {@link KTable} update the filter is evaluated on the update record to produce an update record for the
-     * result {@link KTable}.
+     * For each {@code KTable} update the filter is evaluated on the update record to produce an update record for the
+     * result {@code KTable}.
      * This is a stateless record-by-record operation.
      * <p>
      * Note that {@code filter} for a <i>changelog stream</i> works different to {@link KStream#filter(Predicate)
@@ -80,17 +80,17 @@ public interface KTable<K, V> {
      * is forwarded.
      *
      * @param predicate a filter {@link Predicate} that is applied to each record
-     * @return a {@link KTable} that contains only those records that satisfy the given predicate
+     * @return a {@code KTable} that contains only those records that satisfy the given predicate
      * @see #filterNot(Predicate)
      */
     KTable<K, V> filter(final Predicate<? super K, ? super V> predicate);
 
     /**
-     * Create a new {@link KTable} that consists all records of this {@link KTable} which do <em>not</em> satisfy the
+     * Create a new {@code KTable} that consists all records of this {@code KTable} which do <em>not</em> satisfy the
      * given predicate.
      * All records that <em>do</em> satisfy the predicate are dropped.
-     * For each {@link KTable} update the filter is evaluated on the update record to produce an update record for the
-     * result {@link KTable}.
+     * For each {@code KTable} update the filter is evaluated on the update record to produce an update record for the
+     * result {@code KTable}.
      * This is a stateless record-by-record operation.
      * <p>
      * Note that {@code filterNot} for a <i>changelog stream</i> works different to {@link KStream#filterNot(Predicate)
@@ -102,16 +102,16 @@ public interface KTable<K, V> {
      * forwarded.
      *
      * @param predicate a filter {@link Predicate} that is applied to each record
-     * @return a {@link KTable} that contains only those records that do <em>not</em> satisfy the given predicate
+     * @return a {@code KTable} that contains only those records that do <em>not</em> satisfy the given predicate
      * @see #filter(Predicate)
      */
     KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate);
 
     /**
-     * Create a new {@link KTable} by transforming the value of each record in this {@link KTable} into a new value
-     * (with possible new type)in the new {@link KTable}.
-     * For each {@link KTable} update the provided {@link ValueMapper} is applied to the value of the update record and
-     * computes a new value for it, resulting in an update record for the result {@link KTable}.
+     * Create a new {@code KTable} by transforming the value of each record in this {@code KTable} into a new value
+     * (with possible new type)in the new {@code KTable}.
+     * For each {@code KTable} update the provided {@link ValueMapper} is applied to the value of the update record and
+     * computes a new value for it, resulting in an update record for the result {@code KTable}.
      * Thus, an input record {@code <K,V>} can be transformed into an output record {@code <K:V'>}.
      * This is a stateless record-by-record operation.
      * <p>
@@ -127,23 +127,23 @@ public interface KTable<K, V> {
      * <p>
      * This operation preserves data co-location with respect to the key.
      * Thus, <em>no</em> internal data redistribution is required if a key based operator (like a join) is applied to
-     * the result {@link KTable}.
+     * the result {@code KTable}.
      * <p>
      * Note that {@code mapValues} for a <i>changelog stream</i> works different to {@link KStream#mapValues(ValueMapper)
      * record stream filters}, because {@link KeyValue records} with {@code null} values (so-called tombstone records)
      * have delete semantics.
      * Thus, for tombstones the provided value-mapper is not evaluated but the tombstone record is forwarded directly to
-     * delete the corresponding record in the result {@link KTable}.
+     * delete the corresponding record in the result {@code KTable}.
      *
      * @param mapper a {@link ValueMapper} that computes a new output value
-     * @param <VR>   the value type of the result {@link KTable}
-     * @return a {@link KTable} that contains records with unmodified keys and new values (possibly of different type)
+     * @param <VR>   the value type of the result {@code KTable}
+     * @return a {@code KTable} that contains records with unmodified keys and new values (possibly of different type)
      */
     <VR> KTable<K, VR> mapValues(final ValueMapper<? super V, ? extends VR> mapper);
 
 
     /**
-     * Print the update records of this {@link KTable} to {@code System.out}.
+     * Print the update records of this {@code KTable} to {@code System.out}.
      * This function will use the generated name of the parent processor node to label the key/value pairs printed to
      * the console.
      * <p>
@@ -153,13 +153,13 @@ public interface KTable<K, V> {
      * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
      * {@link Integer} etc. to get meaningful information.
      * <p>
-     * Note that {@code print()} is not applied to the internal state store and only called for each new {@link KTable}
+     * Note that {@code print()} is not applied to the internal state store and only called for each new {@code KTable}
      * update record.
      */
     void print();
 
     /**
-     * Print the update records of this {@link KTable} to {@code System.out}.
+     * Print the update records of this {@code KTable} to {@code System.out}.
      * This function will use the given name to label the key/value pairs printed to the console.
      * <p>
      * The provided serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
@@ -168,7 +168,7 @@ public interface KTable<K, V> {
      * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
      * {@link Integer} etc. to get meaningful information.
      * <p>
-     * Note that {@code print()} is not applied to the internal state store and only called for each new {@link KTable}
+     * Note that {@code print()} is not applied to the internal state store and only called for each new {@code KTable}
      * update record.
      *
      * @param streamName the name used to label the key/value pairs printed to the console
@@ -176,7 +176,7 @@ public interface KTable<K, V> {
     void print(final String streamName);
 
     /**
-     * Print the update records of this {@link KTable} to {@code System.out}.
+     * Print the update records of this {@code KTable} to {@code System.out}.
      * This function will use the generated name of the parent processor node to label the key/value pairs printed to
      * the console.
      * <p>
@@ -186,7 +186,7 @@ public interface KTable<K, V> {
      * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
      * {@link Integer} etc. to get meaningful information.
      * <p>
-     * Note that {@code print()} is not applied to the internal state store and only called for each new {@link KTable}
+     * Note that {@code print()} is not applied to the internal state store and only called for each new {@code KTable}
      * update record.
      *
      * @param keySerde key serde used to deserialize key if type is {@code byte[]},
@@ -196,7 +196,7 @@ void print(final Serde<K> keySerde,
                final Serde<V> valSerde);
 
     /**
-     * Print the update records of this {@link KTable} to {@code System.out}.
+     * Print the update records of this {@code KTable} to {@code System.out}.
      * This function will use the given name to label the key/value pairs printed to the console.
      * <p>
      * The provided serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
@@ -205,7 +205,7 @@ void print(final Serde<K> keySerde,
      * Implementors will need to override {@code toString()} for keys and values that are not of type {@link String},
      * {@link Integer} etc. to get meaningful information.
      * <p>
-     * Note that {@code print()} is not applied to the internal state store and only called for each new {@link KTable}
+     * Note that {@code print()} is not applied to the internal state store and only called for each new {@code KTable}
      * update record.
      *
      * @param keySerde   key serde used to deserialize key if type is {@code byte[]},
@@ -217,7 +217,7 @@ void print(final Serde<K> keySerde,
                final String streamName);
 
     /**
-     * Write the update records of this {@link KTable} to a file at the given path.
+     * Write the update records of this {@code KTable} to a file at the given path.
      * This function will use the generated name of the parent processor node to label the key/value pairs printed to
      * the file.
      * <p>
@@ -228,14 +228,14 @@ void print(final Serde<K> keySerde,
      * {@link Integer} etc. to get meaningful information.
      * <p>
      * Note that {@code writeAsText()} is not applied to the internal state store and only called for each new
-     * {@link KTable} update record.
+     * {@code KTable} update record.
      *
      * @param filePath name of file to write to
      */
     void writeAsText(final String filePath);
 
     /**
-     * Write the update records of this {@link KTable} to a file at the given path.
+     * Write the update records of this {@code KTable} to a file at the given path.
      * This function will use the given name to label the key/value printed to the file.
      * <p>
      * The default serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
@@ -245,7 +245,7 @@ void print(final Serde<K> keySerde,
      * {@link Integer} etc. to get meaningful information.
      * <p>
      * Note that {@code writeAsText()} is not applied to the internal state store and only called for each new
-     * {@link KTable} update record.
+     * {@code KTable} update record.
      *
      * @param filePath   name of file to write to
      * @param streamName the name used to label the key/value pairs printed out to the console
@@ -254,7 +254,7 @@ void writeAsText(final String filePath,
                      final String streamName);
 
     /**
-     * Write the update records of this {@link KTable} to a file at the given path.
+     * Write the update records of this {@code KTable} to a file at the given path.
      * This function will use the generated name of the parent processor node to label the key/value pairs printed to
      * the file.
      * <p>
@@ -265,7 +265,7 @@ void writeAsText(final String filePath,
      * {@link Integer} etc. to get meaningful information.
      * <p>
      * Note that {@code writeAsText()} is not applied to the internal state store and only called for each new
-     * {@link KTable} update record.
+     * {@code KTable} update record.
      *
      * @param filePath name of file to write to
      * @param keySerde key serde used to deserialize key if type is {@code byte[]},
@@ -276,7 +276,7 @@ void  writeAsText(final String filePath,
                       final Serde<V> valSerde);
 
     /**
-     * Write the update records of this {@link KTable} to a file at the given path.
+     * Write the update records of this {@code KTable} to a file at the given path.
      * This function will use the given name to label the key/value printed to the file.
      * <p>
      * The default serde will be used to deserialize the key or value in case the type is {@code byte[]} before calling
@@ -286,7 +286,7 @@ void  writeAsText(final String filePath,
      * {@link Integer} etc. to get meaningful information.
      * <p>
      * Note that {@code writeAsText()} is not applied to the internal state store and only called for each new
-     * {@link KTable} update record.
+     * {@code KTable} update record.
      *
      * @param filePath name of file to write to
      * @param streamName the name used to label the key/value pairs printed to the console
@@ -299,11 +299,11 @@ void writeAsText(final String filePath,
                      final Serde<V> valSerde);
 
     /**
-     * Perform an action on each update record of this {@link KTable}.
+     * Perform an action on each update record of this {@code KTable}.
      * Note that this is a terminal operation that returns void.
      * <p>
      * Note that {@code foreach()} is not applied to the internal state store and only called for each new
-     * {@link KTable} update record.
+     * {@code KTable} update record.
      *
      * @param action an action to perform on each record
      */
@@ -313,9 +313,9 @@ void writeAsText(final String filePath,
      * Convert this changelog stream to a {@link KStream}.
      * <p>
      * Note that this is a logical operation and only changes the "interpretation" of the stream, i.e., each record of
-     * this changelog stream is no longer treated as an update record (cf. {@link KStream} vs {@link KTable}).
+     * this changelog stream is no longer treated as an update record (cf. {@link KStream} vs {@code KTable}).
      *
-     * @return a {@link KStream} that contains the same records as this {@link KTable}
+     * @return a {@link KStream} that contains the same records as this {@code KTable}
      */
     KStream<K, V> toStream();
 
@@ -338,16 +338,16 @@ void writeAsText(final String filePath,
      * {@code table.}{@link #toStream() toStream}{@code ().}{@link KStream#selectKey(KeyValueMapper) selectKey(KeyValueMapper)}.
      * <p>
      * Note that {@link #toStream()} is a logical operation and only changes the "interpretation" of the stream, i.e.,
-     * each record of this changelog stream is no longer treated as an update record (cf. {@link KStream} vs {@link KTable}).
+     * each record of this changelog stream is no longer treated as an update record (cf. {@link KStream} vs {@code KTable}).
      *
      * @param mapper a {@link KeyValueMapper} that computes a new key for each record
      * @param <KR> the new key type of the result stream
-     * @return a {@link KStream} that contains the same records as this {@link KTable}
+     * @return a {@link KStream} that contains the same records as this {@code KTable}
      */
     <KR> KStream<KR, V> toStream(final KeyValueMapper<? super K, ? super V, ? extends KR> mapper);
 
     /**
-     * Materialize this changelog stream to a topic and creates a new {@link KTable} from the topic using default
+     * Materialize this changelog stream to a topic and creates a new {@code KTable} from the topic using default
      * serializers and deserializers and producer's {@link DefaultPartitioner}.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
@@ -355,18 +355,18 @@ void writeAsText(final String filePath,
      * This is equivalent to calling {@link #to(String) #to(someTopicName)} and
      * {@link KStreamBuilder#table(String, String) KStreamBuilder#table(someTopicName, queryableStoreName)}.
      * <p>
-     * The resulting {@link KTable} will be materialized in a local state store with the given store name (cf.
+     * The resulting {@code KTable} will be materialized in a local state store with the given store name (cf.
      * {@link KStreamBuilder#table(String, String)})
      *
      * @param topic     the topic name
-     * @param storeName the state store name used for the result {@link KTable}
-     * @return a {@link KTable} that contains the exact same (and potentially repartitioned) records as this {@link KTable}
+     * @param storeName the state store name used for the result {@code KTable}
+     * @return a {@code KTable} that contains the exact same (and potentially repartitioned) records as this {@code KTable}
      */
     KTable<K, V> through(final String topic,
                          final String storeName);
 
     /**
-     * Materialize this changelog stream to a topic and creates a new {@link KTable} from the topic using default
+     * Materialize this changelog stream to a topic and creates a new {@code KTable} from the topic using default
      * serializers and deserializers and a customizable {@link StreamPartitioner} to determine the distribution of
      * records to partitions.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
@@ -375,21 +375,21 @@ KTable<K, V> through(final String topic,
      * This is equivalent to calling {@link #to(StreamPartitioner, String) #to(partitioner, someTopicName)} and
      * {@link KStreamBuilder#table(String, String) KStreamBuilder#table(someTopicName, queryableStoreName)}.
      * <p>
-     * The resulting {@link KTable} will be materialized in a local state store with the given store name (cf.
+     * The resulting {@code KTable} will be materialized in a local state store with the given store name (cf.
      * {@link KStreamBuilder#table(String, String)})
      *
      * @param partitioner the function used to determine how records are distributed among partitions of the topic,
      *                    if not specified producer's {@link DefaultPartitioner} will be used
      * @param topic       the topic name
-     * @param storeName   the state store name used for the result {@link KTable}
-     * @return a {@link KTable} that contains the exact same (and potentially repartitioned) records as this {@link KTable}
+     * @param storeName   the state store name used for the result {@code KTable}
+     * @return a {@code KTable} that contains the exact same (and potentially repartitioned) records as this {@code KTable}
      */
     KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
                          final String topic,
                          final String storeName);
 
     /**
-     * Materialize this changelog stream to a topic and creates a new {@link KTable} from the topic.
+     * Materialize this changelog stream to a topic and creates a new {@code KTable} from the topic.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
      * <p>
@@ -399,7 +399,7 @@ KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
      * This is equivalent to calling {@link #to(Serde, Serde, String) #to(keySerde, valueSerde, someTopicName)} and
      * {@link KStreamBuilder#table(String, String) KStreamBuilder#table(someTopicName, queryableStoreName)}.
      * <p>
-     * The resulting {@link KTable} will be materialized in a local state store with the given store name (cf.
+     * The resulting {@code KTable} will be materialized in a local state store with the given store name (cf.
      * {@link KStreamBuilder#table(String, String)})
      *
      * @param keySerde  key serde used to send key-value pairs,
@@ -407,15 +407,15 @@ KTable<K, V> through(final StreamPartitioner<? super K, ? super V> partitioner,
      * @param valSerde  value serde used to send key-value pairs,
      *                  if not specified the default value serde defined in the configuration will be used
      * @param topic     the topic name
-     * @param storeName the state store name used for the result {@link KTable}
-     * @return a {@link KTable} that contains the exact same (and potentially repartitioned) records as this {@link KTable}
+     * @param storeName the state store name used for the result {@code KTable}
+     * @return a {@code KTable} that contains the exact same (and potentially repartitioned) records as this {@code KTable}
      */
     KTable<K, V> through(final Serde<K> keySerde, Serde<V> valSerde,
                          final String topic,
                          final String storeName);
 
     /**
-     * Materialize this changelog stream to a topic and creates a new {@link KTable} from the topic using a customizable
+     * Materialize this changelog stream to a topic and creates a new {@code KTable} from the topic using a customizable
      * {@link StreamPartitioner} to determine the distribution of records to partitions.
      * The specified topic should be manually created before it is used (i.e., before the Kafka Streams application is
      * started).
@@ -424,7 +424,7 @@ KTable<K, V> through(final Serde<K> keySerde, Serde<V> valSerde,
      * #to(keySerde, valueSerde, partitioner, someTopicName)} and
      * {@link KStreamBuilder#table(String, String) KStreamBuilder#table(someTopicName, queryableStoreName)}.
      * <p>
-     * The resulting {@link KTable} will be materialized in a local state store with the given store name (cf.
+     * The resulting {@code KTable} will be materialized in a local state store with the given store name (cf.
      * {@link KStreamBuilder#table(String, String)})
      *
      * @param keySerde    key serde used to send key-value pairs,
@@ -436,8 +436,8 @@ KTable<K, V> through(final Serde<K> keySerde, Serde<V> valSerde,
      *                    {@link WindowedStreamPartitioner} will be used&mdash;otherwise {@link DefaultPartitioner} will
      *                    be used
      * @param topic      the topic name
-     * @param storeName  the state store name used for the result {@link KTable}
-     * @return a {@link KTable} that contains the exact same (and potentially repartitioned) records as this {@link KTable}
+     * @param storeName  the state store name used for the result {@code KTable}
+     * @return a {@code KTable} that contains the exact same (and potentially repartitioned) records as this {@code KTable}
      */
     KTable<K, V> through(final Serde<K> keySerde,
                          final Serde<V> valSerde,
@@ -508,11 +508,11 @@ void to(final Serde<K> keySerde,
             final String topic);
 
     /**
-     * Re-groups the records of this {@link KTable} using the provided {@link KeyValueMapper} and default serializers
+     * Re-groups the records of this {@code KTable} using the provided {@link KeyValueMapper} and default serializers
      * and deserializers.
-     * Each {@link KeyValue} pair of this {@link KTable} is mapped to a new {@link KeyValue} pair by applying the
+     * Each {@link KeyValue} pair of this {@code KTable} is mapped to a new {@link KeyValue} pair by applying the
      * provided {@link KeyValueMapper}.
-     * Re-grouping a {@link KTable} is required before an aggregation operator can be applied to the data
+     * Re-grouping a {@code KTable} is required before an aggregation operator can be applied to the data
      * (cf. {@link KGroupedTable}).
      * The {@link KeyValueMapper} selects a new key and value (with should both have unmodified type).
      * If the new record key is {@code null} the record will not be included in the resulting {@link KGroupedTable}
@@ -523,7 +523,7 @@ void to(final Serde<K> keySerde,
      * an internally generated name, and "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * All data of this {@link KTable} will be redistributed through the repartitioning topic by writing all update
+     * All data of this {@code KTable} will be redistributed through the repartitioning topic by writing all update
      * records to and rereading all update records from it, such that the resulting {@link KGroupedTable} is partitioned
      * on the new key.
      * <p>
@@ -533,15 +533,15 @@ void to(final Serde<K> keySerde,
      * @param selector a {@link KeyValueMapper} that computes a new grouping key and value to be aggregated
      * @param <KR>     the key type of the result {@link KGroupedTable}
      * @param <VR>     the value type of the result {@link KGroupedTable}
-     * @return a {@link KGroupedTable} that contains the re-grouped records of the original {@link KTable}
+     * @return a {@link KGroupedTable} that contains the re-grouped records of the original {@code KTable}
      */
     <KR, VR> KGroupedTable<KR, VR> groupBy(final KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector);
 
     /**
-     * Re-groups the records of this {@link KTable} using the provided {@link KeyValueMapper}.
-     * Each {@link KeyValue} pair of this {@link KTable} is mapped to a new {@link KeyValue} pair by applying the
+     * Re-groups the records of this {@code KTable} using the provided {@link KeyValueMapper}.
+     * Each {@link KeyValue} pair of this {@code KTable} is mapped to a new {@link KeyValue} pair by applying the
      * provided {@link KeyValueMapper}.
-     * Re-grouping a {@link KTable} is required before an aggregation operator can be applied to the data
+     * Re-grouping a {@code KTable} is required before an aggregation operator can be applied to the data
      * (cf. {@link KGroupedTable}).
      * The {@link KeyValueMapper} selects a new key and value (both with potentially different type).
      * If the new record key is {@code null} the record will not be included in the resulting {@link KGroupedTable}
@@ -552,7 +552,7 @@ void to(final Serde<K> keySerde,
      * an internally generated name, and "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * All data of this {@link KTable} will be redistributed through the repartitioning topic by writing all update
+     * All data of this {@code KTable} will be redistributed through the repartitioning topic by writing all update
      * records to and rereading all update records from it, such that the resulting {@link KGroupedTable} is partitioned
      * on the new key.
      *
@@ -563,32 +563,32 @@ void to(final Serde<K> keySerde,
      *                   if not specified the default serdes defined in the configs will be used
      * @param <KR>       the key type of the result {@link KGroupedTable}
      * @param <VR>       the value type of the result {@link KGroupedTable}
-     * @return a {@link KGroupedTable} that contains the re-grouped records of the original {@link KTable}
+     * @return a {@link KGroupedTable} that contains the re-grouped records of the original {@code KTable}
      */
     <KR, VR> KGroupedTable<KR, VR> groupBy(final KeyValueMapper<? super K, ? super V, KeyValue<KR, VR>> selector,
                                            final Serde<KR> keySerde,
                                            final Serde<VR> valueSerde);
 
     /**
-     * Join records of this {@link KTable} with another {@link KTable}'s records using non-windowed inner equi join.
+     * Join records of this {@code KTable} with another {@code KTable}'s records using non-windowed inner equi join.
      * The join is a primary key join with join attribute {@code thisKTable.key == otherKTable.key}.
-     * The result is an ever updating {@link KTable} that represents the <em>current</em> (i.e., processing time) result
+     * The result is an ever updating {@code KTable} that represents the <em>current</em> (i.e., processing time) result
      * of the join.
      * <p>
-     * The join is computed by (1) updating the internal state of one {@link KTable} and (2) performing a lookup for a
-     * matching record in the <em>current</em> (i.e., processing time) internal state of the other {@link KTable}.
+     * The join is computed by (1) updating the internal state of one {@code KTable} and (2) performing a lookup for a
+     * matching record in the <em>current</em> (i.e., processing time) internal state of the other {@code KTable}.
      * This happens in a symmetric way, i.e., for each update of either {@code this} or the {@code other} input
-     * {@link KTable} the result gets updated.
+     * {@code KTable} the result gets updated.
      * <p>
-     * For each {@link KTable} record that finds a corresponding record in the other {@link KTable} the provided
+     * For each {@code KTable} record that finds a corresponding record in the other {@code KTable} the provided
      * {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
      * <p>
      * Note that {@link KeyValue records} with {@code null} values (so-called tombstone records) have delete semantics.
      * Thus, for input tombstones the provided value-joiner is not called but a tombstone record is forwarded
-     * directly to delete a record in the result {@link KTable} if required (i.e., if there is anything to be deleted).
+     * directly to delete a record in the result {@code KTable} if required (i.e., if there is anything to be deleted).
      * <p>
-     * A {@link KTable} input record key cannot be {@code null}.
+     * A {@code KTable} input record key cannot be {@code null}.
      * <p>
      * Example:
      * <table border='1'>
@@ -637,16 +637,16 @@ <KR, VR> KGroupedTable<KR, VR> groupBy(final KeyValueMapper<? super K, ? super V
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen both input {@link KTable}s.
-     * For this case, all data of a {@link KTable} will be redistributed through the repartitioning topic by writing all
-     * update records to and rereading all update records from it, such that the join input {@link KTable} is
+     * Repartitioning can happen both input {@code KTable}s.
+     * For this case, all data of a {@code KTable} will be redistributed through the repartitioning topic by writing all
+     * update records to and rereading all update records from it, such that the join input {@code KTable} is
      * partitioned correctly on its key.
      *
-     * @param other  the other {@link KTable} to be joined with this {@link KTable}
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}
      * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
-     * @param <VO>   the value type of the other {@link KTable}
-     * @param <VR>   the value type of the result {@link KTable}
-     * @return a {@link KTable} that contains join-records for each key and values computed by the given
+     * @param <VO>   the value type of the other {@code KTable}
+     * @param <VR>   the value type of the result {@code KTable}
+     * @return a {@code KTable} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key
      * @see #leftJoin(KTable, ValueJoiner)
      * @see #outerJoin(KTable, ValueJoiner)
@@ -655,32 +655,32 @@ <VO, VR> KTable<K, VR> join(final KTable<K, VO> other,
                                 final ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     /**
-     * Join records of this {@link KTable} (left input) with another {@link KTable}'s (right input) records using
+     * Join records of this {@code KTable} (left input) with another {@code KTable}'s (right input) records using
      * non-windowed left equi join.
      * The join is a primary key join with join attribute {@code thisKTable.key == otherKTable.key}.
-     * In contrast to {@link #join(KTable, ValueJoiner) inner-join}, all records from left {@link KTable} will produce
+     * In contrast to {@link #join(KTable, ValueJoiner) inner-join}, all records from left {@code KTable} will produce
      * an output record (cf. below).
-     * The result is an ever updating {@link KTable} that represents the <em>current</em> (i.e., processing time) result
+     * The result is an ever updating {@code KTable} that represents the <em>current</em> (i.e., processing time) result
      * of the join.
      * <p>
-     * The join is computed by (1) updating the internal state of one {@link KTable} and (2) performing a lookup for a
-     * matching record in the <em>current</em> (i.e., processing time) internal state of the other {@link KTable}.
+     * The join is computed by (1) updating the internal state of one {@code KTable} and (2) performing a lookup for a
+     * matching record in the <em>current</em> (i.e., processing time) internal state of the other {@code KTable}.
      * This happens in a symmetric way, i.e., for each update of either {@code this} or the {@code other} input
-     * {@link KTable} the result gets updated.
+     * {@code KTable} the result gets updated.
      * <p>
-     * For each {@link KTable} record that finds a corresponding record in the other {@link KTable}'s state the
+     * For each {@code KTable} record that finds a corresponding record in the other {@code KTable}'s state the
      * provided {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
-     * Additionally, for each record of left {@link KTable} that does not find a corresponding record in the
-     * right {@link KTable}'s state the provided {@link ValueJoiner} will be called with {@code rightValue =
+     * Additionally, for each record of left {@code KTable} that does not find a corresponding record in the
+     * right {@code KTable}'s state the provided {@link ValueJoiner} will be called with {@code rightValue =
      * null} to compute a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
      * <p>
      * Note that {@link KeyValue records} with {@code null} values (so-called tombstone records) have delete semantics.
      * For example, for left input tombstones the provided value-joiner is not called but a tombstone record is
-     * forwarded directly to delete a record in the result {@link KTable} if required (i.e., if there is anything to be
+     * forwarded directly to delete a record in the result {@code KTable} if required (i.e., if there is anything to be
      * deleted).
      * <p>
-     * A {@link KTable} input record key cannot be {@code null}.
+     * A {@code KTable} input record key cannot be {@code null}.
      * <p>
      * Example:
      * <table border='1'>
@@ -729,18 +729,18 @@ <VO, VR> KTable<K, VR> join(final KTable<K, VO> other,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen both input {@link KTable}s.
-     * For this case, all data of a {@link KTable} will be redistributed through the repartitioning topic by writing all
-     * update records to and rereading all update records from it, such that the join input {@link KTable} is
+     * Repartitioning can happen both input {@code KTable}s.
+     * For this case, all data of a {@code KTable} will be redistributed through the repartitioning topic by writing all
+     * update records to and rereading all update records from it, such that the join input {@code KTable} is
      * partitioned correctly on its key.
      *
-     * @param other  the other {@link KTable} to be joined with this {@link KTable}
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}
      * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
-     * @param <VO>   the value type of the other {@link KTable}
-     * @param <VR>   the value type of the result {@link KTable}
-     * @return a {@link KTable} that contains join-records for each key and values computed by the given
+     * @param <VO>   the value type of the other {@code KTable}
+     * @param <VR>   the value type of the result {@code KTable}
+     * @return a {@code KTable} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key plus one for each non-matching record of
-     * left {@link KTable}
+     * left {@code KTable}
      * @see #join(KTable, ValueJoiner)
      * @see #outerJoin(KTable, ValueJoiner)
      */
@@ -748,31 +748,31 @@ <VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other,
                                     final ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     /**
-     * Join records of this {@link KTable} (left input) with another {@link KTable}'s (right input) records using
+     * Join records of this {@code KTable} (left input) with another {@code KTable}'s (right input) records using
      * non-windowed outer equi join.
      * The join is a primary key join with join attribute {@code thisKTable.key == otherKTable.key}.
      * In contrast to {@link #join(KTable, ValueJoiner) inner-join} or {@link #leftJoin(KTable, ValueJoiner) left-join},
-     * all records from both input {@link KTable}s will produce an output record (cf. below).
-     * The result is an ever updating {@link KTable} that represents the <em>current</em> (i.e., processing time) result
+     * all records from both input {@code KTable}s will produce an output record (cf. below).
+     * The result is an ever updating {@code KTable} that represents the <em>current</em> (i.e., processing time) result
      * of the join.
      * <p>
-     * The join is computed by (1) updating the internal state of one {@link KTable} and (2) performing a lookup for a
-     * matching record in the <em>current</em> (i.e., processing time) internal state of the other {@link KTable}.
+     * The join is computed by (1) updating the internal state of one {@code KTable} and (2) performing a lookup for a
+     * matching record in the <em>current</em> (i.e., processing time) internal state of the other {@code KTable}.
      * This happens in a symmetric way, i.e., for each update of either {@code this} or the {@code other} input
-     * {@link KTable} the result gets updated.
+     * {@code KTable} the result gets updated.
      * <p>
-     * For each {@link KTable} record that finds a corresponding record in the other {@link KTable}'s state the
+     * For each {@code KTable} record that finds a corresponding record in the other {@code KTable}'s state the
      * provided {@link ValueJoiner} will be called to compute a value (with arbitrary type) for the result record.
      * Additionally, for each record that does not find a corresponding record in the corresponding other
-     * {@link KTable}'s state the provided {@link ValueJoiner} will be called with {@code null} value for the
+     * {@code KTable}'s state the provided {@link ValueJoiner} will be called with {@code null} value for the
      * corresponding other value to compute a value (with arbitrary type) for the result record.
      * The key of the result record is the same as for both joining input records.
      * <p>
      * Note that {@link KeyValue records} with {@code null} values (so-called tombstone records) have delete semantics.
      * Thus, for input tombstones the provided value-joiner is not called but a tombstone record is forwarded directly
-     * to delete a record in the result {@link KTable} if required (i.e., if there is anything to bedeleted).
+     * to delete a record in the result {@code KTable} if required (i.e., if there is anything to bedeleted).
      * <p>
-     * A {@link KTable} input record key cannot be {@code null}.
+     * A {@code KTable} input record key cannot be {@code null}.
      * <p>
      * Example:
      * <table border='1'>
@@ -821,18 +821,18 @@ <VO, VR> KTable<K, VR> leftJoin(final KTable<K, VO> other,
      * "-repartition" is a fixed suffix.
      * You can retrieve all generated internal topic names via {@link KafkaStreams#toString()}.
      * <p>
-     * Repartitioning can happen both input {@link KTable}s.
-     * For this case, all data of a {@link KTable} will be redistributed through the repartitioning topic by writing all
-     * update records to and rereading all update records from it, such that the join input {@link KTable} is
+     * Repartitioning can happen both input {@code KTable}s.
+     * For this case, all data of a {@code KTable} will be redistributed through the repartitioning topic by writing all
+     * update records to and rereading all update records from it, such that the join input {@code KTable} is
      * partitioned correctly on its key.
      *
-     * @param other  the other {@link KTable} to be joined with this {@link KTable}
+     * @param other  the other {@code KTable} to be joined with this {@code KTable}
      * @param joiner a {@link ValueJoiner} that computes the join result for a pair of matching records
-     * @param <VO>   the value type of the other {@link KTable}
-     * @param <VR>   the value type of the result {@link KTable}
-     * @return a {@link KTable} that contains join-records for each key and values computed by the given
+     * @param <VO>   the value type of the other {@code KTable}
+     * @param <VR>   the value type of the result {@code KTable}
+     * @return a {@code KTable} that contains join-records for each key and values computed by the given
      * {@link ValueJoiner}, one for each matched record-pair with the same key plus one for each non-matching record of
-     * both {@link KTable}s
+     * both {@code KTable}s
      * @see #join(KTable, ValueJoiner)
      * @see #leftJoin(KTable, ValueJoiner)
      */
@@ -840,9 +840,9 @@ <VO, VR> KTable<K, VR> outerJoin(final KTable<K, VO> other,
                                      final ValueJoiner<? super V, ? super VO, ? extends VR> joiner);
 
     /**
-     * Get the name of the local state store used for materializing this {@link KTable}.
+     * Get the name of the local state store used for materializing this {@code KTable}.
      *
-     * @return the underlying state store name, or {@code null} if this {@link KTable} is not materialized
+     * @return the underlying state store name, or {@code null} if this {@code KTable} is not materialized
      */
     String getStoreName();
 }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java
index 13daca8f41ffe..e2434e37692fb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java
@@ -20,7 +20,7 @@
 import org.apache.kafka.streams.KeyValue;
 
 /**
- * The {@link KeyValueMapper} interface for mapping a {@link KeyValue key-value pair} to a new value of arbitrary type.
+ * The {@code KeyValueMapper} interface for mapping a {@link KeyValue key-value pair} to a new value of arbitrary type.
  * For example, it can be used to
  * <ul>
  * <li>map from an input {@link KeyValue} pair to an output {@link KeyValue} pair with different key and/or value type
@@ -29,7 +29,7 @@
  * </ul>
  * This is a stateless record-by-record operation, i.e, {@link #apply(Object, Object)} is invoked individually for each
  * record of a stream (cf. {@link Transformer} for stateful record transformation).
- * {@link KeyValueMapper} is a generalization of {@link ValueMapper}.
+ * {@code KeyValueMapper} is a generalization of {@link ValueMapper}.
  *
  * @param <K>  key type
  * @param <V>  value type
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java
index ef22f24a62aa9..cf1a1f335b2f2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java
@@ -20,7 +20,7 @@
 import org.apache.kafka.streams.KeyValue;
 
 /**
- * The {@link Predicate} interface represents a predicate (boolean-valued function) of a {@link KeyValue} pair.
+ * The {@code Predicate} interface represents a predicate (boolean-valued function) of a {@link KeyValue} pair.
  * This is a stateless record-by-record operation, i.e, {@link #test(Object, Object)} is invoked individually for each
  * record of a stream.
  *
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java
index 328f6236dcd65..5a273e4ee135f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java
@@ -20,13 +20,13 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
 /**
- * The {@link Reducer} interface for combining two values of the same type into a new value.
+ * The {@code Reducer} interface for combining two values of the same type into a new value.
  * In contrast to {@link Aggregator} the result type must be the same as the input type.
  * <p>
  * The provided values can be either original values from input {@link KeyValue} pair records or be a previously
  * computed result from {@link Reducer#apply(Object, Object)}.
  * <p>
- * {@link Reducer} can be used to implement aggregation functions like sum, min, or max.
+ * {@code Reducer} can be used to implement aggregation functions like sum, min, or max.
  *
  * @param <V> value type
  * @see KGroupedStream#reduce(Reducer, String)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
index 2eee43e7335ed..943092365dffd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
@@ -65,10 +65,10 @@
  * @see KGroupedStream#reduce(Reducer, SessionWindows, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, String)
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, org.apache.kafka.streams.processor.StateStoreSupplier)
- * @see org.apache.kafka.streams.processor.TimestampExtractor
+ * @see TimestampExtractor
  */
 @InterfaceStability.Unstable
-public class SessionWindows {
+public final class SessionWindows {
 
     private final long gapMs;
     private long maintainDurationMs;
@@ -121,7 +121,7 @@ public long inactivityGap() {
     /**
      * Return the window maintain duration (retention time) in milliseconds.
      * <p>
-     * For {@link SessionWindows} the maintain duration is at least as small as the window gap.
+     * For {@code SessionWindows} the maintain duration is at least as small as the window gap.
      *
      * @return the window maintain duration
      */
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
index bcd3d8721ded8..14af875bebcaa 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
@@ -56,7 +56,7 @@
  * @see TimestampExtractor
  */
 @InterfaceStability.Unstable
-public class TimeWindows extends Windows<TimeWindow> {
+public final class TimeWindows extends Windows<TimeWindow> {
 
     /** The size of the windows in milliseconds. */
     public final long sizeMs;
@@ -128,12 +128,11 @@ public long size() {
     }
 
     /**
-     * {@inheritDoc}
-     *
      * @param durationMs the window retention time
      * @return itself
      * @throws IllegalArgumentException if {@code duration} is smaller than the window size
      */
+    @Override
     public TimeWindows until(final long durationMs) throws IllegalArgumentException {
         if (durationMs < sizeMs) {
             throw new IllegalArgumentException("Window retention time (durationMs) cannot be smaller than the window size.");
@@ -145,7 +144,7 @@ public TimeWindows until(final long durationMs) throws IllegalArgumentException
     /**
      * {@inheritDoc}
      * <p>
-     * For {@link TimeWindows} the maintain duration is at least as small as the window size.
+     * For {@code TimeWindows} the maintain duration is at least as small as the window size.
      *
      * @return the window maintain duration
      */
@@ -155,7 +154,7 @@ public long maintainMs() {
     }
 
     @Override
-    public final boolean equals(final Object o) {
+    public boolean equals(final Object o) {
         if (o == this) {
             return true;
         }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java
index d9a814b94c33f..6d18477977e17 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java
@@ -25,7 +25,7 @@
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 /**
- * The {@link Transformer} interface for stateful mapping of an input record to zero, one, or multiple new output
+ * The {@code Transformer} interface for stateful mapping of an input record to zero, one, or multiple new output
  * records (both key and value type can be altered arbitrarily).
  * This is a stateful record-by-record operation, i.e, {@link #transform(Object, Object)} is invoked individually for
  * each record of a stream and can access and modify a state that is available beyond a single call of
@@ -33,7 +33,7 @@
  * Additionally, the interface can be called in regular intervals based on the processing progress
  * (cf. {@link #punctuate(long)}.
  * <p>
- * Use {@link TransformerSupplier} to provide new instances of {@link Transformer} to Kafka Stream's runtime.
+ * Use {@link TransformerSupplier} to provide new instances of {@code Transformer} to Kafka Stream's runtime.
  * <p>
  * If only a record's value should be modified {@link ValueTransformer} can be used.
  *
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java
index 0a0148982528d..f5385aeeba5c2 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java
@@ -20,7 +20,7 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * A {@link TransformerSupplier} interface which can create one or more {@link Transformer} instances.
+ * A {@code TransformerSupplier} interface which can create one or more {@link Transformer} instances.
  *
  * @param <K> key type
  * @param <V> value type
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
index cab614204908d..14598e41e87bb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
@@ -45,7 +45,7 @@
  * @see TimestampExtractor
  */
 @InterfaceStability.Unstable
-public class UnlimitedWindows extends Windows<UnlimitedWindow> {
+public final class UnlimitedWindows extends Windows<UnlimitedWindow> {
 
     private static final long DEFAULT_START_TIMESTAMP_MS = 0L;
 
@@ -123,7 +123,7 @@ public long maintainMs() {
     }
 
     @Override
-    public final boolean equals(final Object o) {
+    public boolean equals(final Object o) {
         if (o == this) {
             return true;
         }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java
index ab91bb4f7e327..94de484c28220 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java
@@ -19,7 +19,7 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * The {@link ValueJoiner} interface for joining two values into a new value of arbitrary type.
+ * The {@code ValueJoiner} interface for joining two values into a new value of arbitrary type.
  * This is a stateless operation, i.e, {@link #apply(Object, Object)} is invoked individually for each joining
  * record-pair of a {@link KStream}-{@link KStream}, {@link KStream}-{@link KTable}, or {@link KTable}-{@link KTable}
  * join.
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java
index 71acc1d13c7c2..63e0cc471aaca 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java
@@ -19,10 +19,10 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * The {@link ValueMapper} interface for mapping a value to a new value of arbitrary type.
+ * The {@code ValueMapper} interface for mapping a value to a new value of arbitrary type.
  * This is a stateless record-by-record operation, i.e, {@link #apply(Object)} is invoked individually for each record
  * of a stream (cf. {@link ValueTransformer} for stateful value transformation).
- * If {@link ValueMapper} is applied to a {@link org.apache.kafka.streams.KeyValue key-value pair} record the record's
+ * If {@code ValueMapper} is applied to a {@link org.apache.kafka.streams.KeyValue key-value pair} record the record's
  * key is preserved.
  * If a record's key and value should be modified {@link KeyValueMapper} can be used.
  *
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java
index 02c3ba7270d62..b2008f5ba1f4d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java
@@ -25,15 +25,15 @@
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 /**
- * The {@link ValueTransformer} interface for stateful mapping of a value to a new value (with possible new type).
+ * The {@code ValueTransformer} interface for stateful mapping of a value to a new value (with possible new type).
  * This is a stateful record-by-record operation, i.e, {@link #transform(Object)} is invoked individually for each
  * record of a stream and can access and modify a state that is available beyond a single call of
  * {@link #transform(Object)} (cf. {@link ValueMapper} for stateless value transformation).
  * Additionally, the interface can be called in regular intervals based on the processing progress
  * (cf. {@link #punctuate(long)}.
- * If {@link ValueTransformer} is applied to a {@link KeyValue} pair record the record's key is preserved.
+ * If {@code ValueTransformer} is applied to a {@link KeyValue} pair record the record's key is preserved.
  * <p>
- * Use {@link ValueTransformerSupplier} to provide new instances of {@link ValueTransformer} to Kafka Stream's runtime.
+ * Use {@link ValueTransformerSupplier} to provide new instances of {@code ValueTransformer} to Kafka Stream's runtime.
  * <p>
  * If a record's key and value should be modified {@link Transformer} can be used.
  *
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
index 3d357ee296141..01c920e95e4e9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
@@ -19,7 +19,7 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * A {@link ValueTransformerSupplier} interface which can create one or more {@link ValueTransformer} instances.
+ * A {@code ValueTransformerSupplier} interface which can create one or more {@link ValueTransformer} instances.
  *
  * @param <V>  value type
  * @param <VR> transformed value type
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
index 5388affee9b90..0c8b3047d5560 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
@@ -21,10 +21,10 @@
 
 /**
  * A single window instance, defined by its start and end timestamp.
- * {@link Window} is agnostic if start/end boundaries are inclusive or exclusive; this is defined by concrete
+ * {@code Window} is agnostic if start/end boundaries are inclusive or exclusive; this is defined by concrete
  * window implementations.
  * <p>
- * To specify how {@link Window} boundaries are defined use {@link Windows}.
+ * To specify how {@code Window} boundaries are defined use {@link Windows}.
  * For time semantics, see {@link TimestampExtractor}.
  *
  * @see Windows
@@ -46,7 +46,7 @@ public abstract class Window {
      * @param endMs   the end timestamp of the window
      * @throws IllegalArgumentException if {@code startMs} is negative or if {@code endMs} is smaller than {@code startMs}
      */
-    public Window(long startMs, long endMs) throws IllegalArgumentException {
+    public Window(final long startMs, final long endMs) throws IllegalArgumentException {
         if (startMs < 0) {
             throw new IllegalArgumentException("Window startMs time cannot be negative.");
         }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java
index 2941136dc9856..63074c37b1593 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java
@@ -25,7 +25,7 @@
  * If a {@link KStream} gets grouped and aggregated using a window-aggregation the resulting {@link KTable} is a
  * so-called "windowed {@link KTable}" with a combined key type that encodes the corresponding aggregation window and
  * the original record key.
- * Thus, a windowed {@link KTable} has type {@link Windowed &lt;Windowed&lt;K&gt;,V&gt;}
+ * Thus, a windowed {@link KTable} has type {@code <Windowed<K>,V>}.
  *
  * @param <K> type of the key
  * @see KGroupedStream#count(Windows, String)