[optimised][FLINK-34252][table] Fix lastRecordTime tracking in WatermarkAssignerOperator

[pnowojski]

This is an optimised version of #24211 that supersedes it. This newer version avoids having to check current time per each record, as syscalls are quite expensive.

What is the purpose of the change

In the current implementation, the lastRecordTime variable, which tracks the time of the last received data element, is updated only when the WatermarkStatus transitions from IDLE to ACTIVE. However, it is not updated when WatermarkStatus is ACTIVE, which means even under continuous data flow, the condition (currentTime - lastRecordTime > idleTimeout) will eventually always become true, and the WatermarkStatus will erroneously be marked IDLE.

I believe this bug technically causes incorrect outputs since downstream watermarks advance earlier than they otherwise would. The incorrect state doesn't last forever, though, since when the WatermarkStatus is in in the IDLE state, the next processElement will cause a WatermarkStatus.ACTIVE to be emitted.

The new unit test illustrates the flip-flop behavior before the fix:

[ERROR] org.apache.flink.table.runtime.operators.wmassigners.WatermarkAssignerOperatorTest.testIdleStateAvoidanceWithConsistentDataFlow -- Time elapsed: 0.013 s <<< FAILURE!
java.lang.AssertionError:

Expecting
  [WatermarkStatus(IDLE),
    WatermarkStatus(ACTIVE),
    WatermarkStatus(IDLE),
    WatermarkStatus(ACTIVE),
    WatermarkStatus(IDLE),
    WatermarkStatus(ACTIVE),
    WatermarkStatus(IDLE),
    WatermarkStatus(ACTIVE),
    WatermarkStatus(IDLE)]
not to contain
  [WatermarkStatus(IDLE)]
but found
  [WatermarkStatus(IDLE)]

Brief change log

• Update lastRecordTime in table WatermarkAssignerOperator on each record to prevent the stream from incorrectly being marked idle

Verifying this change

This change added tests and can be verified as follows:

• Added test that validates the WatermarkStatus is not set to idle when records are sent more frequently than the idleTimeout

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): (yes)
• 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)

[flinkbot]

CI report:

• ec646a4 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[1996fanrui]

Thanks @pnowojski for picking up this PR!

I left some minor comments, please take a look in your free time, thanks~

[1996fanrui]

As the comment said, timestamp and now may be different. I'm curious, why using timestamp instead of now here?

If the timer is triggered later, and lastWatermarkPeriodicEmitTime + watermarkInterval may be less than now, but greater than timestamp. For this case, we could advanceWatermark, right?

It means we can use now directly. For old code, it uses currentTime as well.

[pnowojski]

Goog point I think. I will try to switch to now and let's see if some tests will complain or not.

[1996fanrui]

Suggested change

	if (processedElements != lastIdleCheckProcessedElements) {
	idleSince = now;
	}
	lastIdleCheckProcessedElements = processedElements;
	if (processedElements != lastIdleCheckProcessedElements) {
	idleSince = now;
	lastIdleCheckProcessedElements = processedElements;
	}

[pnowojski]

It doesn't matter, does it? But anyway 👍

[1996fanrui]

Thanks @pnowojski for the quick update!

LGTM assuming CI is green.

The 1.20 starts feature freeze, but this PR is a bugfix. So I think we can merge it. Also, it's better to backport this fix to 1.17, 1.18 and 1.19.

[pnowojski]

Ops, I forgot to publish responses.

[pnowojski]

It doesn't matter, does it? But anyway 👍