gearpump-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <>
Subject [jira] [Commented] (GEARPUMP-32) Minimum clock of source Tasks maybe inaccurate
Date Thu, 28 Jul 2016 10:28:20 GMT


ASF GitHub Bot commented on GEARPUMP-32:

Github user manuzhang commented on a diff in the pull request:
    --- Diff: external/kafka/src/main/scala/org/apache/gearpump/streaming/kafka/lib/source/AbstractKafkaSource.scala
    @@ -74,11 +74,10 @@ abstract class AbstractKafkaSource(
       private lazy val kafkaClient: KafkaClient = kafkaClientFactory.getKafkaClient(config)
       private lazy val fetchThread: FetchThread = fetchThreadFactory.getFetchThread(config,
       private lazy val messageDecoder = config.getConfiguredInstance(
    -    KafkaConfig.MESSAGE_DECODER_CLASS_CONFIG, classOf[MessageDecoder])
    -  private lazy val timestampFilter = config.getConfiguredInstance(
    --- End diff --
    previously filter is used to filter out old messages (e.g. timestamp < startTime) and
carried out in KafkaSource implicitly. Now I think it should be defined explicitly in the
following operations by users like `withAllowedLateness` in Beam API although that is not
available in Gearpump API yet. 

> Minimum clock of source Tasks maybe inaccurate
> ----------------------------------------------
>                 Key: GEARPUMP-32
>                 URL:
>             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 [] and reported by [Zhu Yueqian|]
> {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

View raw message