[BEAM-1514] change default timestamp in KafkaIO #2267

mingmxu · 2017-03-17T20:20:18Z

Be sure to do all of the following to help us incorporate your contribution
quickly and easily:

Make sure the PR title is formatted like:
[BEAM-<Jira issue #>] Description of pull request
Make sure tests pass via mvn clean verify. (Even better, enable
Travis-CI on your fork and ensure the whole test matrix passes).
Replace <Jira issue #> in the title with the actual Jira issue
number, if there is one.
If this contribution is large, please file an Apache
Individual Contributor License Agreement.

sync-up code

1). fix issue of NO_TIMESTAMP type in 10; 2). rename field to 'timestamp';

coveralls · 2017-03-17T21:01:25Z

Coverage increased (+0.003%) to 70.12% when pulling 03f8497 on XuMingmin:BEAM-1705 into b672cde on apache:master.

asfbot · 2017-03-17T21:02:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8529/
--none--

davorbonaci · 2017-03-20T17:01:18Z

R: @rangadi

rangadi · 2017-03-21T20:05:43Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java

@@ -57,4 +74,31 @@ public void evaluateAssign(Consumer consumer, Collection<TopicPartition> topicPa
 mapContext.setVariable("tp", topicPartitions);
 assignExpression.getValue(mapContext);
 }
+
+ private boolean hasTimestamp(){


Why do you need this method? Could you check the return type when you check 'timestamp' method?

This function is used to check whether ConsumerRecord.class has a method timestamp which returns Long.TYPE. Kafka0.9 doesn't have it, Kafka0.10 has. The check should be complete as

hasEventTimestamp = timestampMethod != null && timestampMethod.getReturnType().equals(Long.TYPE);

Right?

rangadi · 2017-03-21T20:07:03Z

@xumingmin, did you check the review comments from the earlier PR for this? What do you think of them?

rangadi · 2017-03-21T20:10:14Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java

+ return hasEventTimestamp;
+ }
+
+ public long getEventTimestamp(ConsumerRecord<byte[], byte[]> rawRecord) {


Rename ito getTimestamp or getRecordTimestamp()? Rest of KafkaIO refers each messages as 'record' rather than an 'event.

surely, will do

rangadi · 2017-03-21T20:12:29Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -986,11 +996,13 @@ public boolean advance() throws IOException {
 rawRecord.topic(),
 rawRecord.partition(),
 rawRecord.offset(),
+ consumerSpEL.getEventTimestamp(rawRecord),
 decode(rawRecord.key(), source.spec.getKeyCoder()),
 decode(rawRecord.value(), source.spec.getValueCoder()));

 curTimestamp = (source.spec.getTimestampFn() == null)


curTimestamp is same as record timestamp. Here, we apply user function only to curTimestamp, why? To avoid this confusion, I suggest removing curTimestamp.

curTimestamp is different from KafkaRecord.timestamp when setTimestampFn() is there.
Example like it's a topic of checkout entries, I can setTimestampFn() to extract checkout_timestamp as curTimestamp, neither Kafka record timestamp nor current_timestamp.

ah, right. curTimestamp defaults to kafkaRecord timestamp and could be overridden by the user.

rangadi · 2017-03-21T20:18:58Z

Can you add comment in getWatermark()?
What are the guarantees about Kafka timestamps? If they are server side ingest timestamps, are they expected to be monotonically increasing? What happens if these are producer side publish timestamp? Could they be out of order? getWatermark() should return monotonically increasing timestamp.

mingmxu · 2017-03-21T22:55:02Z

@rangadi should all comments in previous PR included here, kindly let me know if anything missing.

getWatermark() is the one I didn't notice. With message timestamp in Kafka 0.10, I don't think it can guarantee to be monotonically increasing:

with LogAppendTime timestamp may be out-of-order when reading from multiple partitions;
with CreateTime, it's similar, very likely out-of-order;

My thought is, I should change the logic of curTimestamp to

curTimestamp = max(curTimestamp,  
      (source.spec.getTimestampFn() == null) ? new Instant(record.getTimestamp())
                  : source.spec.getTimestampFn().apply(record)
   );

Is that the right way to go?

coveralls · 2017-03-21T23:43:30Z

Coverage decreased (-0.2%) to 69.919% when pulling c8eb75e on XuMingmin:BEAM-1705 into b672cde on apache:master.

asfbot · 2017-03-21T23:43:35Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8639/
--none--

rangadi · 2017-03-21T23:57:37Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java

@@ -50,15 +50,15 @@
 parser.parseExpression("#consumer.assign(#tp)");

 private Method timestampMethod;
- private boolean hasEventTimestamp = false;
+ private boolean hasRecordTimestamp = false;

 public ConsumerSpEL() {
 try {
 timestampMethod = ConsumerRecord.class.getMethod("timestamp", (Class<?>[]) null);
 } catch (NoSuchMethodException | SecurityException e) {


how about hasRecordTimeStamp = timestampMethod.getReturnType().equals(Long.TYPE) here?
That way you can remove lines 78-87.

rangadi · 2017-03-22T00:17:36Z

getWatermark() handling becomes quite important and more tricky now. I think we need to handle it appropriately if kafka timestamp becomes the default.

LogAppendTime :
- We could handle it by keeping minimum timestamp we have seen across the partitions (also update it only after we know have tried to read from the partition with lowest timestamp). i.e. upto 1 sec latency.
with createTime:
- ¯\_(ツ)_/¯
- May be we need to do something like what PubsubIO. Delay a minute or so..

Current watermark is ok only because of processing time. If we make new timestamps the default, it could be surprising to unsuspecting users. What do you think?

LogAppendTime might be the most common use case. It might be good enough for now to handle that. This would handle backlogged pipelines very nicely. The windows will fire correctly.

coveralls · 2017-03-22T01:16:23Z

Coverage decreased (-0.2%) to 69.919% when pulling 55bb5ec on XuMingmin:BEAM-1705 into b672cde on apache:master.

asfbot · 2017-03-22T01:17:53Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8641/
--none--

mingmxu · 2017-03-22T06:04:30Z

Add up the two options you suggested, I think we can provide a default watermarkFn(), it returns a watermark with the minimum value of:

latest_timestamp minus 1 minute;
timestamp less than 99% records in past minutes for example. The percentage/statistics window is adjustable. --It may be costly.

The watermark should cover as many records as possible, to avoid marking as late-record.

rangadi · 2017-03-23T04:52:11Z

(1): What happens when the pipeline is behind by 5 minutes. We will end up marking 4 minutes of that as late data.

I think we can do a pretty good job with LogAppendTime. E.g. we can keep track of latest timestamp from each of the partitions and return the least among them. Only tricky part is handling idle partitions where we might not see any records for some time. We know if we have completely consumed a partition. We can advance the watermark to something like 'current time - 5 seconds' in that case.

(2): Not sure that. Why not less than all of them. In case of 'createTime' we can require user to set the watermarkFn for now.

Implementation wise, we can have multiple watermark policies 'LogAppendTimePolicy', 'ProcessingTimePolicy', 'CreateTimePolicy' (that last one just requires user to set watermakFn).

For this PR, we can default to old behavior of 'processing time' for 'curTimestamp' and follow up with proper support for LogAppendTime. Rest of the PR could remain the same, i.e. 'timestmap' in KafkaRecord would be the timestamp provided by Kafka.

In your Kafka 0.10.x deployment, did you enable per record timestamps?

rangadi · 2017-03-23T04:57:02Z

cc: @amitsela

mingmxu · 2017-03-23T05:59:09Z

Use createTime in my case, however I provide customized TimestampFn() and watermarkFn() to handle all know 'issues', like latency skew between partitions, restart from an offset days ago. Meanwhile need to handle late-data properly.

I would agree to step back to keep the old behavior like:

#event time
curTimestamp = (source.spec.getTimestampFn() == null)
              ? new Instant(record.getTimestamp())
                  : source.spec.getTimestampFn().apply(record);

#watermark
return source.spec.getWatermarkFn() != null
          ? source.spec.getWatermarkFn().apply(curRecord) : Instant.now();

Developers need to tune the watermarkFn, any thoughts?

amitsela

I've added a bunch of minor comments.

amitsela · 2017-03-23T09:11:17Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java

@@ -57,4 +74,20 @@ public void evaluateAssign(Consumer consumer, Collection<TopicPartition> topicPa
 mapContext.setVariable("tp", topicPartitions);
 assignExpression.getValue(mapContext);
 }
+
+ public long getRecordTimestamp(ConsumerRecord<byte[], byte[]> rawRecord) {
+ long timestamp = -1L;


I guess this doesn't have to be initialized.

amitsela · 2017-03-23T09:12:23Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java

+ try {
+ //for Kafka 0.9, set to System.currentTimeMillis();
+ //for kafka 0.10, when NO_TIMESTAMP also set to System.currentTimeMillis();
+ if (!hasRecordTimestamp || (timestamp = (long) timestampMethod.invoke(rawRecord)) == -1L) {


Unless I don't know of something specific that would cause a -1 return value, and since there is a return value if the code gets here, I'd go with < 0 instead of == -1L

-1L is the timestamp returned when NO_TIMESTAMP in Kafka0.10, could change to <=0 to cover more unexpected cases.

amitsela · 2017-03-23T09:13:12Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -31,6 +31,7 @@
 import com.google.common.collect.Iterators;
 import com.google.common.collect.Lists;
 import com.google.common.io.Closeables;
+


Nit: you have a bunch of newlines added in-between imports that probably the IDE's "auto-indent" added, they are unnecessary.

kind of code standard, same for the added @Override in KafkaIO, intend to keep it for better reading.

I usually take care not to impose my code standards in older code it am modifying it anyway. But ok :-).

good point to follow, let me take a look at my format profile.

amitsela · 2017-03-23T09:54:44Z

I think @xumingmin has a point here - by adding the "Kafka timestamp" to the KafkaRecord and providing it to the WM function we give users more information, while still defaulting to a naive "now" WM.

rangadi · 2017-03-23T17:37:12Z

I think @xumingmin has a point here - by adding the "Kafka timestamp" to the KafkaRecord and providing it to the WM function we give users more information

Yes, that is what I am suggesting here short of proper handling of LogAppendTime. See the caveat below.

while still defaulting to a naive "now" WM.

You mean 'curTimestamp'? This policy requires that we should also default 'curTimestamp' to processing timestamp, not kafka record timestamp.

rangadi · 2017-03-23T18:16:57Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

 return source.spec.getWatermarkFn() != null
- ? source.spec.getWatermarkFn().apply(curRecord) : curTimestamp;


I am not sure if this is a good default behavior. Could you explain?

I set the default watermark to current_timestamp if no user-specified WarermarkFn. This should be the same as before since curTimestamp in kafka0.9 equals to current_timestamp if no timestampFn provided.

I guess in kafka0.9, if timestampFn is provided, this watermark cannot guarantee to monotone increasing.

I set the default watermark to current_timestamp if no user-specified WarermarkFn.

As I mentioned earlier, this works only when curTimestamp is also now. Do you agree? Otherwise, most if not all the records could end up as late, which would be a very surprising result for users. I don't think just asking the user to 'set the right watermark fn' as a requirement is a good default. Places a lot of burden even on simple use cases.

I guess in kafka0.9, if timestampFn is provided, this watermark cannot guarantee to monotone increasing.

True, but that is not the default. That needed improvement anyway.

Right, I assume it's part of kafka0.10 client support, so users follow this way need to take care of the new scenario.
kafka_record_timestamp is kind of customized event_timestamp for me, so I would ask developers to handle late-data properly, mostly with Trigger.withAllowedLateness().

Often the consumers of the streams may not be aware of these timestamps or might just want KafkaIO to do the most sensible thing. I am not sure having a default implementation that marks most records as late and expecting user to handle the correctly is a great policy.

To summarize, my proposal (reasons are discussed earlier):

Default to processing timestamps. i.e. KafkaRecord will have the timestamp provided by the Kafka (if any) and 'curTimestamp' and 'watermark' don't change from previous behavior. The simplest option.

OR Handle LogAppendTimestamp properly to return appropriate watermark (essentially min(log_append_time for each partition read). We can do this in this PR or in another one.

I think this is an important aspect of this PR. Would love to hear yours and others thoughts too. cc @kennknowles @dhalperi @davorbonaci

Agree with 1) ,

by default both event_timestamp and watermark as processing timestamps.

Developers can combine event_timestamp and watermark if required.

Sounds good to me. Thanks.

thanks @rangadi , updated as talked.

coveralls · 2017-03-23T19:04:51Z

Coverage increased (+0.05%) to 70.162% when pulling 20a0e06 on XuMingmin:BEAM-1705 into b672cde on apache:master.

asfbot · 2017-03-23T19:48:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8727/
--none--

rangadi

LGTM.
Couple of suggestioins for comments above. Thanks @xumingmin.

rangadi · 2017-03-23T22:03:10Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java

+ timestamp = System.currentTimeMillis();
+ }
+ } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
+ // should not go to here, as it's confirmed that method timestamp() is


Rephrase? "Not expected. timestampMethod() is already checked."

rangadi · 2017-03-23T22:11:12Z

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -203,6 +203,12 @@
 * {@link ProducerConfig} for sink. E.g. if you would like to enable offset
 * <em>auto commit</em> (for external monitoring or other purposes), you can set
 * <tt>"group.id"</tt>, <tt>"enable.auto.commit"</tt>, etc.
+ *
+ * <h3>Event Timestamp and Watermark</h3>
+ * The default event_timestamp and watermark increases along as processing timestamp,


minor: s/Event/Record/ ?
"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)}. Note that {@link KafkaRecord#getTimestamp()} reflects timestamp provided by Kafka if any, otherwise it is set to processing time."

coveralls · 2017-03-23T22:43:01Z

Coverage increased (+0.06%) to 70.172% when pulling 6148ac6 on XuMingmin:BEAM-1705 into b672cde on apache:master.

coveralls · 2017-03-23T22:48:05Z

Coverage increased (+0.06%) to 70.178% when pulling 6148ac6 on XuMingmin:BEAM-1705 into b672cde on apache:master.

asfbot · 2017-03-23T22:54:40Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8739/
--none--

asfbot · 2017-03-23T22:55:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8738/
--none--

rangadi · 2017-03-23T22:56:10Z

👍

coveralls · 2017-03-23T23:18:47Z

Coverage increased (+0.06%) to 70.172% when pulling 6148ac6 on XuMingmin:BEAM-1705 into b672cde on apache:master.

asfbot · 2017-03-23T23:19:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/8743/
--none--

davorbonaci

Thanks @xumingmin and @rangadi. Merging.

mingmxu · 2017-03-24T17:34:43Z

thank you @davorbonaci
special thanks to @rangadi for the discussion, really helpful!

xumingmin and others added 2 commits March 15, 2017 14:45

Merge pull request #3 from apache/master

09df9b5

sync-up code

append to apache#2135, add

03f8497

1). fix issue of NO_TIMESTAMP type in 10; 2). rename field to 'timestamp';

rangadi reviewed Mar 21, 2017

View reviewed changes

change EventTimestamp to RecordTimestamp to avoid confusion.

c8eb75e

rangadi reviewed Mar 21, 2017

View reviewed changes

inline function hasRecordTimestamp()

55bb5ec

amitsela reviewed Mar 23, 2017

View reviewed changes

revert behavior of watermark, and add JavaDoc to highlight it.

20a0e06

rangadi reviewed Mar 23, 2017

View reviewed changes

by default both event_timestamp and watermark as processing timestamps.

0452a57

remove java comments that's not required anymore

dfd3157

rangadi reviewed Mar 23, 2017

View reviewed changes

rephrase JavaDoc

6148ac6

davorbonaci approved these changes Mar 24, 2017

View reviewed changes

asfgit closed this in 5c2da7d Mar 24, 2017

		return source.spec.getWatermarkFn() != null
		? source.spec.getWatermarkFn().apply(curRecord) : curTimestamp;

[BEAM-1514] change default timestamp in KafkaIO #2267

[BEAM-1514] change default timestamp in KafkaIO #2267

Conversation

mingmxu commented Mar 17, 2017

coveralls commented Mar 17, 2017

asfbot commented Mar 17, 2017

davorbonaci commented Mar 20, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rangadi commented Mar 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rangadi commented Mar 21, 2017

mingmxu commented Mar 21, 2017 • edited Loading

coveralls commented Mar 21, 2017

asfbot commented Mar 21, 2017

Choose a reason for hiding this comment

rangadi commented Mar 22, 2017

coveralls commented Mar 22, 2017

asfbot commented Mar 22, 2017

mingmxu commented Mar 22, 2017

rangadi commented Mar 23, 2017 • edited Loading

rangadi commented Mar 23, 2017

mingmxu commented Mar 23, 2017

amitsela left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amitsela commented Mar 23, 2017

rangadi commented Mar 23, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rangadi Mar 23, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Mar 23, 2017

asfbot commented Mar 23, 2017

rangadi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Mar 23, 2017

coveralls commented Mar 23, 2017

asfbot commented Mar 23, 2017

asfbot commented Mar 23, 2017

rangadi commented Mar 23, 2017

coveralls commented Mar 23, 2017

asfbot commented Mar 23, 2017

davorbonaci left a comment

Choose a reason for hiding this comment

mingmxu commented Mar 24, 2017

mingmxu commented Mar 21, 2017 •

edited

Loading

rangadi commented Mar 23, 2017 •

edited

Loading

amitsela left a comment •

edited

Loading

rangadi commented Mar 23, 2017 •

edited

Loading

rangadi Mar 23, 2017 •

edited

Loading