vbhanuchander-lang commented on PR #28491:
URL: https://github.com/apache/flink/pull/28491#issuecomment-5526515837
@wilmerdooley the approach looks right to me — deferring watermark
generation to a downstream
operator is the only way a Python `TimestampAssigner` can run, and it
matches what
`DataStream.assign_timestamps_and_watermarks` already does for the same
reason.
**One problem with landing it on its own, though: FLINK-39725's reported
case would still fail.**
The reproduction in the JIRA uses a CSV source, and the path this PR now
takes runs into
FLINK-39724 on the way:
1. this PR routes through `DataStream.assign_timestamps_and_watermarks`
2. its first step is `self.process(..., Types.TUPLE([self.get_type(),
Types.LONG()]))`
3. `get_type()` is `typeinfo._from_java_type(self._j_data_stream.getType())`
4. `PythonCsvUtils.createCsvReaderFormat` builds the format with
`InternalTypeInfo.of(dataType.getLogicalType())`
(`PythonCsvUtils.java:48`), so that is the
stream's Java type
5. `_from_java_type` on `master` has branches up to `ExternalTypeInfo` and
then
`raise TypeError("The java type info: %s is not supported in PyFlink
currently.")` —
no `InternalTypeInfo` branch
So after this change the reporter's script fails with a `TypeError` from
`get_type()` rather than
silently ignoring the assigner. Better than silent, but still broken.
**The test does not catch this** because `NumberSequenceSource` with
`type_info=Types.LONG()` never
produces an `InternalTypeInfo`, so it exercises the new branch without
exercising the reported
scenario.
Worth noting the two JIRAs came from the same reporter running the same
script — FLINK-39724 and
FLINK-39725 are really one report split in two. Your #28490 is what makes
step 5 work, so I would
either land that one first and note the dependency here, or fold them
together. Either way a CSV
source case in the test would pin the actual bug.
I have not run this — there is no Flink build on my machine, so everything
above is from reading
`master`, not from executing the tests. Worth confirming before you act on
it.
--
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]
