You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Poorvankbhatia (via GitHub)" <gi...@apache.org> on 2025/02/27 08:37:07 UTC
[PR] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Poorvankbhatia opened a new pull request, #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950
## What is the purpose of the change
This pull request migrates the ImpulseSource to use Flink's new Source API, replacing the older SourceFunction.
## Brief change log
- Replaced the old ImpulseSource (which implemented SourceFunction<Long>) with a new implementation using Flink's modern Source API.
- Implemented createReader, createEnumerator, and serialization methods in the new source.
- 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "Poorvankbhatia (via GitHub)" <gi...@apache.org>.
Poorvankbhatia commented on PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#issuecomment-2690791512
> @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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1981589005
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -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.
Review Comment:
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 hardcoded value 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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#issuecomment-2691001489
> 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` and which is already present in the operator.
@gyfora what do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#issuecomment-2691008846
Since this is only an example we should focus on simplicity and add the extra dep to use the built in generator functionality.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1985025598
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -82,20 +82,32 @@ public static void main(String[] args) throws Exception {
* 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.
- * - To determine the number of records emitted per second:
- * 1000 / maxSleepTimeMs
- * Since 1000 ms equals 1 second, this formula calculates the emission rate.
+ * 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,
+ * exactly 10 impulses should be generated.
*
- * Example:
- * - If samplingIntervalMs = 1000 ms:
- * maxSleepTimeMs = 100 ms
- * - This results in:
- * 1000 ms / 100 ms = 10 records per second.
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
*
- * RateLimiterStrategy.perSecond((double) 1000 / ((double) samplingIntervalMs / 10))
- * ensures this rate is maintained efficiently without blocking execution.
+ * Example Calculations:
+ * - If `samplingIntervalMs = 1000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 1000 = 1`
+ * - `impulsesPerSecond = 10 * 1 = 10 records per second`
+ * - If `samplingIntervalMs = 500 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 500 = 2`
+ * - `impulsesPerSecond = 10 * 2 = 20 records per second`
+ * - If `samplingIntervalMs = 2000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+ * - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+ *
+ * This approach ensures that the number of records emitted dynamically scales
Review Comment:
This sounds very AI-generated :) I the intention is clear without the last paragraph.
Also, one example would suffice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "Poorvankbhatia (via GitHub)" <gi...@apache.org>.
Poorvankbhatia commented on PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#issuecomment-2691360898
> > 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "Poorvankbhatia (via GitHub)" <gi...@apache.org>.
Poorvankbhatia commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1987657536
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,37 @@ 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.
+ *
+ * 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,
Review Comment:
Added the constant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov merged PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1987622599
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,37 @@ 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.
+ *
+ * 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,
+ * exactly 10 impulses should be generated.
+ *
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
+ *
+ * Example:
+ * - If `samplingIntervalMs = 500 ms` and `IMPULSES_PER_SAMPLING_INTERVAL = 10`:
Review Comment:
nit: either introduce this constant and use it in the calculations or use plain text in the comment.
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,37 @@ 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.
+ *
+ * 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,
Review Comment:
nit: either introduce this constant and use it in the calculations or use plain text in the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1987622599
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,37 @@ 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.
+ *
+ * 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,
+ * exactly 10 impulses should be generated.
+ *
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
+ *
+ * Example:
+ * - If `samplingIntervalMs = 500 ms` and `IMPULSES_PER_SAMPLING_INTERVAL = 10`:
Review Comment:
nit: either introduce this constant and use it in the calculations or use plain text in the comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1985027312
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,50 @@ 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.
+ *
+ * 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,
+ * exactly 10 impulses should be generated.
+ *
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
+ *
+ * Example Calculations:
+ * - If `samplingIntervalMs = 1000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 1000 = 1`
+ * - `impulsesPerSecond = 10 * 1 = 10 records per second`
+ * - If `samplingIntervalMs = 500 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 500 = 2`
+ * - `impulsesPerSecond = 10 * 2 = 20 records per second`
+ * - If `samplingIntervalMs = 2000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+ * - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+ *
+ * This approach ensures that the number of records emitted dynamically scales
+ * based on the sampling interval while maintaining the target of 10 impulses per interval.
+ * RateLimiterStrategy internally distributes these emissions efficiently over time.
+ */
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
Review Comment:
More importantly, actually adjust the calculation to align with the description
a/b/c -> (a/b)*c
c: 10 -> IMPULSES_PER_SAMPLING_INTERVAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1985027312
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,50 @@ 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.
+ *
+ * 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,
+ * exactly 10 impulses should be generated.
+ *
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
+ *
+ * Example Calculations:
+ * - If `samplingIntervalMs = 1000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 1000 = 1`
+ * - `impulsesPerSecond = 10 * 1 = 10 records per second`
+ * - If `samplingIntervalMs = 500 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 500 = 2`
+ * - `impulsesPerSecond = 10 * 2 = 20 records per second`
+ * - If `samplingIntervalMs = 2000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+ * - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+ *
+ * This approach ensures that the number of records emitted dynamically scales
+ * based on the sampling interval while maintaining the target of 10 impulses per interval.
+ * RateLimiterStrategy internally distributes these emissions efficiently over time.
+ */
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
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "Poorvankbhatia (via GitHub)" <gi...@apache.org>.
Poorvankbhatia commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1982742633
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -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.
Review Comment:
Updated the comment. Let me know if this makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1985025598
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -82,20 +82,32 @@ public static void main(String[] args) throws Exception {
* 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.
- * - To determine the number of records emitted per second:
- * 1000 / maxSleepTimeMs
- * Since 1000 ms equals 1 second, this formula calculates the emission rate.
+ * 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,
+ * exactly 10 impulses should be generated.
*
- * Example:
- * - If samplingIntervalMs = 1000 ms:
- * maxSleepTimeMs = 100 ms
- * - This results in:
- * 1000 ms / 100 ms = 10 records per second.
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
*
- * RateLimiterStrategy.perSecond((double) 1000 / ((double) samplingIntervalMs / 10))
- * ensures this rate is maintained efficiently without blocking execution.
+ * Example Calculations:
+ * - If `samplingIntervalMs = 1000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 1000 = 1`
+ * - `impulsesPerSecond = 10 * 1 = 10 records per second`
+ * - If `samplingIntervalMs = 500 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 500 = 2`
+ * - `impulsesPerSecond = 10 * 2 = 20 records per second`
+ * - If `samplingIntervalMs = 2000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+ * - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+ *
+ * This approach ensures that the number of records emitted dynamically scales
Review Comment:
This sounds very AI-generated :) I the intention is clear without the last paragraph.
Also, one example should suffice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "Poorvankbhatia (via GitHub)" <gi...@apache.org>.
Poorvankbhatia commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1985081631
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -74,8 +78,50 @@ 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.
+ *
+ * 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,
+ * exactly 10 impulses should be generated.
+ *
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
+ *
+ * Example Calculations:
+ * - If `samplingIntervalMs = 1000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 1000 = 1`
+ * - `impulsesPerSecond = 10 * 1 = 10 records per second`
+ * - If `samplingIntervalMs = 500 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 500 = 2`
+ * - `impulsesPerSecond = 10 * 2 = 20 records per second`
+ * - If `samplingIntervalMs = 2000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+ * - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+ *
+ * This approach ensures that the number of records emitted dynamically scales
+ * based on the sampling interval while maintaining the target of 10 impulses per interval.
+ * RateLimiterStrategy internally distributes these emissions efficiently over time.
+ */
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
Review Comment:
Corrected this.
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -82,20 +82,32 @@ public static void main(String[] args) throws Exception {
* 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.
- * - To determine the number of records emitted per second:
- * 1000 / maxSleepTimeMs
- * Since 1000 ms equals 1 second, this formula calculates the emission rate.
+ * 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,
+ * exactly 10 impulses should be generated.
*
- * Example:
- * - If samplingIntervalMs = 1000 ms:
- * maxSleepTimeMs = 100 ms
- * - This results in:
- * 1000 ms / 100 ms = 10 records per second.
+ * To calculate the total number of records emitted per second:
+ * 1. Determine how many sampling intervals fit within one second:
+ * samplingIntervalsPerSecond = 1000 / samplingIntervalMs;
+ * 2. Multiply this by the number of impulses per interval to get the total rate:
+ * impulsesPerSecond = IMPULSES_PER_SAMPLING_INTERVAL * samplingIntervalsPerSecond;
*
- * RateLimiterStrategy.perSecond((double) 1000 / ((double) samplingIntervalMs / 10))
- * ensures this rate is maintained efficiently without blocking execution.
+ * Example Calculations:
+ * - If `samplingIntervalMs = 1000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 1000 = 1`
+ * - `impulsesPerSecond = 10 * 1 = 10 records per second`
+ * - If `samplingIntervalMs = 500 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 500 = 2`
+ * - `impulsesPerSecond = 10 * 2 = 20 records per second`
+ * - If `samplingIntervalMs = 2000 ms`:
+ * - `samplingIntervalsPerSecond = 1000 / 2000 = 0.5`
+ * - `impulsesPerSecond = 10 * 0.5 = 5 records per second`
+ *
+ * This approach ensures that the number of records emitted dynamically scales
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33525] Move ImpulseSource to new Source API [flink-kubernetes-operator]
Posted by "afedulov (via GitHub)" <gi...@apache.org>.
afedulov commented on code in PR #950:
URL: https://github.com/apache/flink-kubernetes-operator/pull/950#discussion_r1981589005
##########
examples/autoscaling/src/main/java/autoscaling/LoadSimulationPipeline.java:
##########
@@ -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.
Review Comment:
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org