Ebaharloo opened a new issue, #3691: URL: https://github.com/apache/fluss/issues/3691
### Search before asking - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar. ### Fluss version 0.9.0 (latest release) ### Please describe the bug 🐞 ## Body ````markdown ### Fluss Version 0.9.1-incubating ### Deployment - Kubernetes - Flink 2.2 - Fluss Coordinator + 2 TabletServers - Nessie Catalog - Iceberg - MinIO (S3) - table.datalake.enabled=true --- ## Description We have a Fluss pipeline that continuously writes Kafka messages into a Fluss table. Example table: ```sql CREATE TABLE rlc_stream_message_v2 ( kafka_partition INT, kafka_offset BIGINT, kafka_timestamp TIMESTAMP(3), message_type STRING, payload STRING ) WITH ( 'table.log.ttl'='24 h', 'table.datalake.enabled'='true', 'table.datalake.format'='iceberg' ); ```` This table has been running for several days without any visible problems. We then create another Flink SQL job that reads from this Fluss table and writes selected records into another Fluss table. Example: ```sql INSERT INTO rlc_trade_flat SELECT ... FROM rlc_stream_message_v2 WHERE message_type='Asa.Mdp.Protobuf.Rlc.Messages.Trade'; ``` The destination table may or may not have DataLake enabled. --- ## Problem After running for some time (sometimes minutes, sometimes hours), the reader job always fails with: ``` org.apache.iceberg.exceptions.RuntimeIOException Caused by: org.apache.http.ConnectionClosedException: Premature end of Content-Length delimited message body (expected: 442,423,704; received: 135,330,000) ``` The failure always references old Iceberg parquet files. Example: ``` 442423708 bytes ``` This file is several days old. --- ## Observations ### 1) The source table continues ingesting Kafka normally. The original ingestion pipeline never fails. --- ### 2) The reader pipeline restarts. After restart it again eventually attempts to read the same historical parquet files and fails again. --- ### 3) Running ```sql SELECT * FROM rlc_stream_message_v2 /*+ OPTIONS('scan.startup.mode'='latest') */; ``` starts from current records only and returns today's data. This appears to avoid reading historical Iceberg files. --- ### 4) Running ```sql EXPLAIN PLAN FOR SELECT * FROM rlc_stream_message_v2; ``` does not indicate that Iceberg is being used. The execution plan only shows ``` TableSourceScan ``` --- ### 5) The table has ``` table.log.ttl='24 h' ``` We expected a normal Fluss scan to read only the Fluss log. Instead it appears that historical Iceberg files are also scanned. --- ### 6) Removing ``` table.datalake.enabled ``` reduces the frequency of the problem, but eventually the same exception still appears. --- ## Expected Behavior A continuous Flink SQL streaming job reading a Fluss table should continue reading from the Fluss log. Old Iceberg files should not be revisited during normal streaming unless explicitly requested. A table with ``` table.log.ttl='24 h' ``` should not require reading parquet files that are several days old. --- ## Actual Behavior The streaming reader eventually attempts to read historical Iceberg parquet files. If one of those large parquet reads fails, the whole Flink job fails and restarts. After restart the same behavior repeats. --- ## Questions 1. Is a normal Fluss streaming scan expected to read historical Iceberg files? 2. Under what conditions does Fluss switch from reading the log to reading Iceberg snapshots? 3. Does `table.log.ttl` affect only the Fluss log while snapshot planning still includes all Iceberg history? 4. Is this expected behavior or a bug? ```` --- I would add **one more piece of evidence** at the bottom before posting. It's quite compelling: ```markdown Additional Observation The source table has been ingesting continuously for multiple days without errors. Only downstream Flink SQL jobs that read from this Fluss table experience the failure. This suggests the issue occurs during Fluss table reading rather than Fluss ingestion. ```` This report gives the Fluss maintainers: * your architecture, * the exact exception, * what you've already ruled out, * your expectation, * and the specific question about why a streaming read appears to fall back to historical Iceberg files. It's focused on facts rather than assumptions, which makes it much easier for maintainers to investigate. ### Solution _No response_ ### Are you willing to submit a PR? - [ ] I'm willing to submit a PR! -- 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]
