Grzegorz Liter created FLINK-32828:
--------------------------------------
Summary: Kafka-partition-aware watermark not handled correctly
shortly after job start up
Key: FLINK-32828
URL: https://issues.apache.org/jira/browse/FLINK-32828
Project: Flink
Issue Type: Bug
Components: API / DataStream, Connectors / Kafka
Affects Versions: 1.17.1
Environment: Affected environments:
* Local MiniCluster + Confluent Kafka run in docker
** See attached files
* Flink job run in Kubernetes using Flink Operator 1.15 + 3 node Kafka cluster
run in Kubernetes cluster
Reporter: Grzegorz Liter
Attachments: docker-compose.yml, test-job.java
When using KafkaSource with partition aware watermarks. Watermarks are being
emitted even when only one partition has some events just after job startup
from savepoint/checkpoint. After it has some events on other partitions the
watermark behaviour is correct and watermark is emited as a minimum watarmark
from each partition.
Steps to reproduce:
# Setup a Kafka cluster with a topic that has 2 or more partitions.
## {{./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic
test-2 --partitions 4}}
# Create a job that (see attached `test-job.java`):
## uses a KafkaSource with
`WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10L)`
## has parallelism lower than number of partitions
## stores checkpoint/savepoint
# Start job
# Send events only on single partition
## {{./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic
test-2 --property "parse.key=true" --property "key.separator=:"}}
{{14:51:19,883 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:51:18.849Z, watermark
-292275055-05-16T16:47:04.192Z}}
{{14:51:32,484 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:51:31.475Z, watermark
-292275055-05-16T16:47:04.192Z}}
{{14:51:35,914 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:51:34.909Z, watermark
-292275055-05-16T16:47:04.192Z}}
Expected: Watermark does not progress. Actual: Watermark does not progress.
5. Stop the job
6. Startup job from last checkpoint/savepoint
7. Send events only on single partitions
{{14:53:41,693 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:53:40.662Z, watermark
-292275055-05-16T16:47:04.192Z}}
{{14:53:46,088 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:53:45.078Z, watermark
2023-08-10T12:53:30.661Z}}
{{14:53:49,520 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:53:48.511Z, watermark
2023-08-10T12:53:35.077Z}}
Expected: Watermark does not progress. {color:#FF0000}*Actual: Watermark has
progress*{color}
{color:#172b4d}To add bit more of context:{color}
{color:#172b4d}8. Send events on other partitions and then send events only on
single partitions{color}
{{{color:#172b4d}14:54:55,112 WARN com.example.TestJob6$InputSink2
[] - == Received: test-2/0: 2 -> a, timestamp 2023-08-10T12:54:54.104Z,
watermark 2023-08-10T12:53:38.510Z
14:54:57,673 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/1: 4 -> a, timestamp 2023-08-10T12:54:56.665Z, watermark
2023-08-10T12:53:38.510Z
14:54:57,673 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/2: 5 -> a, timestamp 2023-08-10T12:54:57.554Z, watermark
2023-08-10T12:53:38.510Z
14:55:12,821 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:55:11.814Z, watermark
2023-08-10T12:54:44.103Z
14:55:16,099 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:55:15.091Z, watermark
2023-08-10T12:54:44.103Z
14:55:19,122 WARN com.example.TestJob6$InputSink2 [] - ==
Received: test-2/3: 1 -> a, timestamp 2023-08-10T12:55:18.114Z, watermark
2023-08-10T12:54:44.103Z
{color}}}
{color:#172b4d}Expected: Watermark should progress a bit and then should not
progress when receiving events only on single partition. {color}
{color:#172b4d}Actual: As expected{color}
{color:#172b4d}This behavior also shows as a burst of late events just after
startup and then no more late events when job operates normally. {color}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
