kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: MINOR: Improve JavaDoc for some public classes.
Date Thu, 03 Mar 2016 00:11:23 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 2a58ba9fd -> f676cfeb8


MINOR: Improve JavaDoc for some public classes.

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Yasuhiro Mastuda <yasuhiro.mastuda@gmail.com>

Closes #999 from guozhangwang/KJavaDoc


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

Branch: refs/heads/trunk
Commit: f676cfeb83195f032e3ad8908288a7e04011c830
Parents: 2a58ba9
Author: Guozhang Wang <wangguoz@gmail.com>
Authored: Wed Mar 2 16:11:19 2016 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Wed Mar 2 16:11:19 2016 -0800

----------------------------------------------------------------------
 .../java/org/apache/kafka/streams/KafkaStreams.java | 16 ++++++++--------
 .../org/apache/kafka/streams/StreamsConfig.java     |  4 ++++
 .../org/apache/kafka/streams/StreamsMetrics.java    |  2 +-
 .../org/apache/kafka/streams/kstream/KStream.java   |  8 +++-----
 .../org/apache/kafka/streams/kstream/KTable.java    |  8 +++-----
 .../apache/kafka/streams/kstream/Transformer.java   |  7 +++++++
 .../kafka/streams/kstream/TransformerSupplier.java  |  3 +++
 .../kafka/streams/kstream/UnlimitedWindows.java     |  3 +++
 .../kafka/streams/kstream/ValueTransformer.java     |  6 ++++++
 .../streams/kstream/ValueTransformerSupplier.java   |  3 +++
 .../org/apache/kafka/streams/kstream/Windowed.java  |  9 +++++++++
 .../streams/processor/DefaultPartitionGrouper.java  |  2 +-
 .../kafka/streams/processor/PartitionGrouper.java   |  7 +++++++
 .../kafka/streams/processor/ProcessorSupplier.java  | 10 ++++++++++
 .../kafka/streams/processor/TimestampExtractor.java |  3 ++-
 .../kafka/streams/state/KeyValueIterator.java       |  2 +-
 .../apache/kafka/streams/state/KeyValueStore.java   | 14 +++++++-------
 .../java/org/apache/kafka/streams/state/Serdes.java |  6 ++++++
 .../java/org/apache/kafka/streams/state/Stores.java |  2 +-
 .../kafka/streams/state/WindowStoreIterator.java    |  5 +++++
 .../kafka/streams/state/WindowStoreUtils.java       |  2 +-
 21 files changed, 91 insertions(+), 31 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
----------------------------------------------------------------------
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 724daac..15d6d8b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -55,26 +55,26 @@ import java.util.concurrent.atomic.AtomicInteger;
  * Internally the {@link KafkaStreams} instance contains a normal {@link org.apache.kafka.clients.producer.KafkaProducer
KafkaProducer}
  * and {@link org.apache.kafka.clients.consumer.KafkaConsumer KafkaConsumer} instance that
is used for reading input and writing output.
  * <p>
+ *
  * A simple example might look like this:
  * <pre>
  *    Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
- *    props.put("bootstrap.servers", "localhost:4242");
- *    props.put("key.deserializer", StringDeserializer.class);
- *    props.put("value.deserializer", StringDeserializer.class);
- *    props.put("key.serializer", StringSerializer.class);
- *    props.put("value.serializer", IntegerSerializer.class);
- *    props.put("timestamp.extractor", MyTimestampExtractor.class);
+ *    props.put(StreamsConfig.JOB_ID_CONFIG, "my-job");
+ *    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+ *    props.put(StreamsConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ *    props.put(StreamsConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ *    props.put(StreamsConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ *    props.put(StreamsConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  *    StreamsConfig config = new StreamsConfig(props);
  *
  *    KStreamBuilder builder = new KStreamBuilder();
- *    builder.from("topic1").mapValue(value -&gt; value.length()).to("topic2");
+ *    builder.from("my-input-topic").mapValue(value -&gt; value.length().toString()).to("my-output-topic");
  *
  *    KafkaStreams streams = new KafkaStreams(builder, config);
  *    streams.start();
  * </pre>
  *
  */
-// TODO: about example may need to be updated after KAFKA-3153
 
 @InterfaceStability.Unstable
 public class KafkaStreams {

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 65ec969..c4b8ffe 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -35,6 +35,10 @@ import java.util.Map;
 import static org.apache.kafka.common.config.ConfigDef.Range.atLeast;
 import static org.apache.kafka.common.config.ConfigDef.ValidString.in;
 
+/**
+ * Configuration for Kafka Streams. Documentation for these configurations can be found in
the <a
+ * href="http://kafka.apache.org/documentation.html#streamsconfigs">Kafka documentation</a>
+ */
 public class StreamsConfig extends AbstractConfig {
 
     private static final ConfigDef CONFIG;

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java b/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
index d392eef..70c3320 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
@@ -20,7 +20,7 @@ package org.apache.kafka.streams;
 import org.apache.kafka.common.metrics.Sensor;
 
 /**
- * The stream metrics interface for adding metric sensors and collecting metric values.
+ * The Kafka Streams metrics interface for adding metric sensors and collecting metric values.
  */
 public interface StreamsMetrics {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
----------------------------------------------------------------------
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 231eb22..6426af9 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
@@ -22,13 +22,11 @@ import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.processor.ProcessorSupplier;
 
-
-// TODO: Javadoc needs to be updated
 /**
- * KStream is an abstraction of a stream of key-value pairs.
+ * KStream is an abstraction of an event stream in key-value pairs.
  *
- * @param <K> the type of keys
- * @param <V> the type of values
+ * @param <K> Type of keys
+ * @param <V> Type of values
  */
 public interface KStream<K, V> {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
----------------------------------------------------------------------
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 a2c6397..485bb20 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
@@ -21,13 +21,11 @@ import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.streams.KeyValue;
 
-// TODO: Javadoc needs to be updated.
 /**
- * KTable is an abstraction of a change log stream.
+ * KTable is an abstraction of a change log stream from a primary-keyed table.
  *
- *
- * @param <K> the type of keys
- * @param <V> the type of values
+ * @param <K> Type of primary keys
+ * @param <V> Type of value changes
  */
 public interface KTable<K, V> {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java
----------------------------------------------------------------------
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 b67f619..47198e4 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
@@ -19,6 +19,13 @@ package org.apache.kafka.streams.kstream;
 
 import org.apache.kafka.streams.processor.ProcessorContext;
 
+/**
+ * A stateful Transformer interface for transform a key-value pair into a new value.
+ *
+ * @param <K> Key type.
+ * @param <V> Value type.
+ * @param <R> Return type.
+ */
 public interface Transformer<K, V, R> {
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java
----------------------------------------------------------------------
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 93d930d..fc7ba60 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
@@ -17,6 +17,9 @@
 
 package org.apache.kafka.streams.kstream;
 
+/**
+ * A transformer supplier which can create one or more {@link Transformer} instances.
+ */
 public interface TransformerSupplier<K, V, R> {
 
     Transformer<K, V, R> get();

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
----------------------------------------------------------------------
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 06882b3..7cadfb4 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
@@ -22,6 +22,9 @@ import org.apache.kafka.streams.kstream.internals.UnlimitedWindow;
 import java.util.HashMap;
 import java.util.Map;
 
+/**
+ * The unlimited window specifications.
+ */
 public class UnlimitedWindows extends Windows<UnlimitedWindow> {
 
     private static final long DEFAULT_START_TIMESTAMP = 0L;

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java
----------------------------------------------------------------------
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 5b9e2ff..b4d2b38 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
@@ -19,6 +19,12 @@ package org.apache.kafka.streams.kstream;
 
 import org.apache.kafka.streams.processor.ProcessorContext;
 
+/**
+ * A stateful Value Transformer interface for transform a value into a new value.
+ *
+ * @param <V> Value type.
+ * @param <R> Return type.
+ */
 public interface ValueTransformer<V, R> {
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
----------------------------------------------------------------------
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 04fa9eb..6bc86bc 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
@@ -17,6 +17,9 @@
 
 package org.apache.kafka.streams.kstream;
 
+/**
+ * A value transformer supplier which can create one or more {@link ValueTransformer} instances.
+ */
 public interface ValueTransformerSupplier<V, R> {
 
     ValueTransformer<V, R> get();

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java
----------------------------------------------------------------------
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 845f9e9..22d52aa 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
@@ -17,6 +17,15 @@
 
 package org.apache.kafka.streams.kstream;
 
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serializer;
+
+/**
+ * The windowed key interface used in {@link KTable}, used for representing a windowed table
result from windowed stream aggregations,
+ * i.e. {@link KStream#aggregateByKey(Initializer, Aggregator, Windows, Serializer, Serializer,
Deserializer, Deserializer)}
+ *
+ * @param <T> Type of the key
+ */
 public class Windowed<T> {
 
     private T value;

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/processor/DefaultPartitionGrouper.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/DefaultPartitionGrouper.java
b/streams/src/main/java/org/apache/kafka/streams/processor/DefaultPartitionGrouper.java
index dad5c6f..06681ac 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/DefaultPartitionGrouper.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/DefaultPartitionGrouper.java
@@ -30,7 +30,7 @@ import java.util.Map;
 import java.util.Set;
 
 /**
- * DefaultPartitionGrouper groups partitions by the partition id. This behavior is assumed
by the join processing in KStream.
+ * Default implementation of the {@link PartitionGrouper} interface that groups partitions
by the partition id.
  *
  * Join operations requires that topics of the joining entities are copartitoned, i.e., being
partitioned by the same key and having the same
  * number of partitions. Copartitioning is ensured by having the same number of partitions
on

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/processor/PartitionGrouper.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/PartitionGrouper.java
b/streams/src/main/java/org/apache/kafka/streams/processor/PartitionGrouper.java
index f8311e7..ae9844d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/PartitionGrouper.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/PartitionGrouper.java
@@ -23,6 +23,13 @@ import org.apache.kafka.common.TopicPartition;
 import java.util.Map;
 import java.util.Set;
 
+/**
+ * A partition grouper that generates partition groups given the list of topic-partitions.
+ *
+ * This grouper also acts as the stream task creation function along with partition distribution
+ * such that each generated partition group is assigned with a distinct {@link TaskId};
+ * the created task ids will then be assigned to Kafka Streams instances that host the stream
job.
+ */
 public interface PartitionGrouper {
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorSupplier.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorSupplier.java
b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorSupplier.java
index 719d3ac..6561899 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorSupplier.java
@@ -17,6 +17,16 @@
 
 package org.apache.kafka.streams.processor;
 
+/**
+ * A processor supplier that can create one or more {@link Processor} instances.
+ *
+ * It is used in {@link TopologyBuilder} for adding new processor operators, whose generated
+ * topology can then be replicated (and thus creating one or more {@link Processor} instances)
+ * and distributed to multiple stream threads.
+ *
+ * @param <K> the type of keys
+ * @param <V> the type of values
+ */
 public interface ProcessorSupplier<K, V> {
 
     Processor<K, V> get();

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
b/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
index ce0ba70..224d580 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
@@ -20,7 +20,8 @@ package org.apache.kafka.streams.processor;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 /**
- * An interface that allows the Kafka Streams framework to extract a timestamp from an instance
of {@link ConsumerRecord}
+ * An interface that allows the Kafka Streams framework to extract a timestamp from an instance
of {@link ConsumerRecord}.
+ * The extracted timestamp is defined as milliseconds.
  */
 public interface TimestampExtractor {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/state/KeyValueIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueIterator.java b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueIterator.java
index 55ec8cf..cdb3de5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueIterator.java
@@ -33,5 +33,5 @@ import java.util.Iterator;
 public interface KeyValueIterator<K, V> extends Iterator<KeyValue<K, V>>,
Closeable {
 
     @Override
-    public void close();
+    void close();
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
index f296230..3e7f6fb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
@@ -39,7 +39,7 @@ public interface KeyValueStore<K, V> extends StateStore {
      * @return The value or null if no value is found.
      * @throws NullPointerException If null is used for key.
      */
-    abstract public V get(K key);
+    V get(K key);
 
     /**
      * Update the value associated with this key
@@ -48,7 +48,7 @@ public interface KeyValueStore<K, V> extends StateStore {
      * @param value The value
      * @throws NullPointerException If null is used for key or value.
      */
-    abstract public void put(K key, V value);
+    void put(K key, V value);
 
     /**
      * Update the value associated with this key, unless a value
@@ -59,7 +59,7 @@ public interface KeyValueStore<K, V> extends StateStore {
      * @return The old value or null if there is no such key.
      * @throws NullPointerException If null is used for key or value.
      */
-    abstract public V putIfAbsent(K key, V value);
+    V putIfAbsent(K key, V value);
 
     /**
      * Update all the given key/value pairs
@@ -67,7 +67,7 @@ public interface KeyValueStore<K, V> extends StateStore {
      * @param entries A list of entries to put into the store.
      * @throws NullPointerException If null is used for any key or value.
      */
-    abstract public void putAll(List<KeyValue<K, V>> entries);
+    void putAll(List<KeyValue<K, V>> entries);
 
     /**
      * Delete the value from the store (if there is one)
@@ -76,7 +76,7 @@ public interface KeyValueStore<K, V> extends StateStore {
      * @return The old value or null if there is no such key.
      * @throws NullPointerException If null is used for key.
      */
-    abstract public V delete(K key);
+    V delete(K key);
 
     /**
      * Get an iterator over a given range of keys. This iterator MUST be closed after use.
@@ -86,13 +86,13 @@ public interface KeyValueStore<K, V> extends StateStore {
      * @return The iterator for this range.
      * @throws NullPointerException If null is used for from or to.
      */
-    abstract public KeyValueIterator<K, V> range(K from, K to);
+    KeyValueIterator<K, V> range(K from, K to);
 
     /**
      * Return an iterator over all keys in the database. This iterator MUST be closed after
use.
      *
      * @return An iterator of all key/value pairs in the store.
      */
-    abstract public KeyValueIterator<K, V> all();
+    KeyValueIterator<K, V> all();
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
index e1e78af..e925312 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Serdes.java
@@ -27,6 +27,12 @@ import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 
+/**
+ * Factory for creating serializers / deserializers for state stores in Kafka Streams.
+ *
+ * @param <K> key type of serdes
+ * @param <V> value type of serdes
+ */
 public final class Serdes<K, V> {
 
     public static <K, V> Serdes<K, V> withBuiltinTypes(String topic, Class<K>
keyClass, Class<V> valueClass) {

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
index e9d82bc..e803832 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
@@ -33,7 +33,7 @@ import org.apache.kafka.streams.state.internals.RocksDBKeyValueStoreSupplier;
 import org.apache.kafka.streams.state.internals.RocksDBWindowStoreSupplier;
 
 /**
- * Factory for creating key-value stores.
+ * Factory for creating state stores in Kafka Streams.
  */
 public class Stores {
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
b/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
index 08cd049..7c474dd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreIterator.java
@@ -23,6 +23,11 @@ import org.apache.kafka.streams.KeyValue;
 
 import java.util.Iterator;
 
+/**
+ * Iterator interface of {@link KeyValue} with key typed {@link Long} used for {@link WindowStore#fetch(Object,
long, long)}.
+ *
+ * @param <E> Type of values
+ */
 public interface WindowStoreIterator<E> extends Iterator<KeyValue<Long, E>>
{
     void close();
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/f676cfeb/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreUtils.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreUtils.java b/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreUtils.java
index 3a3d585..c6bbb23 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreUtils.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/WindowStoreUtils.java
@@ -21,7 +21,7 @@ package org.apache.kafka.streams.state;
 
 import java.nio.ByteBuffer;
 
-public class WindowStoreUtils<K, V> {
+public class WindowStoreUtils {
 
     public static final int TIMESTAMP_SIZE = 8;
     public static final int SEQNUM_SIZE = 4;


Mime
View raw message