[
https://issues.apache.org/jira/browse/GEARPUMP-32?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15395347#comment-15395347
]
ASF GitHub Bot commented on GEARPUMP-32:
----------------------------------------
Github user huafengw commented on a diff in the pull request:
https://github.com/apache/incubator-gearpump/pull/67#discussion_r72412383
--- Diff:
streaming/src/main/scala/org/apache/gearpump/streaming/source/DataSourceTask.scala
---
@@ -46,23 +46,39 @@ class DataSourceTask private[source](context:
TaskContext, conf: UserConfig, sou
}
private val batchSize =
conf.getInt(DataSourceConfig.SOURCE_READ_BATCH_SIZE).getOrElse(1000)
private var startTime = 0L
+ private var lastWatermark: TimeStamp = startTime
override def onStart(newStartTime: StartTime): Unit = {
startTime = newStartTime.startTime
LOG.info(s"opening data source at $startTime")
source.open(context, startTime)
- self ! Message("start", System.currentTimeMillis())
+ self ! Message("start")
}
override def onNext(message: Message): Unit = {
0.until(batchSize).foreach { _ =>
- Option(source.read()).foreach(context.output)
+ Option(source.read()).foreach { msg =>
+ context.output(msg)
+ }
}
- self ! Message("continue", System.currentTimeMillis())
+
+ maybeUpdateWatermark()
+ self ! Message("continue")
}
override def onStop(): Unit = {
LOG.info("closing data source...")
source.close()
}
+
+ private def maybeUpdateWatermark(): Unit = {
+ val curWatermark = source.getWatermark
+ if (curWatermark > lastWatermark) {
+ lastWatermark = curWatermark
+ self ! UpstreamMinClock(curWatermark)
--- End diff --
Once the watermark move forward, the Actor will send an UpstreamMinClock to
self. If DataSource's watermark updates frequently, will it be a problem? I
mean the UpstreamMinClock flood.
> Minimum clock of source Tasks maybe inaccurate
> ----------------------------------------------
>
> Key: GEARPUMP-32
> URL: https://issues.apache.org/jira/browse/GEARPUMP-32
> Project: Apache Gearpump
> Issue Type: Bug
> Components: streaming
> Affects Versions: 0.8.0
> Reporter: Manu Zhang
> Assignee: Manu Zhang
> Fix For: 0.8.1
>
>
> Moved from [https://github.com/gearpump/gearpump/issues/1835] and reported by
> [Zhu Yueqian|https://github.com/yueqianzhu]
> {quote}
> Source tasks have not any upstreamClocks. So, startClock is the minimum of
> pending clocks when recover happen.
> eg below:
> source task1: timeStamp:15,not ACK, minClockValue maybe is 15(<= 15).
> source task2: timeStamp:10,ACKed, minClockValue maybe is Long.MaxValue
> when recover happen,startClock maybe is 15. where is the data between 10 to
> 15 at task2?
> {quote}
> More context on this issue:
> In Gearpump, we maintain a global minimum clock tracked from a message's
> timestamp across all tasks. It means messages with timestamp before this
> clock have all been processed. An application will restart from this value on
> failure, and thus at-least-once message delivery could be guaranteed.
> The global minimum clock is the lower bound of all the Tasks' minimum clocks.
> For a task, the minimum clock is the lower of
> # upstream minimum clock
> # a. the minimum timestamp of unacked messages
> b. Long.MaxValue if all messages have been acked.
>
> Note that 2.b allows the global minimum clock to progress and it is almost
> safe since the clock is also bounded by the upstream minimum clock. I said
> "almost safe" because a source task has no upstream but we assume the
> upstream minimum clock is Long.MaxValue. Thus, the scenario described by Zhu
> Yueqian could happen and breaks at-least-once guarantee.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
