-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #87 from usdot-jpo-ode/broadcast-rate-zero
Broadcast Rate Events for MAPs and SPATs that stop being broadcasted
- Loading branch information
Showing
7 changed files
with
227 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
86 changes: 86 additions & 0 deletions
86
...ava/us/dot/its/jpo/conflictmonitor/monitor/topologies/validation/BaseZeroRateChecker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| package us.dot.its.jpo.conflictmonitor.monitor.topologies.validation; | ||
|
|
||
| import org.apache.kafka.streams.KeyValue; | ||
| import org.apache.kafka.streams.processor.PunctuationType; | ||
| import org.apache.kafka.streams.processor.api.Processor; | ||
| import org.apache.kafka.streams.processor.api.ProcessorContext; | ||
| import org.apache.kafka.streams.processor.api.Record; | ||
| import org.apache.kafka.streams.state.KeyValueStore; | ||
| import org.slf4j.Logger; | ||
| import us.dot.its.jpo.conflictmonitor.monitor.models.events.ProcessingTimePeriod; | ||
| import us.dot.its.jpo.conflictmonitor.monitor.models.events.broadcast_rate.BroadcastRateEvent; | ||
| import us.dot.its.jpo.geojsonconverter.partitioner.RsuIntersectionKey; | ||
|
|
||
| import java.time.Duration; | ||
|
|
||
| /** | ||
| * Base class for zero broadcast rate checker. Maintains a state store with the latest wall clock timestamp for each | ||
| * key, and emits events if the current time is more than the rolling window period later. | ||
| * @param <TItem> Type of input items, ProcessedMap or ProcessedSpat | ||
| * @param <TEvent> Type of output events | ||
| */ | ||
| public abstract class BaseZeroRateChecker<TItem, TEvent extends BroadcastRateEvent> | ||
| implements Processor<RsuIntersectionKey, TItem, RsuIntersectionKey, TEvent> { | ||
|
|
||
| protected abstract Logger getLogger(); | ||
|
|
||
| protected abstract TEvent createEvent(); | ||
|
|
||
| private KeyValueStore<RsuIntersectionKey, Long> store; | ||
| private final int maxAgeMillis; | ||
| private final int checkEveryMillis; | ||
| private final String inputTopicName; | ||
| private final String stateStoreName; | ||
|
|
||
| ProcessorContext<RsuIntersectionKey, TEvent> context; | ||
|
|
||
| public BaseZeroRateChecker(final int rollingPeriodSeconds, final int outputIntervalSeconds, final String inputTopicName, final String stateStoreName) { | ||
| this.maxAgeMillis = rollingPeriodSeconds * 1000; | ||
| this.checkEveryMillis = outputIntervalSeconds * 1000; | ||
| this.inputTopicName = inputTopicName; | ||
| this.stateStoreName = stateStoreName; | ||
| } | ||
|
|
||
| @Override | ||
| public void init(ProcessorContext<RsuIntersectionKey, TEvent> context) { | ||
| this.context = context; | ||
| this.store = context.getStateStore(stateStoreName); | ||
| context.schedule(Duration.ofMillis(checkEveryMillis), | ||
| PunctuationType.WALL_CLOCK_TIME, | ||
| this::punctuate); | ||
| } | ||
|
|
||
| @Override | ||
| public void process(Record<RsuIntersectionKey, TItem> record) { | ||
| store.put(record.key(), System.currentTimeMillis()); | ||
| } | ||
|
|
||
| private void punctuate(long timestamp) { | ||
| // Check if any keys are older than the max age | ||
| try (var storeIterator = store.all()) { | ||
| while (storeIterator.hasNext()) { | ||
| KeyValue<RsuIntersectionKey, Long> item =storeIterator.next(); | ||
| RsuIntersectionKey key = item.key; | ||
| Long lastTimestamp = item.value; | ||
| if (timestamp - lastTimestamp > maxAgeMillis) { | ||
| emitZeroEvent(key, timestamp); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private void emitZeroEvent(RsuIntersectionKey key, long timestamp) { | ||
| getLogger().info("emit zero rate event for key = {} at timestamp = {}", key, timestamp); | ||
| TEvent event = createEvent(); | ||
| event.setSource(key.toString()); | ||
| event.setIntersectionID(key.getIntersectionId()); | ||
| event.setRoadRegulatorID(key.getRegion()); | ||
| event.setTopicName(inputTopicName); | ||
| ProcessingTimePeriod timePeriod = new ProcessingTimePeriod(); | ||
| timePeriod.setBeginTimestamp(timestamp - maxAgeMillis); | ||
| timePeriod.setEndTimestamp(timestamp); | ||
| event.setTimePeriod(timePeriod); | ||
| event.setNumberOfMessages(0); | ||
| context.forward(new Record<>(key, event, timestamp)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
31 changes: 31 additions & 0 deletions
31
...java/us/dot/its/jpo/conflictmonitor/monitor/topologies/validation/MapZeroRateChecker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| package us.dot.its.jpo.conflictmonitor.monitor.topologies.validation; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.slf4j.Logger; | ||
| import us.dot.its.jpo.conflictmonitor.monitor.models.events.broadcast_rate.MapBroadcastRateEvent; | ||
| import us.dot.its.jpo.geojsonconverter.pojos.geojson.LineString; | ||
| import us.dot.its.jpo.geojsonconverter.pojos.geojson.map.ProcessedMap; | ||
|
|
||
|
|
||
| @Slf4j | ||
| public class MapZeroRateChecker | ||
| extends BaseZeroRateChecker<ProcessedMap<LineString>, MapBroadcastRateEvent> { | ||
|
|
||
| public MapZeroRateChecker(int rollingPeriodSeconds, int outputIntervalSeconds, String inputTopicName, String stateStoreName) { | ||
| super(rollingPeriodSeconds, outputIntervalSeconds, inputTopicName, stateStoreName); | ||
| } | ||
|
|
||
| @Override | ||
| protected Logger getLogger() { | ||
| return log; | ||
| } | ||
|
|
||
| @Override | ||
| protected MapBroadcastRateEvent createEvent() { | ||
| return new MapBroadcastRateEvent(); | ||
| } | ||
|
|
||
|
|
||
|
|
||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
25 changes: 25 additions & 0 deletions
25
...ava/us/dot/its/jpo/conflictmonitor/monitor/topologies/validation/SpatZeroRateChecker.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| package us.dot.its.jpo.conflictmonitor.monitor.topologies.validation; | ||
|
|
||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.slf4j.Logger; | ||
| import us.dot.its.jpo.conflictmonitor.monitor.models.events.broadcast_rate.SpatBroadcastRateEvent; | ||
| import us.dot.its.jpo.geojsonconverter.pojos.spat.ProcessedSpat; | ||
|
|
||
| @Slf4j | ||
| public class SpatZeroRateChecker | ||
| extends BaseZeroRateChecker<ProcessedSpat, SpatBroadcastRateEvent> { | ||
|
|
||
| public SpatZeroRateChecker(int rollingPeriodSeconds, int outputIntervalSeconds, String inputTopicName, String stateStoreName) { | ||
| super(rollingPeriodSeconds, outputIntervalSeconds, inputTopicName, stateStoreName); | ||
| } | ||
|
|
||
| @Override | ||
| protected Logger getLogger() { | ||
| return log; | ||
| } | ||
|
|
||
| @Override | ||
| protected SpatBroadcastRateEvent createEvent() { | ||
| return new SpatBroadcastRateEvent(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters