[PR] FLINK-39725: fix(python): apply timestamp assigner in from_source for pyflink [flink]

["wilmerdooley (via GitHub)" <gi...@apache.org>]

wilmerdooley opened a new pull request, #28491:
URL: https://github.com/apache/flink/pull/28491

   ## What is the purpose of this PR
   
   When using `StreamExecutionEnvironment.from_source` in PyFlink with a `WatermarkStrategy` that includes a custom Python `TimestampAssigner`, the assigner was never invoked. The watermark strategy was passed through to the Java source operator, which executes the assigner on the JVM side and therefore ignores the Python implementation entirely.
   
   This change detects whether the supplied `WatermarkStrategy` carries a Python timestamp assigner. If so, it creates the source without watermarks and applies the watermark strategy downstream via `DataStream.assign_timestamps_and_watermarks`, so the Python `TimestampAssigner` is actually executed.
   
   The fix is implemented in `flink-python/pyflink/datastream/stream_execution_environment.py`, and a regression test has been added in `flink-python/pyflink/datastream/tests/test_stream_execution_environment.py` that uses `NumberSequenceSource` together with a custom `TimestampAssigner` and a `ProcessFunction` to verify the extracted timestamps.
   
   ##### Was generative AI tooling used to co-author this PR?
   
   - [X] Yes (please specify the tool below)
   - [x] No
   
   JIRA: https://issues.apache.org/jira/browse/FLINK-39725
   
   Generated-by: Claude Code


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] FLINK-39725: fix(python): apply timestamp assigner in from_source for pyflink [flink]

["flinkbot (via GitHub)" <gi...@apache.org>]

flinkbot commented on PR #28491:
URL: https://github.com/apache/flink/pull/28491#issuecomment-4754376916

   <!--
   Meta data
   {
     "version" : 1,
     "metaDataEntries" : [ {
       "hash" : "5d809c799c2b6df924320633ed01797a3db61b6a",
       "status" : "UNKNOWN",
       "url" : "TBD",
       "triggerID" : "5d809c799c2b6df924320633ed01797a3db61b6a",
       "triggerType" : "PUSH"
     } ]
   }-->
   ## CI report:
   
   * 5d809c799c2b6df924320633ed01797a3db61b6a UNKNOWN
   
   <details>
   <summary>Bot commands</summary>
     The @flinkbot bot supports the following commands:
   
    - `@flinkbot run azure` re-run the last Azure build
   </details>


-- 
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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [FLINK-39725][python] Apply the Python TimestampAssigner in from_source [flink]

["vbhanuchander-lang (via GitHub)" <gi...@apache.org>]

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: issues-unsubscribe@flink.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org