Re: Custom AuthenticateCallbackHandler

[Roy van der Valk]

No I want to check plain user credentials for authentication against an 
external system. Where best to plug in this check. I thought the new 
AuthenticateCallbackHandler would be convenient, but I am very much in doubt if 
this is the right way. It is also not clear for me from the docs how to 
implement this.

Please see my stackoverflow question for a minimal test setup which I try to 
get working.

https://stackoverflow.com/questions/50412589/kafka-custom-authenticatecallbackhandler

Any help/pointers/examples is much appreciated!

> Op 1 jun. 2018 om 21:05 heeft Martin Gainty  het 
> volgende geschreven:
> 
> are you referring to onCompleteCallback from
> 
> https://github.com/apache/kafka/commit/837f31dd1850b179918f83338b4b4487486b2c58
> 
> [https://avatars2.githubusercontent.com/u/14958864?s=200&v=4]
> 
> KAFKA-6927; Chunked down-conversion to prevent out of memory errors o… · 
> apache/kafka@837f31d
> github.com
> …n broker [KIP-283] (#4871) Implementation for lazy down-conversion in a 
> chunked manner for efficient memory usage during down-conversion. This pull 
> request is mainly to get initial feedback on th...
> 
> 
> 
> 
> ?
> 
> Martin
> _
> 
> 
> 
> 
> From: Roy van der Valk 
> Sent: Friday, June 1, 2018 2:28 PM
> To: users@kafka.apache.org
> Subject: Re: Custom AuthenticateCallbackHandler
> 
> Thank you Manikumar!
> 
>> On Fri, Jun 1, 2018 at 7:30 PM, Manikumar  wrote:
>> 
>> This feature will be part upcoming Kafka 2.0.0 release.
>> 
>> Doc PR is here : https://github.com/apache/kafka/pull/4890
> [https://avatars2.githubusercontent.com/u/13164074?s=400&v=4]
> 
> KAFKA-6800: Update SASL/PLAIN and SCRAM docs to use KIP-86 callbacks by 
> rajinisivaram · Pull Request #4890 · 
> apache/kafka
> github.com
> Committer Checklist (excluded from commit message) Verify design and 
> implementation Verify test coverage and CI build status Verify documentation 
> (including upgrade notes)
> 
> 
> 
>> 
>> configs here:
>> https://github.com/apache/kafka/blob/trunk/clients/src/
>> main/java/org/apache/kafka/common/config/SaslConfigs.java#L57
>> 
>> On Fri, Jun 1, 2018 at 10:51 PM, Roy van der Valk <
>> roy.van.der.v...@gmail.com> wrote:
>> 
>>> Dear Kafka community,
>>> 
>>> Can somebody help me setting up a custom AuthenticateCallbackHandler as
>>> described in KIP-86 recently added by Rajini Sivaram or point me to good
>>> documentation?
>>> 
>>> I described my question in more detail on Stackoverflow:
>>> https://stackoverflow.com/questions/50412589/kafka-custom-
>>> authenticatecallbackhandler
>>> 
>>> All help is greatly appreciated!
>>> 
>>> Roy
>>> 
>> 

Re: Kafka consumer loop exception handling

[M. Manna]

You should set the reset to latest, commit offsets manually using a
rebalance listener. In this way upon seek() you should get all data right.

Also when you say “Uncommitted” offset, that means you haven’t really
processed them. So you should determine such failure, manually control
offset commits, and protect from duplication.



Regards,


On Fri, 1 Jun 2018, 18:32 pradeep s,  wrote:

> Than you. In my case i am receiving messages , doing a small transformation
> and sending to a output topic .
> If i am running 4 consumers against 4 partitions and one of the consumer
> dies , will there be duplicate messages sent in this case
> Since when the new consumer comes up , it will again process from the
> uncommitted offset .
> So do i need transaction semantics in this scenario.
>
>
> On Fri, Jun 1, 2018 at 4:56 AM, M. Manna  wrote:
>
> > This is actually quite nicely explained by Jason Gustafson on this
> article
> > -
> > https://www.confluent.io/blog/tutorial-getting-started-with-
> > the-new-apache-kafka-0-9-consumer-client/
> >
> > It's technically up to the application on how to determine whether
> message
> > is fully received. If you have database txn involved, I would say that
> > CommitFailedException should revert all changes you have done. Because
> you
> > couldn't commit the offset successfully, you haven't "Really" consumed
> any
> > message.
> >
> > Tailoring your code a little bit:
> >
> > @Override
> > public void run() {
> > try {
> > do {
> > processRecords(kafkaConsumer.poll(kafkaConfig.
> > getPollTimeoutMs()));
> > kafkaConsumer.commitSync();
> > } while (!isConsumerLoopClosed.get());
> > } catch (WakeupException wakeupException) {
> > //do nothing if wakeupException is from shutdown hook
> > if (!isConsumerLoopClosed.get()) {
> > handleConsumerLoopException(wakeupException);
> > }
> > } catch (RuntimeException ex) { // RuntimeException could also happen
> > for other reasons here
> > if (ex instanceof CommitFailedException) {
> > // revert db txn etc. to avoid false positives
> > } else if (ex instanceof KafkaException) {
> > // do something else.
> > } else {
> >// alternatively, do this
> > }
> > handleConsumerLoopException(ex);
> > } finally {
> > kafkaConsumer.close();
> > }
> >
> > }
> >
> > One thing to remember is that when you are sending data, as of 1.0.0 API
> > you can have a "Txn-like" finer control to determine when you have
> > successfully committed a transaction. You can check beginTransaction(),
> > commitTransaction(), abortTransaction() methods to see how they can be
> > utilised to have even finer control over your message delivery.
> >
> > Regards,
> >
> >
> > On 1 June 2018 at 05:54, pradeep s  wrote:
> >
> > > Hi,
> > > I am running a poll loop for kafka consumer and the app is deployed in
> > > kubernetes.I am using manual commits.Have couple of questions on
> > exception
> > > handling in the poll loop
> > >
> > > 1) Do i need to handle consumer rebalance scenario(when any of the
> > consumer
> > > pod dies) by adding a listener or will the commits be taken care after
> > > rebalance .
> > >
> > > 2) Do i need to handle CommitFailedException specifically
> > >
> > > Consume loop code below
> > >
> > >
> > > @Override
> > > public void run() {
> > > try {
> > > do {
> > > processRecords(kafkaConsumer.poll(kafkaConfig.
> > > getPollTimeoutMs()));
> > > kafkaConsumer.commitSync();
> > > } while (!isConsumerLoopClosed.get());
> > > } catch (WakeupException wakeupException) {
> > > //do nothing if wakeupException is from shutdown hook
> > > if (!isConsumerLoopClosed.get()) {
> > > handleConsumerLoopException(wakeupException);
> > > }
> > > } catch (RuntimeException ex) {
> > > handleConsumerLoopException(ex);
> > > } finally {
> > > kafkaConsumer.close();
> > > }
> > >
> > >
> > > }
> > >
> > > Thanks
> > > Pradeep
> > >
> >
>

Re: Custom AuthenticateCallbackHandler

[Martin Gainty]

are you referring to onCompleteCallback from

https://github.com/apache/kafka/commit/837f31dd1850b179918f83338b4b4487486b2c58

[https://avatars2.githubusercontent.com/u/14958864?s=200&v=4]

KAFKA-6927; Chunked down-conversion to prevent out of memory errors o… · 
apache/kafka@837f31d
github.com
…n broker [KIP-283] (#4871) Implementation for lazy down-conversion in a 
chunked manner for efficient memory usage during down-conversion. This pull 
request is mainly to get initial feedback on th...




?

Martin
_




From: Roy van der Valk 
Sent: Friday, June 1, 2018 2:28 PM
To: users@kafka.apache.org
Subject: Re: Custom AuthenticateCallbackHandler

Thank you Manikumar!

On Fri, Jun 1, 2018 at 7:30 PM, Manikumar  wrote:

> This feature will be part upcoming Kafka 2.0.0 release.
>
> Doc PR is here : https://github.com/apache/kafka/pull/4890
[https://avatars2.githubusercontent.com/u/13164074?s=400&v=4]

KAFKA-6800: Update SASL/PLAIN and SCRAM docs to use KIP-86 callbacks by 
rajinisivaram · Pull Request #4890 · 
apache/kafka
github.com
Committer Checklist (excluded from commit message) Verify design and 
implementation Verify test coverage and CI build status Verify documentation 
(including upgrade notes)



>
> configs here:
> https://github.com/apache/kafka/blob/trunk/clients/src/
> main/java/org/apache/kafka/common/config/SaslConfigs.java#L57
>
> On Fri, Jun 1, 2018 at 10:51 PM, Roy van der Valk <
> roy.van.der.v...@gmail.com> wrote:
>
> > Dear Kafka community,
> >
> > Can somebody help me setting up a custom AuthenticateCallbackHandler as
> > described in KIP-86 recently added by Rajini Sivaram or point me to good
> > documentation?
> >
> > I described my question in more detail on Stackoverflow:
> > https://stackoverflow.com/questions/50412589/kafka-custom-
> > authenticatecallbackhandler
> >
> > All help is greatly appreciated!
> >
> > Roy
> >
>

Re: Is there expiration for committed Offset in the partition

[Matthias J. Sax]

It is an know issue. You can increase the retention time for stored
offsets via configs thought.

There is already an open PR to fix this issue:
https://issues.apache.org/jira/browse/KAFKA-4682


-Matthias

On 6/1/18 2:00 AM, Dinesh Subramanian wrote:
> Hi M. Manna,
> 
> Am planning to store outside kafka.. will this be a solution ?
> 
> 
> *Thanks & Regards,*
> 
> *Dinesh S*
> 
> On Fri, Jun 1, 2018 at 2:05 PM, M. Manna  wrote:
> 
>> This can happen for two reasons:
>>
>> 1) Your offsets are expired and removed. So your consumers don't know where
>> to start from - earliest means "Start from the beginning"
>> 2) You are actually starting as part of a totally new consumer group - in
>> which case it's as designed too - start from the beginning.
>>
>> I would check your offset retention policy by size/time - and tune that if
>> necessary.
>>
>> On 1 June 2018 at 09:03, Dinesh Subramanian 
>> wrote:
>>
>>> Hi,
>>>
>>> Facing duplication in below scenario
>>>
>>> Last commit is happened in 3 days back in the consumer, after that no
>>> messages produced in the topic. so no commits..
>>> so after 3 days am stopping and restarting the consumer..  this time i
>>> faced duplication issue in the consumer as i have this consumer
>>> property "*auto.offset.reset
>>> = earliest*", It is consumed again from the beginning.. any helps will be
>>> appreciated.
>>>
>>> *Thanks & Regards,*
>>>
>>> *Dinesh S*
>>>
>>
> 



signature.asc
Description: OpenPGP digital signature

Re: Custom AuthenticateCallbackHandler

[Roy van der Valk]

Thank you Manikumar!

On Fri, Jun 1, 2018 at 7:30 PM, Manikumar  wrote:

> This feature will be part upcoming Kafka 2.0.0 release.
>
> Doc PR is here : https://github.com/apache/kafka/pull/4890
>
> configs here:
> https://github.com/apache/kafka/blob/trunk/clients/src/
> main/java/org/apache/kafka/common/config/SaslConfigs.java#L57
>
> On Fri, Jun 1, 2018 at 10:51 PM, Roy van der Valk <
> roy.van.der.v...@gmail.com> wrote:
>
> > Dear Kafka community,
> >
> > Can somebody help me setting up a custom AuthenticateCallbackHandler as
> > described in KIP-86 recently added by Rajini Sivaram or point me to good
> > documentation?
> >
> > I described my question in more detail on Stackoverflow:
> > https://stackoverflow.com/questions/50412589/kafka-custom-
> > authenticatecallbackhandler
> >
> > All help is greatly appreciated!
> >
> > Roy
> >
>

Re: How to gracefully stop Kafka

[Raghav]

Thanks guys, I will try this and update to see if that worked.

On Fri, Jun 1, 2018 at 1:42 AM, M. Manna  wrote:

> Regarding graceful shutdown - I have got a response from Jan in the past -
> I am simply quoting that below:
>
> "A gracefully shutdown means the broker is only shutting down when it is
> not the leader of any partition.
> Therefore you should not be able to gracefully shut down your entire
> cluster."
>
> That said, you should allow some flexibility in your startup. I do my
> testbed (3-node) startup always the following way - and it works nicely
>
> 1) Start each zookeeper node - allow 5 seconds between each startup.
> 2) When all ZKs are up - wait for another 10 seconds
> 3) Start all brokers - allow 5 seconds between each startup
>
> Provided that your index files aren't corrupted - it should always start up
> normally.
>
> Regards,
>
>
>
>
> On 1 June 2018 at 07:37, Pena Quijada Alexander 
> wrote:
>
> > Hi,
> >
> > From my point of view, if you don't have any tool that help you in the
> > management of your broker services, in other to do a rolling restart
> > manually, you should shut down one broker at a time.
> >
> > In this way, you leave time to the broker controller service to balance
> > the active replicas into the healthy nodes.
> >
> > The same procedure when you start up your nodes.
> >
> > Regards!
> >
> > Alex
> >
> > Inviato da BlueMail Il giorno 1 giu
> > 2018, alle ore 07:31, Raghav  > raghavas...@gmail.com>> ha scritto:
> >
> > Hi
> >
> > We have a 3 Kafka brokers setup on 0.10.2.1. We have a requirement in our
> > company environment that we have to first stop our 3 Kafka Broker setup,
> > then do some operations stuff that takes about 1 hours, and then bring up
> > Kafka (version 1.1) brokers again.
> >
> > In order to achieve this, we issue:
> >
> > 1. Run *bin/kafka-server-stop.sh > //kafka-server-stop.sh>* at the same time on all three brokers.
> > 2. Do operations on our environment for about 1 hour.
> > 3. Run bin/kafka-server.-start.sh at
> > the same time on all three brokers.
> >
> > Upon start, we observe that leadership for lot of partition is messed up.
> > The leadership shows up as -1 for lot of partitions. And ISR has no
> > servers. Because of this our Kafka cluster is unusable, and even restart
> of
> > brokers doesn't help.
> >
> > 1. Could it be because we are not doing rolling stop ?
> > 2. What's the best way to do rollling stop ?
> >
> > Please advise.
> > Thanks.
> >
> > R
> >
> > 
> >
> > --
> > The information transmitted is intended for the person or entity to which
> > it is addressed and may contain confidential and/or privileged material.
> > Any review, retransmission, dissemination or other use of, or taking of
> any
> > action in reliance upon, this information by persons or entities other
> than
> > the intended recipient is prohibited. If you received this in error,
> please
> > contact the sender and delete the material from any computer.
> >
>



-- 
Raghav

Re: Kafka consumer loop exception handling

[pradeep s]

Than you. In my case i am receiving messages , doing a small transformation
and sending to a output topic .
If i am running 4 consumers against 4 partitions and one of the consumer
dies , will there be duplicate messages sent in this case
Since when the new consumer comes up , it will again process from the
uncommitted offset .
So do i need transaction semantics in this scenario.


On Fri, Jun 1, 2018 at 4:56 AM, M. Manna  wrote:

> This is actually quite nicely explained by Jason Gustafson on this article
> -
> https://www.confluent.io/blog/tutorial-getting-started-with-
> the-new-apache-kafka-0-9-consumer-client/
>
> It's technically up to the application on how to determine whether message
> is fully received. If you have database txn involved, I would say that
> CommitFailedException should revert all changes you have done. Because you
> couldn't commit the offset successfully, you haven't "Really" consumed any
> message.
>
> Tailoring your code a little bit:
>
> @Override
> public void run() {
> try {
> do {
> processRecords(kafkaConsumer.poll(kafkaConfig.
> getPollTimeoutMs()));
> kafkaConsumer.commitSync();
> } while (!isConsumerLoopClosed.get());
> } catch (WakeupException wakeupException) {
> //do nothing if wakeupException is from shutdown hook
> if (!isConsumerLoopClosed.get()) {
> handleConsumerLoopException(wakeupException);
> }
> } catch (RuntimeException ex) { // RuntimeException could also happen
> for other reasons here
> if (ex instanceof CommitFailedException) {
> // revert db txn etc. to avoid false positives
> } else if (ex instanceof KafkaException) {
> // do something else.
> } else {
>// alternatively, do this
> }
> handleConsumerLoopException(ex);
> } finally {
> kafkaConsumer.close();
> }
>
> }
>
> One thing to remember is that when you are sending data, as of 1.0.0 API
> you can have a "Txn-like" finer control to determine when you have
> successfully committed a transaction. You can check beginTransaction(),
> commitTransaction(), abortTransaction() methods to see how they can be
> utilised to have even finer control over your message delivery.
>
> Regards,
>
>
> On 1 June 2018 at 05:54, pradeep s  wrote:
>
> > Hi,
> > I am running a poll loop for kafka consumer and the app is deployed in
> > kubernetes.I am using manual commits.Have couple of questions on
> exception
> > handling in the poll loop
> >
> > 1) Do i need to handle consumer rebalance scenario(when any of the
> consumer
> > pod dies) by adding a listener or will the commits be taken care after
> > rebalance .
> >
> > 2) Do i need to handle CommitFailedException specifically
> >
> > Consume loop code below
> >
> >
> > @Override
> > public void run() {
> > try {
> > do {
> > processRecords(kafkaConsumer.poll(kafkaConfig.
> > getPollTimeoutMs()));
> > kafkaConsumer.commitSync();
> > } while (!isConsumerLoopClosed.get());
> > } catch (WakeupException wakeupException) {
> > //do nothing if wakeupException is from shutdown hook
> > if (!isConsumerLoopClosed.get()) {
> > handleConsumerLoopException(wakeupException);
> > }
> > } catch (RuntimeException ex) {
> > handleConsumerLoopException(ex);
> > } finally {
> > kafkaConsumer.close();
> > }
> >
> >
> > }
> >
> > Thanks
> > Pradeep
> >
>

Re: Custom AuthenticateCallbackHandler

[Manikumar]

This feature will be part upcoming Kafka 2.0.0 release.

Doc PR is here : https://github.com/apache/kafka/pull/4890

configs here:
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/common/config/SaslConfigs.java#L57

On Fri, Jun 1, 2018 at 10:51 PM, Roy van der Valk <
roy.van.der.v...@gmail.com> wrote:

> Dear Kafka community,
>
> Can somebody help me setting up a custom AuthenticateCallbackHandler as
> described in KIP-86 recently added by Rajini Sivaram or point me to good
> documentation?
>
> I described my question in more detail on Stackoverflow:
> https://stackoverflow.com/questions/50412589/kafka-custom-
> authenticatecallbackhandler
>
> All help is greatly appreciated!
>
> Roy
>

Custom AuthenticateCallbackHandler

[Roy van der Valk]

Dear Kafka community,

Can somebody help me setting up a custom AuthenticateCallbackHandler as
described in KIP-86 recently added by Rajini Sivaram or point me to good
documentation?

I described my question in more detail on Stackoverflow:
https://stackoverflow.com/questions/50412589/kafka-custom-authenticatecallbackhandler

All help is greatly appreciated!

Roy

Re: Frequent consumer rebalances, auto commit failures

[Ken Chen]

1. Any detail logs ?
2. How do you process the records after you polled the records?
3. How much time does it take for every round of poll ? 

Thanks !

--
Sent from my iPhone

On May 28, 2018, at 10:44 PM, Shantanu Deshmukh  wrote:

Can anyone here help me please? I am at my wit's end. I now have
max.poll.records set to just 2. Still I am getting Auto offset commit
failed warning. Log file is getting full because of this warning. Session
timeout is 5 minutes, max.poll.interval.ms is 10 minutes.

On Wed, May 23, 2018 at 12:42 PM Shantanu Deshmukh 
wrote:

> 
> Hello,
> 
> We have a 3 broker Kafka 0.10.0.1 cluster. There we have 3 topics with 10
> partitions each. We have an application which spawns threads as consumers.
> We spawn 5 consumers for each topic. I am observing that consider group
> randomly keeps rebalancing. Then many times we see logs saying "Revoking
> partitions for". This happens almost every 10 minutes. Consumption during
> this time completely stops.
> 
> I have applied this configuration
> max.poll.records 20
> heartbeat.interval.ms 1
> Session.timeout.ms 6000
> 
> Still this did not help. Strange thing is I observed consumer writing logs
> saying "auto commit failed because poll() loop spent too much time
> processing records" even when there was no data in partition to process. We
> have polling interval of 500 ms, specified as argument in poll(). Initially
> I had set same consumer group for all three topics' consumers. Then I
> specified different CGs for different topics' consumers. Even this is not
> helping.
> 
> I am trying to search over the web, checked my code, tried many
> combinations of configuration but still no luck. Please help me.
> 
> Thanks & Regards,
> 
> Shantanu Deshmukh
> 

Kafka Connect restarts while creating nw

[Chintan Patel]

Hello,

I'm using Kafka connect in distributed mode. When I create new connector
through Rest API, It restarts all the connectors.

I don't know It's normal behavior or I have to configure something so It
will not restarts all the connectors when I create new connector.

Thanks.
-- 
CHINTAN PATEL
*SR. SOFTWARE DEVELOPER*

Calle 99 # 7a - 77
Oficina 201 Ed. Advance
(57-1) 605-88-80

Re: StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG

[Guozhang Wang]

You can find the release plan for 2.0 here:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=80448820

Beside, as Matthias mentioned you can still work around it as of now by
hard-coding the string values in your app:


props.put("default.deserialization.exception.handler", ...)


Guozhang


On Fri, Jun 1, 2018 at 4:03 AM, Sumit Baurai 
wrote:

> Thanks Guozhang !!
>
> Any idea when would that be?
>
> *Sumit*
>
> On 31 May 2018 at 16:48, Guozhang Wang  wrote:
>
> > Hello Sumit,
> >
> > We are going to release 2.0 soon which should contain this fix:
> > https://issues.apache.org/jira/browse/KAFKA-6825
> >
> > I'm going to cherry-pick it into 1.1 as well so that for the coming
> bug-fix
> > release 1.1.1 it will also contain this.
> >
> >
> > Guozhang
> >
> > On Thu, May 31, 2018 at 5:48 AM, Sumit Baurai  >
> > wrote:
> >
> > > Hi,
> > >
> > > I am trying to use the default production exception handler. I am
> > managing
> > > all my dependencies using Maven. Following are the co-ordinates that I
> am
> > > using :
> > >
> > > 
> > > org.apache.kafka
> > > kafka-streams
> > > 1.1.0
> > > 
> > > 
> > > org.apache.kafka
> > > kafka-clients
> > > 1.1.0
> > > 
> > >
> > > My Problem is that I am not able to use the property :
> > > StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG
> > > because this has been declared as "private".
> > >
> > > Upon going through the apache commit archives, I can see that this is
> > > resolved in the latest code base on github, but the jars on maven
> > > repositories are still not up to date.
> > >
> > > How can I overcome this limitation and use the latest codebase.
> > >
> > >
> > >
> > >
> > >
> > > *Sumit Baurai*Senior Consultant
> > > Architecture & Implementation
> > >
> > > +31 6 41 90 06 11 <+316%2041%2090%2006%2011>
> > > sumit.bau...@devoteam.com
> > >
> >
> >
> >
> > --
> > -- Guozhang
> >
>



-- 
-- Guozhang

Re: Kafka consumer loop exception handling

[M. Manna]

This is actually quite nicely explained by Jason Gustafson on this article
-
https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/

It's technically up to the application on how to determine whether message
is fully received. If you have database txn involved, I would say that
CommitFailedException should revert all changes you have done. Because you
couldn't commit the offset successfully, you haven't "Really" consumed any
message.

Tailoring your code a little bit:

@Override
public void run() {
try {
do {
processRecords(kafkaConsumer.poll(kafkaConfig.
getPollTimeoutMs()));
kafkaConsumer.commitSync();
} while (!isConsumerLoopClosed.get());
} catch (WakeupException wakeupException) {
//do nothing if wakeupException is from shutdown hook
if (!isConsumerLoopClosed.get()) {
handleConsumerLoopException(wakeupException);
}
} catch (RuntimeException ex) { // RuntimeException could also happen
for other reasons here
if (ex instanceof CommitFailedException) {
// revert db txn etc. to avoid false positives
} else if (ex instanceof KafkaException) {
// do something else.
} else {
   // alternatively, do this
}
handleConsumerLoopException(ex);
} finally {
kafkaConsumer.close();
}

}

One thing to remember is that when you are sending data, as of 1.0.0 API
you can have a "Txn-like" finer control to determine when you have
successfully committed a transaction. You can check beginTransaction(),
commitTransaction(), abortTransaction() methods to see how they can be
utilised to have even finer control over your message delivery.

Regards,


On 1 June 2018 at 05:54, pradeep s  wrote:

> Hi,
> I am running a poll loop for kafka consumer and the app is deployed in
> kubernetes.I am using manual commits.Have couple of questions on exception
> handling in the poll loop
>
> 1) Do i need to handle consumer rebalance scenario(when any of the consumer
> pod dies) by adding a listener or will the commits be taken care after
> rebalance .
>
> 2) Do i need to handle CommitFailedException specifically
>
> Consume loop code below
>
>
> @Override
> public void run() {
> try {
> do {
> processRecords(kafkaConsumer.poll(kafkaConfig.
> getPollTimeoutMs()));
> kafkaConsumer.commitSync();
> } while (!isConsumerLoopClosed.get());
> } catch (WakeupException wakeupException) {
> //do nothing if wakeupException is from shutdown hook
> if (!isConsumerLoopClosed.get()) {
> handleConsumerLoopException(wakeupException);
> }
> } catch (RuntimeException ex) {
> handleConsumerLoopException(ex);
> } finally {
> kafkaConsumer.close();
> }
>
>
> }
>
> Thanks
> Pradeep
>

kafka doc question/clarification regarding disk peformance

[Michael Howard]

I'm a newbie reading kafka documentation

https://kafka.apache.org/documentation/#persistence says:
> the performance of linear writes on a JBOD configuration with six 7200rpm
SATA RAID-5 array is ...

I am confused.
I don't know how the configuration can be both "JBOD" and "RAID-5 array" at
the same time.

> ... about 600MB/sec ...

I strongly suspect that this is intended to be a JBOD configuration and
that the reference "RAID-5 array" should be removed from the sentence.

Pls clarify ... both for me and in the doc :)

Michael

Re: StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG

[Sumit Baurai]

Thanks Guozhang !!

Any idea when would that be?

*Sumit*

On 31 May 2018 at 16:48, Guozhang Wang  wrote:

> Hello Sumit,
>
> We are going to release 2.0 soon which should contain this fix:
> https://issues.apache.org/jira/browse/KAFKA-6825
>
> I'm going to cherry-pick it into 1.1 as well so that for the coming bug-fix
> release 1.1.1 it will also contain this.
>
>
> Guozhang
>
> On Thu, May 31, 2018 at 5:48 AM, Sumit Baurai 
> wrote:
>
> > Hi,
> >
> > I am trying to use the default production exception handler. I am
> managing
> > all my dependencies using Maven. Following are the co-ordinates that I am
> > using :
> >
> > 
> > org.apache.kafka
> > kafka-streams
> > 1.1.0
> > 
> > 
> > org.apache.kafka
> > kafka-clients
> > 1.1.0
> > 
> >
> > My Problem is that I am not able to use the property :
> > StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG
> > because this has been declared as "private".
> >
> > Upon going through the apache commit archives, I can see that this is
> > resolved in the latest code base on github, but the jars on maven
> > repositories are still not up to date.
> >
> > How can I overcome this limitation and use the latest codebase.
> >
> >
> >
> >
> >
> > *Sumit Baurai*Senior Consultant
> > Architecture & Implementation
> >
> > +31 6 41 90 06 11 <+316%2041%2090%2006%2011>
> > sumit.bau...@devoteam.com
> >
>
>
>
> --
> -- Guozhang
>

Re: Is there expiration for committed Offset in the partition

[Dinesh Subramanian]

Hi M. Manna,

Am planning to store outside kafka.. will this be a solution ?


*Thanks & Regards,*

*Dinesh S*

On Fri, Jun 1, 2018 at 2:05 PM, M. Manna  wrote:

> This can happen for two reasons:
>
> 1) Your offsets are expired and removed. So your consumers don't know where
> to start from - earliest means "Start from the beginning"
> 2) You are actually starting as part of a totally new consumer group - in
> which case it's as designed too - start from the beginning.
>
> I would check your offset retention policy by size/time - and tune that if
> necessary.
>
> On 1 June 2018 at 09:03, Dinesh Subramanian 
> wrote:
>
> > Hi,
> >
> > Facing duplication in below scenario
> >
> > Last commit is happened in 3 days back in the consumer, after that no
> > messages produced in the topic. so no commits..
> > so after 3 days am stopping and restarting the consumer..  this time i
> > faced duplication issue in the consumer as i have this consumer
> > property "*auto.offset.reset
> > = earliest*", It is consumed again from the beginning.. any helps will be
> > appreciated.
> >
> > *Thanks & Regards,*
> >
> > *Dinesh S*
> >
>

Re: How to gracefully stop Kafka

[M. Manna]

Regarding graceful shutdown - I have got a response from Jan in the past -
I am simply quoting that below:

"A gracefully shutdown means the broker is only shutting down when it is
not the leader of any partition.
Therefore you should not be able to gracefully shut down your entire
cluster."

That said, you should allow some flexibility in your startup. I do my
testbed (3-node) startup always the following way - and it works nicely

1) Start each zookeeper node - allow 5 seconds between each startup.
2) When all ZKs are up - wait for another 10 seconds
3) Start all brokers - allow 5 seconds between each startup

Provided that your index files aren't corrupted - it should always start up
normally.

Regards,




On 1 June 2018 at 07:37, Pena Quijada Alexander 
wrote:

> Hi,
>
> From my point of view, if you don't have any tool that help you in the
> management of your broker services, in other to do a rolling restart
> manually, you should shut down one broker at a time.
>
> In this way, you leave time to the broker controller service to balance
> the active replicas into the healthy nodes.
>
> The same procedure when you start up your nodes.
>
> Regards!
>
> Alex
>
> Inviato da BlueMail Il giorno 1 giu
> 2018, alle ore 07:31, Raghav  raghavas...@gmail.com>> ha scritto:
>
> Hi
>
> We have a 3 Kafka brokers setup on 0.10.2.1. We have a requirement in our
> company environment that we have to first stop our 3 Kafka Broker setup,
> then do some operations stuff that takes about 1 hours, and then bring up
> Kafka (version 1.1) brokers again.
>
> In order to achieve this, we issue:
>
> 1. Run *bin/kafka-server-stop.sh //kafka-server-stop.sh>* at the same time on all three brokers.
> 2. Do operations on our environment for about 1 hour.
> 3. Run bin/kafka-server.-start.sh at
> the same time on all three brokers.
>
> Upon start, we observe that leadership for lot of partition is messed up.
> The leadership shows up as -1 for lot of partitions. And ISR has no
> servers. Because of this our Kafka cluster is unusable, and even restart of
> brokers doesn't help.
>
> 1. Could it be because we are not doing rolling stop ?
> 2. What's the best way to do rollling stop ?
>
> Please advise.
> Thanks.
>
> R
>
> 
>
> --
> The information transmitted is intended for the person or entity to which
> it is addressed and may contain confidential and/or privileged material.
> Any review, retransmission, dissemination or other use of, or taking of any
> action in reliance upon, this information by persons or entities other than
> the intended recipient is prohibited. If you received this in error, please
> contact the sender and delete the material from any computer.
>

Re: Is there expiration for committed Offset in the partition

[M. Manna]

This can happen for two reasons:

1) Your offsets are expired and removed. So your consumers don't know where
to start from - earliest means "Start from the beginning"
2) You are actually starting as part of a totally new consumer group - in
which case it's as designed too - start from the beginning.

I would check your offset retention policy by size/time - and tune that if
necessary.

On 1 June 2018 at 09:03, Dinesh Subramanian 
wrote:

> Hi,
>
> Facing duplication in below scenario
>
> Last commit is happened in 3 days back in the consumer, after that no
> messages produced in the topic. so no commits..
> so after 3 days am stopping and restarting the consumer..  this time i
> faced duplication issue in the consumer as i have this consumer
> property "*auto.offset.reset
> = earliest*", It is consumed again from the beginning.. any helps will be
> appreciated.
>
> *Thanks & Regards,*
>
> *Dinesh S*
>

Re: Is there expiration for committed Offset in the partition

[Hans Jespersen]

You should just recommit the same offsets sooner than every 24 hours (or 
whatever your commit topic retention period is set to). The expiry of offsets 
is based on the timestamp of the commits.

-hans

> On Jun 1, 2018, at 1:03 AM, Dinesh Subramanian  
> wrote:
> 
> Hi,
> 
> Facing duplication in below scenario
> 
> Last commit is happened in 3 days back in the consumer, after that no
> messages produced in the topic. so no commits..
> so after 3 days am stopping and restarting the consumer..  this time i
> faced duplication issue in the consumer as i have this consumer
> property "*auto.offset.reset
> = earliest*", It is consumed again from the beginning.. any helps will be
> appreciated.
> 
> *Thanks & Regards,*
> 
> *Dinesh S*

Is there expiration for committed Offset in the partition

[Dinesh Subramanian]

Hi,

Facing duplication in below scenario

Last commit is happened in 3 days back in the consumer, after that no
messages produced in the topic. so no commits..
so after 3 days am stopping and restarting the consumer..  this time i
faced duplication issue in the consumer as i have this consumer
property "*auto.offset.reset
= earliest*", It is consumed again from the beginning.. any helps will be
appreciated.

*Thanks & Regards,*

*Dinesh S*

Re: Best Practice for Consumer Liveliness and avoid frequent rebalancing

[M. Manna]

What you are talking about is manual partition assignment, which is
different than reassignment upon rebalancing.

Consumer informs group coordinator that close() is invoked and that would
eventually cause rebalancing. I believe what you are talking about is the
rebalance listener.

On 1 Jun 2018 05:41, "Shantanu Deshmukh"  wrote:

Do you want to avoid rebalancing in such way that if a consumer exits then
its previously owned partition should be left disowned? But then who will
consume from partition that was deserted by a exiting consumer? In such
case you can go for manual partition assignment. Then there is no question
of consumer-group management and subsequently rebalancing.


On Thu, May 31, 2018 at 6:00 PM M. Manna  wrote:

> Hello,
>
> We are trying to move from single partition to multi-partition approach
for
> our topics. The purpose is:
>
> 1) Each production/testbed server will have a non-Daemon thread (consumer)
> running.
> 2) It will consume messages, commit offset (manual), and determine next
> steps if commit fails, app fails etc.
> 3) Ideally, 1 partition per server (consumer). If rebalance occurs, first
> (lexi ordered) server will end up having additional partition(s).
>
> As I previously understood, and also read Consumer article by Jason
> Gustafson
> <
>
https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-kafka-0-9-consumer-client/
> >,

> we should always close consumers for resource optimisation. But departing
> from a consumer group means that a rebalance will occur. In our case, we
> would like every consumer to be alive (and sleep for a while) but still
> send heartbeat so that rebalancing effort is saved. But we re worried
> whether this might cause memory leak in our application.
>
> In other words, if we don't restart the servers (shutdown hook), we would
> like to avoid invoking KafkaConsumer#close().
>
> Has anyone got similar use case that they can share with us? We are simply
> interested to know whether this is a "use-case" scenario or not a good
> practice to keep consumers alive.
>
> Any suggestion/help is appreciated.
>
> Regards,
>