Home | About | Sematext search-lucene.com search-hadoop.com
NEW: Monitor These Apps!
elasticsearch, apache solr, apache hbase, hadoop, redis, casssandra, amazon cloudwatch, mysql, memcached, apache kafka, apache zookeeper, apache storm, ubuntu, centOS, red hat, debian, puppet labs, java, senseiDB
 Search Hadoop and all its subprojects:

Switch to Threaded View
Kafka >> mail # user >> producer exceptions when broker dies


Copy link to this message
-
Re: producer exceptions when broker dies
Or, to rephrase it more generally, is there a way to know exactly if
message was committed or no?
On Fri, Oct 25, 2013 at 10:43 AM, Kane Kane <[EMAIL PROTECTED]> wrote:

> Hello Guozhang,
>
> My partitions are split almost evenly between broker, so, yes - broker
> that I shutdown is the leader for some of them. Does it mean i can get an
> exception and data is still being written? Is there any setting on the
> broker where i can control this? I.e. can i make broker replication timeout
> shorter than producer timeout, so i can ensure if i get an exception data
> is not being committed?
>
> Thanks.
>
>
> On Fri, Oct 25, 2013 at 10:36 AM, Guozhang Wang <[EMAIL PROTECTED]>wrote:
>
>> Hello Kane,
>>
>> As discussed in the other thread, even if a timeout response is sent back
>> to the producer, the message may still be committed.
>>
>> Did you shut down the leader broker of the partition or a follower broker?
>>
>> Guozhang
>>
>> On Fri, Oct 25, 2013 at 8:45 AM, Kane Kane <[EMAIL PROTECTED]> wrote:
>>
>> > I have cluster of 3 kafka brokers. With the following script I send some
>> > data to kafka and in the middle do the controlled shutdown of 1 broker.
>> All
>> > 3 brokers are ISR before I start sending. When i shutdown the broker i
>> get
>> > a couple of exceptions and I expect data shouldn't be written. Say, I
>> send
>> > 1500 lines and get 50 exceptions. I expect to consume 1450 lines, but
>> > instead i always consume more, i.e. 1480 or 1490. I want to decide if I
>> > want to retry sending myself, not using message.send.max.retries. But
>> looks
>> > like if I retry sending if there is an exception - I will end up with
>> > duplicates. Is there anything I'm doing wrong or having wrong
>> assumptions
>> > about kafka?
>> >
>> > Thanks.
>> >
>> > val prod = new MyProducer("10.80.42.147:9092,10.80.42.154:9092,
>> > 10.80.42.156:9092")
>> > var count = 0
>> > for(line <- Source.fromFile(file).getLines()){
>> >     try {
>> >       prod.send("benchmark", buffer.toList)
>> >       count += 1
>> >       println("sent %s", count)
>> >     } catch {
>> >       case _ => println("Exception!")
>> >     }
>> > }
>> >
>> > class MyProducer(brokerList: String) {
>> >   val sync = true
>> >   val requestRequiredAcks = "-1"
>> >
>> >   val props = new Properties()
>> >   props.put("metadata.broker.list", brokerList)
>> >   props.put("producer.type", if(sync) "sync" else "async")
>> >   props.put("request.required.acks", requestRequiredAcks)
>> >   props.put("key.serializer.class", classOf[StringEncoder].getName)
>> >   props.put("serializer.class", classOf[StringEncoder].getName)
>> >   props.put("message.send.max.retries", "0")
>> >   props.put("request.timeout.ms", "2000")
>> >
>> >   val producer = new Producer[AnyRef, AnyRef](new ProducerConfig(props))
>> >
>> >   def send(topic: String, messages: List[String]) = {
>> >     val requests = new ArrayBuffer[KeyedMessage[AnyRef, AnyRef]]
>> >     for (message <- messages) {
>> >       requests += new KeyedMessage(topic, null, message, message)
>> >     }
>> >     producer.send(requests)
>> >   }
>> > }
>> >
>>
>>
>>
>> --
>> -- Guozhang
>>
>
>

 
NEW: Monitor These Apps!
elasticsearch, apache solr, apache hbase, hadoop, redis, casssandra, amazon cloudwatch, mysql, memcached, apache kafka, apache zookeeper, apache storm, ubuntu, centOS, red hat, debian, puppet labs, java, senseiDB