sangkyoonnam commented on code in PR #29312:
URL: https://github.com/apache/flink/pull/29312#discussion_r4303920883
##########
flink-python/pyflink/datastream/data_stream.py:
##########
@@ -704,9 +709,15 @@ def process_element(self, value, ctx:
'ProcessFunction.Context'):
# step 3: remove the timestamp field which is added in step 1
JRemoveTimestampMapFunction =
gateway.jvm.org.apache.flink.streaming.api.functions \
.python.eventtime.RemoveTimestampMapFunction
- result = DataStream(j_watermarked_data_stream.map(
- JRemoveTimestampMapFunction(), self._j_data_stream.getType()))
- result.name("Remove-Timestamp")
+ result = _TimestampsAndWatermarksDataStream(
+ j_watermarked_data_stream.map(
+ JRemoveTimestampMapFunction(),
self._j_data_stream.getType()),
+ [(UID_SUFFIX_EXTRACT_TIMESTAMP,
+ timestamped_data_stream._j_data_stream.getTransformation()),
+ (UID_SUFFIX_TIMESTAMPS_AND_WATERMARKS,
+ j_watermarked_data_stream.getTransformation())])
+ # Not result.name(), which would be taken for a name set by the
user.
+ result._j_data_stream.name("Remove-Timestamp")
Review Comment:
Nit: `_TimestampsAndWatermarksDataStream` inherits `DataStream.name()`,
which forwards the name to Java and returns `self`; it doesn't track whether a
name came from the user. The return value isn't used here, so the original call
works and the comment can go:
```suggestion
result.name("Remove-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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
