Skip to content

Conversation

@X-czh
Copy link
Contributor

@X-czh X-czh commented Jan 16, 2026

What is the purpose of the change

Fix incorrect scheduled timestamp in ProcessingTimeCallback with scheduleWithFixedDelay.

Verifying this change

This change is a trivial rework.

Does this pull request potentially affect one of the following parts:

  • Dependencies (does it add or upgrade a dependency): (yes / no)
  • The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
  • The serializers: (yes / no / don't know)
  • The runtime per-record code paths (performance sensitive): (yes / no / don't know)
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
  • The S3 file system connector: (yes / no / don't know)

Documentation

  • Does this pull request introduce a new feature? (yes / no)
  • If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

@X-czh
Copy link
Contributor Author

X-czh commented Jan 16, 2026

@Zakelly @xiangforever2014 Would you mind taking a look?

@flinkbot
Copy link
Collaborator

flinkbot commented Jan 16, 2026

CI report:

Bot commands The @flinkbot bot supports the following commands:
  • @flinkbot run azure re-run the last Azure build

ProcessingTimeCallback callback, long initialDelay, long period, boolean fixedDelay) {
final long nextTimestamp = getCurrentProcessingTime() + initialDelay;
final Runnable task = wrapOnTimerCallback(callback, nextTimestamp, period);
final Runnable task = wrapOnTimerCallback(callback, nextTimestamp, period, fixedDelay);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please could you add a test for this change

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure

ProcessingTimeCallback callback, long initialDelay, long period, boolean fixedDelay) {
final long nextTimestamp = getCurrentProcessingTime() + initialDelay;
final Runnable task = wrapOnTimerCallback(callback, nextTimestamp, period);
final Runnable task = wrapOnTimerCallback(callback, nextTimestamp, period, fixedDelay);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am checking the logic on the first time - I would expect the initial delay to take effect, then subsequent time the fixed delay. As the code is written the first time we get the initial plus the fixed delay, can you confirm this is what you want?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The initial delay is accounted when calling scheduleRepeatedly, where nextTimestamp is initialized to getCurrentProcessingTime() + initialDelay

@github-actions github-actions bot added the community-reviewed PR has been reviewed by the community. label Jan 16, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

community-reviewed PR has been reviewed by the community.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants