Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1699#discussion_r53986228
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/TimestampedCollector.java
---
@@ -51,17 +48,19 @@ public TimestampedCollector(Output<StreamRecord<T>>
output) {
@Override
public void collect(T record) {
- output.collect(reuse.replace(record, timestamp));
+ output.collect(reuse.replace(record));
+ }
+
+ public void setTimestamp(StreamRecord<?> timestampBase) {
+ if (timestampBase.hasTimestamp()) {
+ reuse.setTimestamp(timestampBase.getTimestamp());
+ } else {
+ reuse.eraseTimestamp();
+ }
}
- /**
- * Sets the timestamp (long milliseconds) that is attached to elements
that get emitted using
- * {@link #collect(Object)}
- *
- * @param timestamp The timestamp in milliseconds
- */
- public void setTimestamp(long timestamp) {
- this.timestamp = timestamp;
+ public void setAbsoluteTimestamp(long timestamp) {
--- End diff --
Why not call both methods setTimestamp. This name seems to imply that the
timestamp is not absolute in the other method.
Maybe that's just me :sweat_smile:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
