Hi Adam and John, thank you for your effort!
We are implementing full idem-potency in our projects so that's nothing to
worry about.
As to what John said - we only have one partition, I personally assured
that.
So as i wrote in section 2. of my first message in this conversation - my
stream should have processed the "asd" message again because it is not
committed yet.
That's why i suspect it has something to do with the stream's cache; maybe
something like:
1. "asd" got processed and restored in cache
2. "{}" got processed and cached too.
3. commit interval makes the stream commit the offset of "{}"
B.t.w
If you want to run my application you should:
1. open it in some java editor as maven project
2. run it as a normal java application
3. setup kafka server & zookeeper on localhost
4. then you can send the above messages via cli
John - even if you send "asd1", "asd2", "asd3" you will see in the logs
that my app takes the latest each time
Of course that's far beyond what i can ask from you guys to do, thanks a
lot for your help.
On Wed, Sep 12, 2018 at 8:14 PM John Roesler <j...@confluent.io> wrote:
> Hi!
>
> As Adam said, if you throw an exception during processing, it should cause
> Streams to shut itself down and *not* commit that message. Therefore, when
> you start up again, it should again attempt to process that same message
> (and shut down again).
>
> Within a single partition, messages are processed in order, so a bad
> message will block the queue, and you should not see subsequent messages
> get processed.
>
> However, if your later message "{}" goes to a different partition than the
> bad message, then there's no relationship between them, and the later,
> good, message might get processed.
>
> Does that help?
> -John
>
> On Wed, Sep 12, 2018 at 8:38 AM Adam Bellemare <adam.bellem...@gmail.com>
> wrote:
>
> > Hi Yui Yoi
> >
> >
> > Keep in mind that Kafka Consumers don't traditionally request only a
> single
> > message at a time, but instead requests them in batches. This allows for
> > much higher throughput, but does result in the scenario of
> "at-least-once"
> > processing. Generally what will happen in this scenario is the following:
> >
> > 1) Client requests the next set of messages from offset (t). For example,
> > assume it gets 10 messages and message 6 is "bad".
> > 2) The client's processor will then process the messages one at a time.
> > Note that the offsets are not committed after the message is processed,
> but
> > only at the end of the batch.
> > 3) The bad message it hit by the processor. At this point you can decide
> to
> > skip the message, throw an exception, etc.
> > 4a) If you decide to skip the message, processing will continue. Once all
> > 10 messages are processed, the new offset (t+10) offset is committed back
> > to Kafka.
> > 4b) If you decide to throw an exception and terminate your app, you will
> > have still processed the messages that came before the bad message.
> Because
> > the offset (t+10) is not committed, the next time you start the app it
> will
> > consume from offset t, and those messages will be processed again. This
> is
> > "at-least-once" processing.
> >
> >
> > Now, if you need exactly-once processing, you have two choices -
> > 1) Use Kafka Streams with exactly-once semantics (though, as I am not
> > familiar with your framework, it may support it as well).
> > 2) Use idempotent practices (ie: it doesn't matter if the same messages
> get
> > processed more than once).
> >
> >
> > Hope this helps -
> >
> > Adam
> >
> >
> > On Wed, Sep 12, 2018 at 7:59 AM, Yui Yoi <shalosh...@gmail.com> wrote:
> >
> > > Hi Adam,
> > > Thanks a lot for the rapid response, it did helped!
> > >
> > > Let me though ask one more simple question: Can I make a stream
> > application
> > > stuck on an invalid message? and not consuming any further messages?
> > >
> > > Thanks again
> > >
> > > On Wed, Sep 12, 2018 at 2:35 PM Adam Bellemare <
> adam.bellem...@gmail.com
> > >
> > > wrote:
> > >
> > > > Hi Yui Yoi
> > > >
> > > > Preface: I am not familiar with the spring framework.
> > > >
> > > > "Earliest" when it comes to consuming from Kafka means, "Start
> reading
> > > from
> > > > the first message in the topic, *if there is no offset stored for
> that
> > > > consumer group*". It sounds like you are expecting it to re-read each
> > > > message whenever a new message comes in. This is not going to happen,
> > as
> > > > there will be a committed offset and "earliest" will no longer be
> used.
> > > If
> > > > you were to use "latest" instead, if a consumer is started that does
> > not
> > > > have a valid offset, it would use the very latest message in the
> topic
> > as
> > > > the starting offset for message consumption.
> > > >
> > > > Now, if you are using the same consumer group each time you run the
> > > > application (which it seems is true, as you have "test-group"
> hardwired
> > > in
> > > > your application.yml), but you do not tear down your local cluster
> and
> > > > clear out its state, you will indeed see the behaviour you describe.
> > > > Remember that Kafka is durable, and maintains the offsets when the
> > > > individual applications go away. So you are probably seeing this:
> > > >
> > > > 1) start application instance 1. It realizes it has no offset when it
> > > tries
> > > > to register as a consumer on the input topic, so it creates a new
> > > consumer
> > > > entry for "earliest" for your consumer group.
> > > > 2) send message "asd"
> > > > 3) application instance 1 receives "asd", processes it, and updates
> the
> > > > offset (offset head = 1)
> > > > 4) Terminate instance 1
> > > > 5) Start application instance 2. It detects correctly that consumer
> > group
> > > > "test-group" is available and reads that offset as its starting
> point.
> > > > 6) send message "{}"
> > > > 7) application instance 2 receives "{}", processes it, and updates
> the
> > > > offset (offset head = 2)
> > > > *NOTE:* App instance 2 NEVER received "asd", nor should it, as it is
> > > > telling the Kafka cluster that it belongs to the same consumer group
> as
> > > > application 1.
> > > >
> > > > Hope this helps,
> > > >
> > > > Adam
> > > >
> > > >
> > > >
> > > >
> > > >
> > > > On Wed, Sep 12, 2018 at 6:57 AM, Yui Yoi <shalosh...@gmail.com>
> wrote:
> > > >
> > > > > TL;DR:
> > > > > my streams application skips uncommitted messages
> > > > >
> > > > > Hello,
> > > > > I'm using streams API via spring framework and experiencing a weird
> > > > > behavior which I would like to get an explanation to:
> > > > > First of all: The attached zip is my test project, I used kafka cli
> > to
> > > > run
> > > > > a localhost broker and zookeeper
> > > > >
> > > > > what is happening is as follows:
> > > > > 1. I send an invalid message, such as "asd", and my consumer has a
> > lag
> > > > and
> > > > > error message as expected
> > > > > 2. I send a valid message such as "{}", but instead of rereading
> the
> > > > first
> > > > > message as expected from an "earliest" configured application - my
> > > > > application reads the latest message, commits it and ignoring the
> one
> > > in
> > > > > error, thus i have no lag!
> > > > > 3. When I'm running my application when there are uncommitted
> > messages
> > > -
> > > > > my application reads the FIRST not committed message, as if it IS
> an
> > > > > "earliest" configured application!
> > > > >
> > > > > In your documentation you assure "at least once" behavior, but
> > > according
> > > > > to section 2. it happens so my application does not receive those
> > > > messages
> > > > > not even once (as i said, those messages are uncommitted)
> > > > >
> > > > > My guess is that it has something to do with the stream's cache...
> I
> > > > would
> > > > > very like to have an explanation or even a solution
> > > > >
> > > > > I'm turning to you as a last resort, after long weeks of research
> and
> > > > > experiments
> > > > >
> > > > > Thanks alot
> > > > >
> > > >
> > >
> >
>
