haishui created FLINK-31632:
-------------------------------
Summary: watermark aligned idle source can't resume
Key: FLINK-31632
URL: https://issues.apache.org/jira/browse/FLINK-31632
Project: Flink
Issue Type: Bug
Components: API / DataStream
Affects Versions: 1.15.4, 1.16.1, 1.17.0
Reporter: haishui
{code:java}
WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy
.<String>forBoundedOutOfOrderness(Duration.ofMillis(0))
.withTimestampAssigner((element, recordTimestamp) ->
Long.parseLong(element))
.withWatermarkAlignment("group", Duration.ofMillis(10),
Duration.ofSeconds(2))
.withIdleness(Duration.ofSeconds(10));
DataStreamSource<String> s1 = env.fromSource(kafkaSource("topic1"),
watermarkStrategy, "S1");
DataStreamSource<String> s2 = env.fromSource(kafkaSource("topic2"),
watermarkStrategy, "S2");{code}
send "0" to kafka topic1 and topic2
After 10s, source1 and source2 is idle，and logs are
{code:java}
09:44:30,403 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ -1 (1970-01-01 07:59:59.999) from subTaskId=0
09:44:30,404 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ -1 (1970-01-01 07:59:59.999) from subTaskId=0
09:44:32,019 INFO
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Distributing
maxAllowedWatermark=9 to subTaskIds=[0]
09:44:32,019 INFO
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Distributing
maxAllowedWatermark=9 to subTaskIds=[0]
09:44:32,417 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ -1 (1970-01-01 07:59:59.999) from subTaskId=0
09:44:32,418 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ -1 (1970-01-01 07:59:59.999) from subTaskId=0
09:44:34,028 INFO
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Distributing
maxAllowedWatermark=9 to subTaskIds=[0]
09:44:34,028 INFO
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Distributing
maxAllowedWatermark=9 to subTaskIds=[0]
09:44:34,423 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ 9223372036854775807 (292278994-08-17 15:12:55.807) from
subTaskId=0
09:44:34,424 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ 9223372036854775807 (292278994-08-17 15:12:55.807) from
subTaskId=0
09:44:36,023 INFO
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Distributing
maxAllowedWatermark=-9223372036854775799 to subTaskIds=[0]
09:44:36,023 INFO
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - Distributing
maxAllowedWatermark=-9223372036854775799 to subTaskIds=[0]
09:44:36,433 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ 9223372036854775807 (292278994-08-17 15:12:55.807) from
subTaskId=0
09:44:36,433 DEBUG
org.apache.flink.runtime.source.coordinator.SourceCoordinator [] - New reported
watermark=Watermark @ 9223372036854775807 (292278994-08-17 15:12:55.807) from
subTaskId=0 {code}
send message to topic1 or topic2 now, the message can't be consumed。
the reason is:
when a source is marked idle, the lastEmittedWatermark = Long.MAX_VALUE and
currentMaxDesiredWatermark = Long.MAX_VALUE + maxAllowedWatermarkDrift in
org.apache.flink.streaming.api.operators.SourceOperator.
currentMaxDesiredWatermark is negative and always less than lastEmittedWatermark
operatingMode always is WAITING_FOR_ALIGNMENT
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
