boyuanzz commented on a change in pull request #11749:
URL: https://github.com/apache/beam/pull/11749#discussion_r442566478
##########
File path:
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
##########
@@ -795,6 +828,12 @@ public void setValueDeserializer(String valueDeserializer)
{
return withWatermarkFn2(unwrapKafkaAndThen(watermarkFn));
}
+ /** A function to the compute output timestamp from a {@link KafkaRecord}.
*/
+ public Read<K, V> withExtractOutputTimestampFn(
Review comment:
> How is this different from withTimestampFn2?
`withTimestampFn2` has been deprecated in `KafkaIO.Read`. The major concern
of reusing `withTimestampFn2` is, it will means differently under SDF and
UnboundedSource, which causes confusion.
> Setting the top level properties allow us to say that this property is
supported when used as an SDF.
This is what I want to do by having `withExtractOutputTimestampFn `
> Also, what prevents us from supporting TimestampPolicy? We should be able
to call it and give it the three pieces of information it requests (message
backlog / backlog check time / current kafka record).
The difficulty is the message backlog / backlog check time is not memorized
per (element, restriction). With SDF framework, the backlog is retrieved by
called `RestricitonTracker.getProgress()`, we cannot call it per element in
order to extract timestamp.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
