This is an automated email from the ASF dual-hosted git repository.
mingmxu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new ae79032 [BEAM-591]: Update KafkaIO JavaDoc to reflect new timestamp
API. (#4749)
ae79032 is described below
commit ae790320312e0ee35e8bd1ba3b112edaa64b7f93
Author: Raghu Angadi <[email protected]>
AuthorDate: Mon Feb 26 16:42:30 2018 -0800
[BEAM-591]: Update KafkaIO JavaDoc to reflect new timestamp API. (#4749)
* Update KafkaIO JavaDoc to reflect new timestamp API.
* fix a typo.
---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 20 ++++++++------------
1 file changed, 8 insertions(+), 12 deletions(-)
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index f031003..55113b8 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -106,21 +106,18 @@ import org.slf4j.LoggerFactory;
* .withKeyDeserializer(LongDeserializer.class)
* .withValueDeserializer(StringDeserializer.class)
*
- * // above four are required configuration. returns
PCollection<KafkaRecord<Long, String>>
+ * // Above four are required configuration. returns
PCollection<KafkaRecord<Long, String>>
*
- * // rest of the settings are optional :
+ * // Rest of the settings are optional :
*
* // you can further customize KafkaConsumer used to read the records
by adding more
* // settings for ConsumerConfig. e.g :
- * .updateConsumerProperties(ImmutableMap.of("receive.buffer.bytes",
1024 * 1024))
+ * .updateConsumerProperties(ImmutableMap.of("group.id",
"my_beam_app_1"))
*
* // set event times and watermark based on LogAppendTime. To provide a
custom
* // policy see withTimestampPolicyFactory(). withProcessingTime() is
the default.
* .withLogAppendTime()
*
- * // custom function for watermark (default is record timestamp)
- * .withWatermarkFn(new MyWatermarkFunction())
- *
* // restrict reader to committed messages on Kafka (see method
documentation).
* .withReadCommitted()
*
@@ -209,12 +206,11 @@ import org.slf4j.LoggerFactory;
* <tt>"group.id"</tt>, <tt>"enable.auto.commit"</tt>, etc.
*
* <h3>Event Timestamp and Watermark</h3>
- * By default record timestamp and watermark are based on processing time in
KafkaIO reader.
- * This can be overridden by providing {@code WatermarkFn} with
- * {@link Read#withWatermarkFn(SerializableFunction)}, and {@code TimestampFn}
with
- * {@link Read#withTimestampFn(SerializableFunction)}.<br>
- * Note that {@link KafkaRecord#getTimestamp()} reflects timestamp provided by
Kafka if any,
- * otherwise it is set to processing time.
+ * By default, record timestamp (event time) is set to processing time in
KafkaIO reader and
+ * source watermark is current wall time. If a topic has Kafka server-side
ingestion timestamp
+ * enabled ('LogAppendTime'), it can enabled with {@link
Read#withLogAppendTime()}.
+ * A custom timestamp policy can be provided by implementing {@link
TimestampPolicyFactory}. See
+ * {@link Read#withTimestampPolicyFactory(TimestampPolicyFactory)} for more
information.
*/
@Experimental(Experimental.Kind.SOURCE_SINK)
public class KafkaIO {
--
To stop receiving notification emails like this one, please contact
[email protected].
