You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "MartijnVisser (via GitHub)" <gi...@apache.org> on 2026/07/03 14:39:16 UTC
[PR] [FLINK-40064][python][e2e] Migrate PyFlink DataStream e2e test from Kafka to FileSource/FileSink [flink]
MartijnVisser opened a new pull request, #28629:
URL: https://github.com/apache/flink/pull/28629
## What is the purpose of the change
FLINK-40048 removed the flink-sql-connector-kafka dependency and its maven-dependency-plugin copy from flink-sql-client-test, so no Kafka sql-jar is staged into target/sql-jars anymore. test_pyflink.sh still executed `KAFKA_SQL_JAR=$(find "$SQL_JARS_DIR" | grep "kafka")` under `set -Eeuo pipefail`, which now exits 1 and aborts the "PyFlink end-to-end test" nightly leg (e2e_4_ci) before any test runs (first seen in build 76685).
The jar and all Kafka scaffolding in the script only served the "Test PyFlink DataStream job" case, disabled since FLINK-36185 because data_stream_job.py used the legacy FlinkKafkaConsumer/FlinkKafkaProducer. This PR rewrites that job to the in-repo filesystem connector and re-enables the case, fixing the nightly leg and restoring PyFlink DataStream coverage (KeyedProcessFunction + event-time timers) dormant since September 2024. No active Kafka e2e coverage is lost; PyFlink-Kafka e2e coverage belongs in the externalized flink-connector-kafka repository.
## Brief change log
- Rewrite data_stream_job.py onto the bounded FileSource and unified FileSink; JSON lines are parsed in a map function
- Disable periodic watermarks so all timers fire at MIN_VALUE + 1500 on the end-of-input MAX_WATERMARK, making the output deterministic (the same output the Kafka-based test asserted)
- Use blocking env.execute(); the committer commits all pending part files at end of input, so no checkpointing is needed
- Re-enable the DataStream case in test_pyflink.sh with a sorted diff over the committed part files, and remove the Kafka scaffolding (kafka_sql_common.sh stays; test_confluent_schema_registry.sh still sources it)
## Verifying this change
This change added tests and can be verified as follows:
- Ran the rewritten job twice standalone on a local minicluster (source-built PyFlink, Python 3.12); both runs produced exactly the 16 expected lines, byte-identical, with a cleanly committed part file and no in-progress leftovers
- The re-enabled case fails loudly on missing or wrong output (empty part-file set diffs against the expected lines and exits 1)
- The nightly e2e_4 leg runs the full script in CI
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Claude Code (Fable 5))
Generated-by: Claude Code (Fable 5)
--
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-40064][python][e2e] Migrate PyFlink DataStream e2e test from Kafka to FileSource/FileSink [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser merged PR #28629:
URL: https://github.com/apache/flink/pull/28629
--
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-40064][python][e2e] Migrate PyFlink DataStream e2e test from Kafka to FileSource/FileSink [flink]
Posted by "autophagy (via GitHub)" <gi...@apache.org>.
autophagy commented on code in PR #28629:
URL: https://github.com/apache/flink/pull/28629#discussion_r3520680940
##########
flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py:
##########
@@ -16,45 +16,60 @@
# limitations under the License.
################################################################################
+import json
+import sys
from typing import Any
-from pyflink.common import Duration
-from pyflink.common.serialization import SimpleStringSchema
+from pyflink.common import Duration, Encoder, Row
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import TimestampAssigner, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
-from pyflink.datastream.connectors import FlinkKafkaProducer, FlinkKafkaConsumer
-from pyflink.datastream.formats.json import JsonRowDeserializationSchema
+from pyflink.datastream.connectors.file_system import (FileSink, FileSource, RollingPolicy,
+ StreamFormat)
from pyflink.datastream.functions import KeyedProcessFunction
from functions import MyKeySelector
-def python_data_stream_example():
+def python_data_stream_example(input_path: str, output_path: str):
env = StreamExecutionEnvironment.get_execution_environment()
- # Set the parallelism to be one to make sure that all data including fired timer and normal data
- # are processed by the same worker and the collected result would be in order which is good for
- # assertion.
+ # Process everything on one worker so the timer behavior is deterministic and the output
+ # lands in a single part file.
env.set_parallelism(1)
+ # No periodic watermarks: current_watermark() stays at Long.MIN_VALUE until the bounded
+ # source emits MAX_WATERMARK at end of input, making the fired-timer output deterministic.
+ env.get_config().set_auto_watermark_interval(0)
type_info = Types.ROW_NAMED(['createTime', 'orderId', 'payAmount', 'payPlatform', 'provinceId'],
[Types.LONG(), Types.LONG(), Types.DOUBLE(), Types.INT(),
Types.INT()])
- json_row_schema = JsonRowDeserializationSchema.builder().type_info(type_info).build()
- kafka_props = {'bootstrap.servers': 'localhost:9092', 'group.id': 'pyflink-e2e-source'}
- kafka_consumer = FlinkKafkaConsumer("timer-stream-source", json_row_schema, kafka_props)
- kafka_producer = FlinkKafkaProducer("timer-stream-sink", SimpleStringSchema(), kafka_props)
+ source = FileSource.for_record_stream_format(StreamFormat.text_line_format(), input_path) \
+ .process_static_file_set().build()
- watermark_strategy = WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5))\
- .with_timestamp_assigner(KafkaRowTimestampAssigner())
+ watermark_strategy = WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5)) \
Review Comment:
Does setting the auto watermark interval above 0 with a wm strategy with a 5 second duration do the same thing as `WatermarkStrategy.no_watermarks()`? Could be a simplification if so, and wouldnt need to set the env config?
--
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-40064][python][e2e] Migrate PyFlink DataStream e2e test from Kafka to FileSource/FileSink [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28629:
URL: https://github.com/apache/flink/pull/28629#issuecomment-4877427956
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "398e57ad93467bb4ba820ee82acef5d04ed19409",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "398e57ad93467bb4ba820ee82acef5d04ed19409",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 398e57ad93467bb4ba820ee82acef5d04ed19409 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-40064][python][e2e] Migrate PyFlink DataStream e2e test from Kafka to FileSource/FileSink [flink]
Posted by "MartijnVisser (via GitHub)" <gi...@apache.org>.
MartijnVisser commented on code in PR #28629:
URL: https://github.com/apache/flink/pull/28629#discussion_r3520734042
##########
flink-end-to-end-tests/flink-python-test/python/datastream/data_stream_job.py:
##########
@@ -16,45 +16,60 @@
# limitations under the License.
################################################################################
+import json
+import sys
from typing import Any
-from pyflink.common import Duration
-from pyflink.common.serialization import SimpleStringSchema
+from pyflink.common import Duration, Encoder, Row
from pyflink.common.typeinfo import Types
from pyflink.common.watermark_strategy import TimestampAssigner, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
-from pyflink.datastream.connectors import FlinkKafkaProducer, FlinkKafkaConsumer
-from pyflink.datastream.formats.json import JsonRowDeserializationSchema
+from pyflink.datastream.connectors.file_system import (FileSink, FileSource, RollingPolicy,
+ StreamFormat)
from pyflink.datastream.functions import KeyedProcessFunction
from functions import MyKeySelector
-def python_data_stream_example():
+def python_data_stream_example(input_path: str, output_path: str):
env = StreamExecutionEnvironment.get_execution_environment()
- # Set the parallelism to be one to make sure that all data including fired timer and normal data
- # are processed by the same worker and the collected result would be in order which is good for
- # assertion.
+ # Process everything on one worker so the timer behavior is deterministic and the output
+ # lands in a single part file.
env.set_parallelism(1)
+ # No periodic watermarks: current_watermark() stays at Long.MIN_VALUE until the bounded
+ # source emits MAX_WATERMARK at end of input, making the fired-timer output deterministic.
+ env.get_config().set_auto_watermark_interval(0)
type_info = Types.ROW_NAMED(['createTime', 'orderId', 'payAmount', 'payPlatform', 'provinceId'],
[Types.LONG(), Types.LONG(), Types.DOUBLE(), Types.INT(),
Types.INT()])
- json_row_schema = JsonRowDeserializationSchema.builder().type_info(type_info).build()
- kafka_props = {'bootstrap.servers': 'localhost:9092', 'group.id': 'pyflink-e2e-source'}
- kafka_consumer = FlinkKafkaConsumer("timer-stream-source", json_row_schema, kafka_props)
- kafka_producer = FlinkKafkaProducer("timer-stream-sink", SimpleStringSchema(), kafka_props)
+ source = FileSource.for_record_stream_format(StreamFormat.text_line_format(), input_path) \
+ .process_static_file_set().build()
- watermark_strategy = WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5))\
- .with_timestamp_assigner(KafkaRowTimestampAssigner())
+ watermark_strategy = WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5)) \
Review Comment:
Almost: with the interval above 0 they are not quite equivalent, because the bounded-out-of-orderness generator emits periodically, so on a slow read the watermark could advance mid-stream and change which timestamps the timers register at. But you are right that `no_watermarks()` is the cleaner way to get exactly the interval-0 behavior: it never emits during processing, the timestamp assigner still chains onto it (needed for the `ctx.timestamp()` assertions), and the end-of-input `MAX_WATERMARK` is forwarded by `TimestampsAndWatermarksOperator` regardless of the strategy, so the timers still fire. Applied in fe0a7c4c337 and re-verified locally (same 16 lines, identical across runs).
--
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