Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Mar 10, 2025
Merged

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

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
618d50a
Move ImpulseSource to new Source API
Poorvankbhatia Feb 27, 2025
de5fb01
Added DataGeneratorSource to replace Impulse source.
Poorvankbhatia Feb 28, 2025
0d2bd2d
Added a descriptive comment.
Poorvankbhatia Feb 28, 2025
eb8ef85
Corrected comment logic, based on the review.
Poorvankbhatia Mar 6, 2025
de713b7
Adjusted calculation based on comment.
Poorvankbhatia Mar 7, 2025
fabd090
Addressed comment
Poorvankbhatia Mar 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions examples/autoscaling/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,13 @@ under the License.
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-datagen</artifactId>
<version>${flink.version}</version>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
Expand Down
63 changes: 36 additions & 27 deletions examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,16 @@

package autoscaling;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import org.slf4j.Logger;
Expand Down Expand Up @@ -74,8 +78,38 @@ public static void main(String[] args) throws Exception {
for (String branch : maxLoadPerTask.split("\n")) {
String[] taskLoads = branch.split(";");

/*
* Creates an unbounded stream that continuously emits the constant value 42L.
* Flink's DataGeneratorSource with RateLimiterStrategy is used to control the emission rate.
*
* Rate Calculation:
* - samplingIntervalMs / 10 gives maxSleepTimeMs, which represents the interval between emissions.
Copy link
Contributor

@afedulov afedulov Mar 5, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

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).

Copy link
Contributor Author

@Poorvankbhatia Poorvankbhatia Mar 6, 2025

Choose a reason for hiding this comment

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

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

* - To determine the number of records emitted per second:
* 1000 / maxSleepTimeMs
* Since 1000 ms equals 1 second, this formula calculates the emission rate.
*
* Example:
* - If samplingIntervalMs = 1000 ms:
* maxSleepTimeMs = 100 ms
* - This results in:
* 1000 ms / 100 ms = 10 records per second.
*
* RateLimiterStrategy.perSecond((double) 1000 / ((double) samplingIntervalMs / 10))
* ensures this rate is maintained efficiently without blocking execution.
*/
DataStream<Long> stream =
env.addSource(new ImpulseSource(samplingIntervalMs)).name("ImpulseSource");
env.fromSource(
new DataGeneratorSource<>(
(GeneratorFunction<Long, Long>)
(index) -> 42L, // Emits constant value 42
Long.MAX_VALUE, // Unbounded stream
RateLimiterStrategy.perSecond(
(double) 1000
Copy link
Contributor

@afedulov afedulov Mar 7, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

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

Copy link
Contributor Author

@Poorvankbhatia Poorvankbhatia Mar 7, 2025

Choose a reason for hiding this comment

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

Corrected this.

/ ((double) samplingIntervalMs
/ 10)), // Controls rate
Types.LONG),
WatermarkStrategy.noWatermarks(),
"ImpulseSource (Using DataGeneratorSource)");

for (String load : taskLoads) {
double maxLoad = Double.parseDouble(load);
Expand All @@ -97,31 +131,6 @@ public static void main(String[] args) throws Exception {
+ ")");
}

private static class ImpulseSource implements SourceFunction<Long> {
private final int maxSleepTimeMs;
volatile boolean canceled;

public ImpulseSource(int samplingInterval) {
this.maxSleepTimeMs = samplingInterval / 10;
}

@Override
public void run(SourceContext<Long> sourceContext) throws Exception {
while (!canceled) {
synchronized (sourceContext.getCheckpointLock()) {
sourceContext.collect(42L);
}
// Provide an impulse to keep the load simulation active
Thread.sleep(maxSleepTimeMs);
}
}

@Override
public void cancel() {
canceled = true;
}
}

private static class LoadSimulationFn extends RichFlatMapFunction<Long, Long> {

private final double maxLoad;
Expand Down