NicoK commented on a change in pull request #31:
URL: https://github.com/apache/flink-training/pull/31#discussion_r694703353
##########
File path:
long-ride-alerts/src/solution/java/org/apache/flink/training/solutions/longrides/LongRidesSolution.java
##########
@@ -71,36 +109,53 @@ public void open(Configuration config) {
}
@Override
- public void processElement(TaxiRide ride, Context context,
Collector<TaxiRide> out)
+ public void processElement(TaxiRide ride, Context context,
Collector<Long> out)
throws Exception {
- TaxiRide previousRideEvent = rideState.value();
- if (previousRideEvent == null) {
+ TaxiRide firstRideEvent = rideState.value();
+
+ if (firstRideEvent == null) {
rideState.update(ride);
+
if (ride.isStart) {
context.timerService().registerEventTimeTimer(getTimerTime(ride));
+ } else {
+ if (rideTooLong(ride)) {
+ out.collect(ride.rideId);
+ }
}
} else {
- if (!ride.isStart) {
- // it's an END event, so event saved was the START event
and has a timer
- // the timer hasn't fired yet, and we can safely kill the
timer
-
context.timerService().deleteEventTimeTimer(getTimerTime(previousRideEvent));
+ if (ride.isStart) {
+ // There's nothing to do but clear the state (which is
done below).
+ } else {
+ // There may be a timer that hasn't fired yet.
+
context.timerService().deleteEventTimeTimer(getTimerTime(firstRideEvent));
+
+ // It could be that the ride has gone on too long, but the
timer hasn't fired.
Review comment:
let me unresolve this since it isn't in the PR yet
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
