Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/1699#discussion_r53994466
--- 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 --
I renamed that to make sure I do not overlook any places where the better
method would have been `setTimestamp(StreamRecord)`. Ideally, we'd not have the
`setAbsoluteTimestamp(long)` method at all. But that would need changes in the
window operators.
---
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.
---
