samza-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Roger Hoover <>
Subject Re: Re-processing a la Kappa/Liquid
Date Fri, 20 Feb 2015 17:30:05 GMT

Thank you for this really great response.  I don't need this right now but
wanted to understand the limitations.  If Kafka could guarantee to keep the
last N hours dirty, that would provide foundation to build on.



On Fri, Feb 20, 2015 at 8:16 AM, Chris Riccomini <>

> Hey Roger,
> I believe your description is correct. Kafka has a "dirty ratio" concept
> for its log-compacted topics. Once the dirty (unclean) portion of the log
> passes the "dirty ratio" (e.g. 50% of the log hasn't been clean, based on
> bytes on disk), the log cleaner kicks in. Once this happens, you can't
> restore to prior intermediate states.
> There has been some verbal discussion about changing the concept of a
> "dirty ratio" to be time-based. Something like, "always keep the last N
> hours dirty". This would allow you to restore to any state within the last
> N hours (but not prior, as you point out). The nice thing about this
> approach is that it gives a very clear guarantee about how far back you can
> go.
> If you wanted to restore to a point in time prior to the last cleaning,
> then you'd have to keep some sort of snapshot, as you suggest. There might
> be other tricks you could play, but nothing is coming to mind.
> Where this gets a little funky for Samza is in a case where a state store's
> changelog is ahead of the last checkpointed input offsets. In an ideal
> world, you'd like to restore up to the offset in the changelog that matched
> the point where the input offsets were committed. For example, if you have
> a message order like so:
>   cl1, cl2, cl3, cp1, cl4, cl5, <fail>
> When you start up, you'd like to read the changelog up to cl3, and start
> the consumers at offset cp1. If your processing is idempotent and
> deterministic, then cl4 and cl5 will get over-written, and you'll have
> consistent state moving forward. In practice, we don't do this. If a log
> cleaner were to kick in, though, and delete cl3 (say it has the same key as
> cl4) before you restart, then you *can't* restore to this point.
> In practice we haven't really worried about this because, as it is, we
> simply restore up to cl5 on restart. There is a lot of complexity with not
> restoring up to cl5:
> 1. What if someone is consuming from the changelog topic? You can't simply
> truncate the changelog on restart.
> 2. What if the log cleaner has deleted cl3, or prior CL messages (as you
> suggested)?
> 3. What if your processing is non-deterministic or non-idempotent?
> The most ideal scenario is simply to have Kafka support transactions. This
> will allow us to atomically commit cp1 and all output (both changelog, and
> regular). This would at least guarantee state-consistent restore for the
> *current* checkpoint. Going backwards in time would have to be addressed by
> the time-based dirty policy that I described above.
> In the absence of this, we *could* make the checkpoints contain changelog
> offsets, and only restore up to that point on restart (cp3 in the example).
> We haven't done this, but I think it's do-able. This will at least make
> things restore properly for the *current* checkpoint, when the processing
> is *deterministic* and *idempotent*. You can open a JIRA for this if you
> need it.
> Cheers,
> Chris
> On Thu, Feb 19, 2015 at 6:55 PM, Roger Hoover <>
> wrote:
> > Chris + Samza Devs,
> >
> > I was wondering whether Samza could support re-processing as described by
> > the Kappa architecture or Liquid (
> >
> >
> > It seems that a changelog is not sufficient to be able to restore state
> > backward in time.  Kafka compaction will guarantee that local state can
> be
> > restored from where it left off but I don't see how it can restore past
> > state.
> >
> > Imagine the case where a stream job has a lot of state in it's local
> store
> > but it has not updated any keys in a long time.
> >
> > Time t1: All of the data would be in the tail of the Kafka log (past the
> > cleaner point).
> > Time t2:  The job updates some keys.   Now we're in a state where the
> next
> > compaction will blow away the old values for those keys.
> > Time t3:  Compaction occurs and old values are discarded.
> >
> > Say we want to launch a re-processing job that would begin from t1.  If
> we
> > launch that job before t3, it will correctly restore it's state.
> However,
> > if we launch the job after t3, it will be missing old values, right?
> >
> > Unless I'm misunderstanding something, the only way around this is to
> keep
> > snapshots in addition to the changelog.  Has there been any discussion of
> > providing an option in Samza of taking RocksDB snapshots and persisting
> > them to an object store or HDFS?
> >
> > Thanks,
> >
> > Roger
> >

  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message