DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
Real life use cases can be taken from the energy sector. The automatic closed loop balancing of the power grid requires calculations to be published at the start of every 10th second. Today, this is solved by a work-around utilising a "has published this interval"-flag, and a state store to ensure a punctuation is only triggered once per interval:
| Code Block | ||||
|---|---|---|---|---|
| ||||
class PunctuateProcessor(
private val hasForwardedStoreName: String,
private val forwardTime: Duration,
) : ContextualProcessor<String, avro_value, String, avro_value>(),
ILogging by Logging<PunctuateProcessor>() {
private lateinit var hasForwardStore: WindowStore<String, Boolean>
private lateinit var forwardSchedule: Cancellable
private val hasForwardStoreKey = "hasPublished"
override fun init(context: ProcessorContext<String, avro_value>) {
super.init(context)
this.hasForwardStore = context.getStateStore(hasForwardedStoreName)
forwardSchedule =
context().schedule(Duration.ofMillis(500), PunctuationType.WALL_CLOCK_TIME) { forwardRecordsIfTime() }
}
override fun process(record: Record<String, avro_value>) {
// Store incoming records
}
private fun forwardRecordsIfTime() {
val currentTime = Duration.ofMillis(context().currentSystemTimeMs())
val flooredTime = Duration.ofMillis(floorTo10Second(currentTime.toMillis()))
if (isTimeToForward(currentTime) && !hasForwardedThisInterval(flooredTime)) {
forwardRecords()
hasForwardStore.put(hasForwardStoreKey, true, flooredTime.toMillis())
}
}
private fun isTimeToForward(currentTime: Duration): Boolean = (currentTime.toSecondsPart() % 10) >= (forwardTime.toSecondsPart() % 10)
private fun hasForwardedThisInterval(intervalStart: Duration): Boolean =
hasForwardStore.fetch(hasForwardStoreKey, intervalStart.toMillis()) ?: false |
...
| Code Block | ||
|---|---|---|
| ||
package org.apache.kafka.streams.processor.api;
public interface ProcessingContext {
/* code */
// New method allowing for anchored punctuation
Cancellable schedule(final Duration interval, final long startTime, final PunctuationType type, final Punctuator callback);
// Existing method
Cancellable schedule(final Duration interval, final long startTime, final PunctuationType type, final Punctuator callback) {
schedule(interval, null, type, callback);
}
} |
...
The `startTime`, together with the `interval` and the wall clock, will determine the next trigger time for the callback. Hence, the anchored punctuation is only supporting wall clock (i.e. stream time) in this first iteration. The method for calculating the next, anchored trigger time, could look something like:
| Code Block | ||||
|---|---|---|---|---|
| ||||
long currentTime = System.currentTimeMillis();
// If currentTime is before startTime, return the difference between startTime and currentTime
if (currentTime < startTime) {
return startTime - currentTime;
}
// Calculate how many intervals have passed since startTime
long elapsedTime = currentTime - startTime;
// Calculate how many full intervals have passed
long intervalsPassed = elapsedTime / interval;
// Calculate the time of the next trigger time
long nextTriggerTime = startTime + (intervalsPassed + 1) * interval;
// If the current time is already a trigger time, return 0
return nextTriggerTime - currentTime; |
Compatibility, Deprecation, and Migration Plan
...
The plan is to test the new schedule option in the same way that the current schedule options are tested. The anchored wall-clock punctuation is a new feature, and the feature should therefore not affect any current features or users.
Rejected Alternatives
Cron job
Creating a new schedule method that takes in a cron job expression as a parameter:
| Code Block | ||||
|---|---|---|---|---|
| ||||
package org.apache.kafka.streams.processor.api;
public interface ProcessingContext {
// New method allowing for anchored punctuation using cron expressions
Cancellable schedule(final String cronExpression, final Punctuator callback);
// Existing method
Cancellable schedule(final Duration interval, final long startTime, final PunctuationType type, final Punctuator callback);
} |
The usage of cron expressions would require the inclusion of a new dependency, such as Quartz. Generally, we wish to avoid bringing in new dependencies if it is possible to avoid. Also, it is wise to start simple when implementing a new future, and then build up the feature incrementally. If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.