[FLINK-33525] Move ImpulseSource to new Source API #950

Poorvankbhatia · 2025-02-27T08:37:04Z

What is the purpose of the change

This pull request migrates the ImpulseSource (used in LoadSimulationPipeline) to use Flink's DataGeneratorSource, replacing the older SourceFunction.

Brief change log

Replaced the old ImpulseSource (which implemented SourceFunction) with a new implementation using Flink's DataGeneratorSource.
Updated LoadSimulationPipeline to use env.fromSource(...) instead of env.addSource(...).
Ensured that the new implementation preserves the same functionality as the old one.

Verifying this change

This change is already covered by existing tests, such as:

Existing load simulation tests that verify the source is producing expected records.
The Flink job runs with the updated ImpulseSource without functional changes.
The change was manually verified by running a Flink job with the new ImpulseSource.

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 changes to the CustomResourceDescriptors: no
Core observer or reconciler logic that is regularly executed: no

Documentation

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

afedulov

@Poorvankbhatia thanks for the PR. Did you evaluate if this can be achieved based off of the existing DataGeneratorSource/GeneratorFunction? Generally, our approach should be to try to capture usage patterns that might also be required for generating synthetic data by Flink users and make them available by providing utilities in the generator package.

Poorvankbhatia · 2025-02-28T14:31:52Z

@Poorvankbhatia thanks for the PR. Did you evaluate if this can be achieved based off of the existing DataGeneratorSource/GeneratorFunction? Generally, our approach should be to try to capture usage patterns that might also be required for generating synthetic data by Flink users and make them available by providing utilities in the generator package.

Yes, absolutely! However, using DataGeneratorSource requires adding a dependency on flink-connector-datagen in autoscaling's pom.xml. I wasn't sure if that was acceptable, so I replicated the original source instead. Let me know if i can add that dependency.

afedulov · 2025-02-28T15:55:11Z

However, using DataGeneratorSource requires adding a dependency on flink-connector-datagen in autoscaling's pom.xml.

I think it is fine - datagen's only major transitive dependency is on flink-core which is already present in the operator.
@gyfora what do you think?

gyfora · 2025-02-28T15:58:47Z

Since this is only an example we should focus on simplicity and add the extra dep to use the built in generator functionality. We should not add new dependencies to the autoscaler, only this example module

Poorvankbhatia · 2025-02-28T19:06:28Z

However, using DataGeneratorSource requires adding a dependency on flink-connector-datagen in autoscaling's pom.xml.

I think it is fine - datagen's only major transitive dependency is on flink-core which is already present in the operator. @gyfora what do you think?

Updated the code. Please have a look. Thanks.

afedulov · 2025-03-05T15:08:10Z

examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java

+             * Flink's DataGeneratorSource with RateLimiterStrategy is used to control the emission rate.
+             *
+             * Rate Calculation:
+             * - samplingIntervalMs / 10 gives maxSleepTimeMs, which represents the interval between emissions.


I believe this section needs to be adjusted. Since we do not control the maxSleepTimeMs directly, but it is rather controlled internally by the Guava's token bucket algorithm, it should not be explicitly mentioned here. It is also not strictly guaranteed to be max - at the startup there can be a short burst. The value 10 seems to just be a hardcoded parameter meaning that we want at least 10 impulses per sampling interval. Basically it should rather explain: check how many sampling intervals are there within a second, make sure that 10 impulses are generated for each sampling interval (IMPULSES_PER_SAMPLING_INTERVAL).

Updated the comment. Let me know if this makes sense.

afedulov · 2025-03-07T13:01:11Z

examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java

+             *      - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+             *      - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+             *
+             * This approach ensures that the number of records emitted dynamically scales


This sounds very AI-generated :) I the intention is clear without the last paragraph.
Also, one example should suffice.

afedulov · 2025-03-07T13:02:36Z

examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java

+                                            (index) -> 42L, // Emits constant value 42
+                                    Long.MAX_VALUE, // Unbounded stream
+                                    RateLimiterStrategy.perSecond(
+                                            (double) 1000


More importantly, please actually adjust the calculation to align with the comments description
a/b/c -> (a/b)*c
c: 10 -> IMPULSES_PER_SAMPLING_INTERVAL

Corrected this.

afedulov · 2025-03-10T16:18:24Z

examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java

+             * Emission Rate Logic:
+             * - The goal is to generate a fixed number of impulses per sampling interval.
+             * - `samplingIntervalMs` defines the duration of one sampling interval in milliseconds.
+             * - We define `IMPULSES_PER_SAMPLING_INTERVAL = 10`, meaning that for every sampling interval,


nit: either introduce this constant and use it in the calculations or use plain text in the comment.

Added the constant.

afedulov

Looks good, thanks @Poorvankbhatia

Move ImpulseSource to new Source API

618d50a

Poorvankbhatia changed the title ~~Move ImpulseSource to new Source API~~ [FLINK-33525] Move ImpulseSource to new Source API Feb 27, 2025

afedulov reviewed Feb 28, 2025

View reviewed changes

afedulov requested a review from mxm February 28, 2025 14:16

Poorvankbhatia added 2 commits February 28, 2025 20:10

Added DataGeneratorSource to replace Impulse source.

de5fb01

Added a descriptive comment.

0d2bd2d

afedulov reviewed Mar 5, 2025

View reviewed changes

Corrected comment logic, based on the review.

eb8ef85

afedulov reviewed Mar 7, 2025

View reviewed changes

Adjusted calculation based on comment.

de713b7

afedulov reviewed Mar 10, 2025

View reviewed changes

Addressed comment

fabd090

afedulov approved these changes Mar 10, 2025

View reviewed changes

afedulov merged commit 48359a3 into apache:main Mar 10, 2025
1 of 16 checks passed

Poorvankbhatia deleted the FLINK-33525-NewImpulseSource branch April 21, 2025 07:00

[FLINK-33525] Move ImpulseSource to new Source API #950

[FLINK-33525] Move ImpulseSource to new Source API #950

Uh oh!

Conversation

Poorvankbhatia commented Feb 27, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

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

Documentation

Uh oh!

afedulov left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Poorvankbhatia commented Feb 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

afedulov commented Feb 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gyfora commented Feb 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Poorvankbhatia commented Feb 28, 2025

Uh oh!

afedulov Mar 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Poorvankbhatia Mar 6, 2025

Choose a reason for hiding this comment

Uh oh!

afedulov Mar 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Poorvankbhatia Mar 7, 2025

Choose a reason for hiding this comment

Uh oh!

afedulov Mar 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Poorvankbhatia Mar 7, 2025

Choose a reason for hiding this comment

Uh oh!

afedulov Mar 10, 2025

Choose a reason for hiding this comment

Uh oh!

Poorvankbhatia Mar 10, 2025

Choose a reason for hiding this comment

Uh oh!

afedulov left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Poorvankbhatia commented Feb 27, 2025 •

edited

Loading

afedulov left a comment •

edited

Loading

Poorvankbhatia commented Feb 28, 2025 •

edited

Loading

afedulov commented Feb 28, 2025 •

edited

Loading

gyfora commented Feb 28, 2025 •

edited

Loading

afedulov Mar 5, 2025 •

edited

Loading

afedulov Mar 7, 2025 •

edited

Loading

afedulov Mar 7, 2025 •

edited

Loading