kl0u commented on a change in pull request #6871: [FLINK-10583][table] Add base
TwoInputStreamOperator with TTL operator.
URL: https://github.com/apache/flink/pull/6871#discussion_r226333482
##########
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/AbstractTwoInputStreamOperatorWithTTL.scala
##########
@@ -80,51 +87,52 @@ abstract class AbstractTwoInputStreamOperatorWithTTL(
timerService = new SimpleTimerService(internalTimerService)
}
- override def processElement1(element: StreamRecord[CRow]): Unit = {
- registerProcessingCleanupTimer()
- }
-
- override def processElement2(element: StreamRecord[CRow]): Unit = {
- registerProcessingCleanupTimer()
- }
-
- private def registerProcessingCleanupTimer(): Unit = {
+ /**
+ * If the user has specified a `minRetentionTime` and `maxRetentionTime`,
this
+ * method registers a cleanup timer for `currentProcessingTime +
minRetentionTime`.
+ *
+ * <p>When this timer fires, the
[[AbstractTwoInputStreamOperatorWithTTL.cleanUpState()]]
+ * method is called.
+ */
+ protected def registerProcessingCleanUpTimer(): Unit = {
if (stateCleaningEnabled) {
val currentProcessingTime = timerService.currentProcessingTime()
- val currentCleanupTime = cleanupTimeState.value()
+ val currentCleanUpTime = Option(cleanUpTimeState.value())
- if (currentCleanupTime == null
- || (currentProcessingTime + minRetentionTime) > currentCleanupTime) {
+ if (currentCleanUpTime.isEmpty
+ || (currentProcessingTime + minRetentionTime) >
currentCleanUpTime.get) {
- updateCleanupTimer(currentProcessingTime, currentCleanupTime)
+ updateCleanUpTimer(currentProcessingTime, currentCleanUpTime)
}
}
}
- /**
- * Deletes the processing time timer with timestamp `currentCleanupTime`
(if any) and
- * registers a new one with timestamp `currentProcessingTime +
maxRetentionTime`. This
- * method is used by the `registerProcessingCleanupTimer()` to guarantee
that only one
- * cleanup timer is registered per key.
- */
- private def updateCleanupTimer(currentProcessingTime: JLong,
currentCleanupTime: JLong): Unit = {
- if (currentCleanupTime != null) {
- timerService.deleteProcessingTimeTimer(currentCleanupTime)
+ private def updateCleanUpTimer(
+ currentProcessingTime: JLong,
+ currentCleanUpTime: Option[JLong]): Unit = {
+
+ val cleanUpTime: JLong = currentCleanUpTime.getOrElse(-1L)
+ if (cleanUpTime != -1) {
Review comment:
I agree about the cleanness of the first solution.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
