gearpump-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (GEARPUMP-32) Minimum clock of source Tasks maybe inaccurate
Date Wed, 27 Jul 2016 09:59:21 GMT

    [ 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)

Mime
View raw message