0.8/HEAD Console consumer breakage?

[ben fleis]

So I was testing my own code, and using the console consumer against my
seemingly-working-producer code.  Since the last update, the console
consumer crashes.  I am going to try to track it down in the debugger and
will come back with a patch if found.

Re: 0.8/HEAD Console consumer breakage?

[ben fleis]

Dah.  Misfire.  Please ignore if this makes it to an inbox ;)

-b



On Fri, Dec 7, 2012 at 4:13 PM, ben fleis  wrote:

> So I was testing my own code, and using the console consumer against my
> seemingly-working-producer code.  Since the last update, the console
> consumer crashes.  I am going to try to track it down in the debugger and
> will come back with a patch if found.
>
>
>
>
>
>
>

Re: 0.8 Protocol Status

[ben fleis]

I was testing my own code, and using the console consumer against my
seemingly-working-producer code.  Since the last update, the console
consumer crashes.  I am going to try to track it down in the debugger and
will come back with a patch if found.

Command line:
KAFKA_OPTS="-Xmx512M -server
-Dlog4j.configuration=file:$PWD/config/log4j.properties -Xdebug
-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=4244" \
bin/kafka-console-consumer.sh config/consumer.properties --zookeeper
localhost:2181 --topic types

Stacktrace:
[2012-12-07 16:11:34,421] ERROR Error processing message, stopping
consumer:  (kafka.consumer.ConsoleConsumer$)
java.lang.IllegalArgumentException
at java.nio.Buffer.limit(Buffer.java:247)
at kafka.message.Message.sliceDelimited(Message.scala:225)
at kafka.message.Message.payload(Message.scala:207)
at
kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:110)
at
kafka.consumer.ConsumerIterator.makeNext(ConsumerIterator.scala:33)
at
kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:61)
at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:53)
at scala.collection.Iterator$class.foreach(Iterator.scala:631)
at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:32)
at
scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
at kafka.consumer.KafkaStream.foreach(KafkaStream.scala:25)
at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:189)
at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)

All advice gladly accepted, including "you blew it, you blind fool!" ;)

-b

Re: 0.8 Protocol Status

[ben fleis]

Neha, thanks for the tip.  Useful util!

My problem was simple -- I missed one of the size field changes in the
producer, which led to a completely wrong size field.

Am I correct in believing that the broker doesn't sanity check the message
size field against the received data?  In this case, the size of 200k+
makes no sense, and in particular, the broker isn't recording 200k+ worth
of data, so it clearly understand at some level that the payload is shorter.

ben

Re: 0.8 Protocol Status

[ben fleis]

I haven't gone back to check the code, but it feels like every size that's
given can be used to verify how to proceed.

In the case I was having, 200k made no sense because there wasn't 200k
worth of data to write, but the system still flushed the message out to
disk, presumably due to using an earlier size.  My main point is that all
the sizes should add up, and that it costs nothing to check.  For a large
scale system, it never hurts to check, if it's free.  And if something like
that fails, you can always throw an error, and drop the connection --
better than taking on corrupt packets, and redistributing them.

Ok, enough opining for the moment :)

b

order guarantee failure: better/best effort?

[ben fleis]

Hello all,

While running my own system tests against 0.8/HEAD, I was seeing repeated
ordering failures, and tracked it down a bit further.  In simple summary, I
am sending out 2 consecutive ProduceRequests, X and Y.  Y gets written to
disk before X.  Consumer sees Y before X.

Bug #382  discusses this
possibility, and it appears to describe my issue.  My client is, like any
normal client written in Node, asynchronous by nature.  I can add waits to
make it synchronous, but this feels like it fails the "best effort" test.
 I.e., it seems reasonable that a pipelining client running without errors
ought to maintain order, all the time.

Thoughts?

b

order guarantee failure: better/best effort?

[ben fleis]

Hello all,

While running my own system tests against 0.8/HEAD, I was seeing repeated
ordering failures, and tracked it down a bit further.  In simple summary, I
am sending out 2 consecutive ProduceRequests, X and Y.  Y gets written to
disk before X.  Consumer sees Y before X.

Bug #382  discusses this
possibility, and it appears to describe my issue.  My client is, like any
normal client written in Node, asynchronous by nature.  I can add waits to
make it synchronous, but this feels like it fails the "best effort" test.
 I.e., it seems reasonable that a pipelining client running without errors
ought to maintain order, all the time.

Thoughts?

ben

Re: order guarantee failure: better/best effort?

[ben fleis]

Hey Jay,  (yes, I just like the sound of that!)

I have a testing harness for my Node client, but I can't (at the moment)
put it out in the open.  If you'd like, I cab share it via LI channels and
get you into full simulation without too much pain, I think.

And yes, they are same partition, same socket.  I have the tcpdump logs, my
own console and producer logs and the raw kafka files stored away if that's
useful.  They are stored in wrong order on disk, so it's definitely on the
incoming side.

ben

unfortunate test -- delete me

[ben fleis]

testing mailing list funk, sorry.

ben

Re: Splitting client code

[ben fleis]

As a somewhat related follow up, what are the team's intentions wrt
releasing jar files, period?  I see source tarballs, but no precompiled,
ready-to-run jars in the download section.  Is this in the roadmap, easily
obtained, etc?  For some deployment (like mine, where my client is in
another language, and just uses the console consumer as a proxy), there is
little want or need to get the whole build system involved.

Also, I agree with you about the extra pain of developing a pure java
client.  The benefits don't seem particularly strong.  Developing a C/++
client in parallel, that would skip the code duplication problem, help with
the "single perspective" problem, and be an somewhat easy-entrance for
other languages capable of easily integrating via extension compilation.

ben

Re: order guarantee failure: better/best effort?

[ben fleis]

Heey Jy,

Any luck with a java triggering test?

b

Re: Please clean your ZK and Kafka data if you check out 0.8 HEAD

[ben fleis]

Hi Swapnil,

I have done exactly this (killed and wiped everything), and have found that
my custom formatter based ConsoleConsumer fails:

[2013-02-27 12:34:31,420] ERROR Error processing message, stopping
consumer:  (kafka.consumer.ConsoleConsumer$)
java.lang.NumberFormatException: For input string: ""192.168.2.132",
"jmx_port""
at
java.lang.NumberFormatException.forInputString(NumberFormatException.java:48)
at java.lang.Integer.parseInt(Integer.java:449)
at java.lang.Integer.parseInt(Integer.java:499)
at
scala.collection.immutable.StringLike$class.toInt(StringLike.scala:207)
at scala.collection.immutable.StringOps.toInt(StringOps.scala:31)
at kafka.cluster.Broker$.createBroker(Broker.scala:34)
at
kafka.utils.ZkUtils$$anonfun$getCluster$1.apply(ZkUtils.scala:450)
at
kafka.utils.ZkUtils$$anonfun$getCluster$1.apply(ZkUtils.scala:448)
at scala.collection.Iterator$class.foreach(Iterator.scala:631)
at
scala.collection.JavaConversions$JIteratorWrapper.foreach(JavaConversions.scala:474)
at
scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
at
scala.collection.JavaConversions$JListWrapper.foreach(JavaConversions.scala:521)
at kafka.utils.ZkUtils$.getCluster(ZkUtils.scala:448)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener$$anonfun$syncedRebalance$1.apply$mcVI$sp(ZookeeperConsumerConnector.scala:371)
at
scala.collection.immutable.Range$ByOne$class.foreach$mVc$sp(Range.scala:282)
at
scala.collection.immutable.Range$$anon$2.foreach$mVc$sp(Range.scala:265)
at
kafka.consumer.ZookeeperConsumerConnector$ZKRebalancerListener.syncedRebalance(ZookeeperConsumerConnector.scala:368)
at
kafka.consumer.ZookeeperConsumerConnector.kafka$consumer$ZookeeperConsumerConnector$$reinitializeConsumer(ZookeeperConsumerConnector.scala:697)
at
kafka.consumer.ZookeeperConsumerConnector$WildcardStreamsHandler.(ZookeeperConsumerConnector.scala:732)
at
kafka.consumer.ZookeeperConsumerConnector.createMessageStreamsByFilter(ZookeeperConsumerConnector.scala:141)
at kafka.consumer.ConsoleConsumer$.main(ConsoleConsumer.scala:183)
at kafka.consumer.ConsoleConsumer.main(ConsoleConsumer.scala)

I haven't tracked down why the mismatch only occurs with my formatter, but
advice is appreciated.

ben

Re: Please clean your ZK and Kafka data if you check out 0.8 HEAD

[ben fleis]

Sorry, I should have clarified something -- using the normal
ConsoleConsumer does *not* produce this error.  So something in my
formatter, or in the logic of a custom formatter must change?

ben

Re: Please clean your ZK and Kafka data if you check out 0.8 HEAD

[ben fleis]

Is it possible to upgrade one without upgrading the other?  I did a git
pull, ./sbt update package.  Then for kicks I did sbt clean package, just
to be sure.

ben

[jira] [Commented] (KAFKA-689) Can't append to a topic/partition that does not already exist

[ben fleis (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-689?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13549500#comment-13549500
 ] 

ben fleis commented on KAFKA-689:
-

Although it's not precisely the same, perhaps thinking about topic|partition as 
a remote file open is a useful metaphor.  An open() call is where you would set 
normal open params (flush interval, O_CREAT, etc.), and stat() is where you get 
broker and other real time updates.  Of course, if create is explicit, where 
does delete come into play?

@Jun - I don't see where in server.properties anything about topic creation 
exists?  And further, does an extra RPC matter if it's only during 
setup/periodic?

> Can't append to a topic/partition that does not already exist
> -
>
> Key: KAFKA-689
> URL: https://issues.apache.org/jira/browse/KAFKA-689
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.8
>Reporter: David Arthur
> Attachments: kafka.log, produce-payload.bin
>
>
> With a totally fresh Kafka (empty logs dir and empty ZK), if I send a 
> ProduceRequest for a new topic, Kafka responds with 
> "kafka.common.UnknownTopicOrPartitionException: Topic test partition 0 
> doesn't exist on 0". This is when sending a ProduceRequest over the network 
> (from Python, in this case).
> If I use the console producer it works fine (topic and partition get 
> created). If I then send the same payload from before over the network, it 
> works.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Commented] (KAFKA-133) Publish kafka jar to a public maven repository

[ben fleis (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-133?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13549515#comment-13549515
 ] 

ben fleis commented on KAFKA-133:
-

I'm new to maven, so this may be a dumb question: would it be reasonable/easy 
to publish nightly jars (via Apache?), under the 0.8.0-SNAPSHOT tag?  I have an 
automated build system that currently git's and builds the whole thing, which I 
would love to replace with simple jar files.

> Publish kafka jar to a public maven repository
> --
>
> Key: KAFKA-133
> URL: https://issues.apache.org/jira/browse/KAFKA-133
> Project: Kafka
>  Issue Type: Improvement
>Affects Versions: 0.6, 0.8
>Reporter: Neha Narkhede
>  Labels: patch
> Fix For: 0.8
>
> Attachments: KAFKA-133.patch, pom.xml
>
>
> The released kafka jar must be download manually and then deploy to a private 
> repository before they can be used by a developer using maven2.
> Similar to other Apache projects, it will be nice to have a way to publish 
> Kafka releases to a public maven repo. 
> In the past, we gave it a try using sbt publish to Sonatype Nexus maven repo, 
> but ran into some authentication problems. It will be good to revisit this 
> and get it resolved.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Updated] (KAFKA-692) ConsoleConsumer outputs diagnostic message to stdout instead of stderr

[ben fleis (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-692?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

ben fleis updated KAFKA-692:


Status: Patch Available  (was: Open)

from stdout -> stderr

> ConsoleConsumer outputs diagnostic message to stdout instead of stderr
> --
>
> Key: KAFKA-692
> URL: https://issues.apache.org/jira/browse/KAFKA-692
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.8
>Reporter: ben fleis
>Priority: Minor
> Fix For: 0.8
>
>   Original Estimate: 1m
>  Remaining Estimate: 1m
>
> At the end of its handling loop, ConsoleConsumer prints "Consumed %d 
> messages" to standard out.  Clients who use customer formatters, and read 
> this output, shouldn't need to special case this line, or accept a parse 
> error.
> It should instead go (as all diagnostics should) to stderr.
> patch attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Created] (KAFKA-692) ConsoleConsumer outputs diagnostic message to stdout instead of stderr

[ben fleis (JIRA)]

ben fleis created KAFKA-692:
---

 Summary: ConsoleConsumer outputs diagnostic message to stdout 
instead of stderr
 Key: KAFKA-692
 URL: https://issues.apache.org/jira/browse/KAFKA-692
 Project: Kafka
  Issue Type: Bug
  Components: clients
Affects Versions: 0.8
Reporter: ben fleis
Priority: Minor
 Fix For: 0.8


At the end of its handling loop, ConsoleConsumer prints "Consumed %d messages" 
to standard out.  Clients who use customer formatters, and read this output, 
shouldn't need to special case this line, or accept a parse error.

It should instead go (as all diagnostics should) to stderr.

patch attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Updated] (KAFKA-692) ConsoleConsumer outputs diagnostic message to stdout instead of stderr

[ben fleis (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-692?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

ben fleis updated KAFKA-692:


Attachment: kafka_692_v1.diff

stdout -> stderr

> ConsoleConsumer outputs diagnostic message to stdout instead of stderr
> --
>
> Key: KAFKA-692
> URL: https://issues.apache.org/jira/browse/KAFKA-692
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.8
>Reporter: ben fleis
>Priority: Minor
> Fix For: 0.8
>
> Attachments: kafka_692_v1.diff
>
>   Original Estimate: 1m
>  Remaining Estimate: 1m
>
> At the end of its handling loop, ConsoleConsumer prints "Consumed %d 
> messages" to standard out.  Clients who use customer formatters, and read 
> this output, shouldn't need to special case this line, or accept a parse 
> error.
> It should instead go (as all diagnostics should) to stderr.
> patch attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Issue Comment Deleted] (KAFKA-692) ConsoleConsumer outputs diagnostic message to stdout instead of stderr

[ben fleis (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-692?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

ben fleis updated KAFKA-692:


Comment: was deleted

(was: from stdout -> stderr)

> ConsoleConsumer outputs diagnostic message to stdout instead of stderr
> --
>
> Key: KAFKA-692
> URL: https://issues.apache.org/jira/browse/KAFKA-692
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.8
>Reporter: ben fleis
>Priority: Minor
> Fix For: 0.8
>
> Attachments: kafka_692_v1.diff
>
>   Original Estimate: 1m
>  Remaining Estimate: 1m
>
> At the end of its handling loop, ConsoleConsumer prints "Consumed %d 
> messages" to standard out.  Clients who use customer formatters, and read 
> this output, shouldn't need to special case this line, or accept a parse 
> error.
> It should instead go (as all diagnostics should) to stderr.
> patch attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Updated] (KAFKA-692) ConsoleConsumer outputs diagnostic message to stdout instead of stderr

[ben fleis (JIRA)]

 [ 
https://issues.apache.org/jira/browse/KAFKA-692?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
 ]

ben fleis updated KAFKA-692:


Description: 
At the end of its handling loop, ConsoleConsumer prints "Consumed %d messages" 
to standard out.  Clients who use custom formatters, and read this output, 
shouldn't need to special case this line, or accept a parse error.

It should instead go (as all diagnostics should) to stderr.

patch attached.

  was:
At the end of its handling loop, ConsoleConsumer prints "Consumed %d messages" 
to standard out.  Clients who use customer formatters, and read this output, 
shouldn't need to special case this line, or accept a parse error.

It should instead go (as all diagnostics should) to stderr.

patch attached.


> ConsoleConsumer outputs diagnostic message to stdout instead of stderr
> --
>
> Key: KAFKA-692
> URL: https://issues.apache.org/jira/browse/KAFKA-692
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>    Affects Versions: 0.8
>Reporter: ben fleis
>Priority: Minor
> Fix For: 0.8
>
> Attachments: kafka_692_v1.diff
>
>   Original Estimate: 1m
>  Remaining Estimate: 1m
>
> At the end of its handling loop, ConsoleConsumer prints "Consumed %d 
> messages" to standard out.  Clients who use custom formatters, and read this 
> output, shouldn't need to special case this line, or accept a parse error.
> It should instead go (as all diagnostics should) to stderr.
> patch attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Commented] (KAFKA-692) ConsoleConsumer outputs diagnostic message to stdout instead of stderr

[ben fleis (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-692?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13552214#comment-13552214
 ] 

ben fleis commented on KAFKA-692:
-

Agreed - I just patched it so it didn't get in my way on my setup.  In general, 
it's not clear why it's particular useful/helpful.

> ConsoleConsumer outputs diagnostic message to stdout instead of stderr
> --
>
> Key: KAFKA-692
> URL: https://issues.apache.org/jira/browse/KAFKA-692
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>    Affects Versions: 0.8
>        Reporter: ben fleis
>Assignee: ben fleis
>Priority: Minor
> Fix For: 0.8
>
> Attachments: kafka_692_v1.diff
>
>   Original Estimate: 1m
>  Remaining Estimate: 1m
>
> At the end of its handling loop, ConsoleConsumer prints "Consumed %d 
> messages" to standard out.  Clients who use custom formatters, and read this 
> output, shouldn't need to special case this line, or accept a parse error.
> It should instead go (as all diagnostics should) to stderr.
> patch attached.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Created] (KAFKA-704) ConsumerFetcherThread can create illegal clientId

[ben fleis (JIRA)]

ben fleis created KAFKA-704:
---

 Summary: ConsumerFetcherThread can create illegal clientId
 Key: KAFKA-704
 URL: https://issues.apache.org/jira/browse/KAFKA-704
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Affects Versions: 0.8
Reporter: ben fleis
Assignee: Neha Narkhede


When updating to recent changes, I found that my ConsoleConsumer with a custom 
formatter no longer works.  I tracked down the error to a change in 
ConsumerFetcherThread, whereby the clientId is now the concatenation of user 
specified config.clientId and 'name'.  In turn, 'name' can include the 
hostname, which in my case was "mn-bfleis.local".  '.' is an illegal character, 
thus in my case, the consumer systematically feeds itself an invalid value.  
And of course, it breaks.

I don't know what the proper fix is -- to allow '.', or change the way 'name' 
is generated.  The previous revision didn't concatenate, so I don't know 
whether it's truly necessary.  Locally I am just removed the concat step until 
this is resolved.

The breaking change occurred at git hash 03eb903c, whose commit log read:
KAFKA-683 Fix correlation id in all requests sent to kafka; reviewed by Jun 
Rao



I doubt it's needed, but jic, original trace below:

[2013-01-15 14:13:19,732] WARN 
[_bfleis-mn.local-1358255598567-67426e89-leader-finder-thread], Failed to 
find leader for Set([types,2], [types,1], [types,0]) 
(kafka.consumer.ConsumerFetcherManager$$anon$1)
kafka.common.InvalidConfigException: client.id 
-ConsumerFetcherThread-_bfleis-mn.local-1358255598567-67426e89-0-0 is 
illegal, contains a character other than ASCII alphanumerics, _ and -
at kafka.common.Config$class.validateChars(Config.scala:32)
at kafka.consumer.ConsumerConfig$.validateChars(ConsumerConfig.scala:25)
at 
kafka.consumer.ConsumerConfig$.validateClientId(ConsumerConfig.scala:55)
at kafka.consumer.SimpleConsumer.(SimpleConsumer.scala:89)
at 
kafka.server.AbstractFetcherThread.(AbstractFetcherThread.scala:44)
at 
kafka.consumer.ConsumerFetcherThread.(ConsumerFetcherThread.scala:27)
at 
kafka.consumer.ConsumerFetcherManager.createFetcherThread(ConsumerFetcherManager.scala:93)
at 
kafka.server.AbstractFetcherManager.addFetcher(AbstractFetcherManager.scala:44)
at 
kafka.consumer.ConsumerFetcherManager$$anon$1$$anonfun$doWork$3.apply(ConsumerFetcherManager.scala:75)
at 
kafka.consumer.ConsumerFetcherManager$$anon$1$$anonfun$doWork$3.apply(ConsumerFetcherManager.scala:72)
at 
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:80)
at 
scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:80)
at scala.collection.Iterator$class.foreach(Iterator.scala:631)
at 
scala.collection.mutable.HashTable$$anon$1.foreach(HashTable.scala:161)
at 
scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:194)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:39)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:80)
at 
kafka.consumer.ConsumerFetcherManager$$anon$1.doWork(ConsumerFetcherManager.scala:72)
at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:50)

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Created] (KAFKA-706) broker appears to be encoding ProduceResponse, but never sending it

[ben fleis (JIRA)]

ben fleis created KAFKA-706:
---

 Summary: broker appears to be encoding ProduceResponse, but never 
sending it
 Key: KAFKA-706
 URL: https://issues.apache.org/jira/browse/KAFKA-706
 Project: Kafka
  Issue Type: Bug
  Components: core
Affects Versions: 0.8
 Environment: reproduced on both Mac OS and RH linux, via private 
node.js client
Reporter: ben fleis


By all appearances, I seem to be able to convince a broker to periodically 
encode, but never transmit, a ProduceResponse.  Unfortunately my client is 
proprietary, but I will share it with Neha via LI channels.  But I will 
describe what's going on in the hopes that there's another trivial way to 
reproduce it.  (I did search through JIRA, and haven't found anything that 
looks like this.)

I am running a single instance zookeeper and single broker.  I have a client 
that generates configurable amounts of data, tracking what is produced (both 
sent and ACK'd), and what is consumed.  I was noticing that when using high 
transfer rates via high frequency single messages, my unack'd queue appeared to 
be getting continuously larger.  So, I outfitted my client to log more 
information about correlation ids at various stages, and modified the kafka 
ProducerRequest/ProducerResponse to log (de)serialization of the same.  I then 
used tcpdump to intercept all communications between my client and the broker.  
Finally, I configured my client to generate 1 message per ~10ms, each payload 
being approximately 33 bytes; requestAckTimeout was set to 2000ms, and 
requestAcksRequired was set to 1.  I used 10ms as I found that 5ms or less 
caused my unacked queue to build up due to system speed -- it simply couldn't 
keep up.  10ms keeps the load high, but just manageable.  YMMV with that param. 
 All of this is done on a single host, over loopback.  I ran it on both my 
airbook, and a well setup RH linux box, and found the same problem.

At startup, my system logged "expired" requests - meaning reqs that were sent, 
but for which no ACK, positive or negative, was seen from the broker, within 
1.25x the requestAckTimeout (ie, 2500ms).  I would let it settle until the 
unacked queue was stable at or around 0.

What I found is this: ACKs are normally generated within milliseconds.  This 
was demonstrated by my logging added to the scala ProducerRe* classes, and they 
are normally seen quickly by my client.  But when the actual error occurs, 
namely that a request is ignored, the ProducerResponse class *does* encode the 
correct correlationId; however, a response containing that ID is never sent 
over the network, as evidenced by my tcpdump traces.  In my experience this 
would take anywhere from 3-15 seconds to occur after the system was warm, 
meaning that it's 1 out of several hundred on average that shows the condition.

While I can't attach my client code, I could attach logs; but since my 
intention is to share the code with LI people, I will wait to see if that's 
useful here.


--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Commented] (KAFKA-736) Add an option to the 0.8 producer to mimic 0.7 producer behavior

[ben fleis (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-736?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13568192#comment-13568192
 ] 

ben fleis commented on KAFKA-736:
-

I tried the v2 patch linked above this morning against v0.8 HEAD.  It appears 
(after full throttle testing for over an hour) to have eliminated the problem 
originally reported in KAFKA-706.

> Add an option to the 0.8 producer to mimic 0.7 producer behavior
> 
>
> Key: KAFKA-736
> URL: https://issues.apache.org/jira/browse/KAFKA-736
> Project: Kafka
>  Issue Type: Improvement
>  Components: producer 
>Affects Versions: 0.8
>Reporter: Neha Narkhede
>Assignee: Neha Narkhede
>Priority: Blocker
>  Labels: p2, replication-performance
> Attachments: kafka-736-draft.patch, kafka-736-v1.patch, 
> kafka-736-v2.patch
>
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> I profiled a producer throughput benchmark between a producer and a remote 
> broker. It turns out that the background send threads spends ~97% of its time 
> waiting to read the acknowledgement from the broker.
> I propose we change the current behavior of request.required.acks=0 to mean 
> no acknowledgement from the broker. This will mimic the 0.7 producer behavior 
> and will enable tuning the producer for very high throughput.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Commented] (KAFKA-736) Add an option to the 0.8 producer to mimic 0.7 producer behavior

[ben fleis (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-736?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13569294#comment-13569294
 ] 

ben fleis commented on KAFKA-736:
-

Went back and tested the 'draft' patch, and this failed in the same way that 
706 was previously failing.  I have not looked at the patches at all, merely 
blindly applied.  Perhaps I made an error when applying/testing the v2 patch, 
although I believe I ran the same steps...  YMMV.  In any case draft patch has 
reliably failed 3x in a row, and a la 706, tcpdump confirms the lost messages.

> Add an option to the 0.8 producer to mimic 0.7 producer behavior
> 
>
> Key: KAFKA-736
> URL: https://issues.apache.org/jira/browse/KAFKA-736
> Project: Kafka
>  Issue Type: Improvement
>  Components: producer 
>Affects Versions: 0.8
>Reporter: Neha Narkhede
>Assignee: Neha Narkhede
>Priority: Blocker
>  Labels: p2, replication-performance
> Attachments: kafka-736-draft.patch, kafka-736-v1.patch, 
> kafka-736-v2.patch
>
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> I profiled a producer throughput benchmark between a producer and a remote 
> broker. It turns out that the background send threads spends ~97% of its time 
> waiting to read the acknowledgement from the broker.
> I propose we change the current behavior of request.required.acks=0 to mean 
> no acknowledgement from the broker. This will mimic the 0.7 producer behavior 
> and will enable tuning the producer for very high throughput.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira

[jira] [Commented] (KAFKA-736) Add an option to the 0.8 producer to mimic 0.7 producer behavior

[ben fleis (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-736?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=13569320#comment-13569320
 ] 

ben fleis commented on KAFKA-736:
-

This is more-or-less what my node command line tool does, among other things.  
Not threaded, but event driven.  But spawns both sides, uses custom 
correlation_ids for debugging, etc.  Unfortunately, I can't release it, but if 
there's something to learn from it, it's available to you.

> Add an option to the 0.8 producer to mimic 0.7 producer behavior
> 
>
> Key: KAFKA-736
> URL: https://issues.apache.org/jira/browse/KAFKA-736
> Project: Kafka
>  Issue Type: Improvement
>  Components: producer 
>Affects Versions: 0.8
>Reporter: Neha Narkhede
>Assignee: Neha Narkhede
>Priority: Blocker
>  Labels: p2, replication-performance
> Attachments: kafka-736-draft.patch, kafka-736-v1.patch, 
> kafka-736-v2.patch
>
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> I profiled a producer throughput benchmark between a producer and a remote 
> broker. It turns out that the background send threads spends ~97% of its time 
> waiting to read the acknowledgement from the broker.
> I propose we change the current behavior of request.required.acks=0 to mean 
> no acknowledgement from the broker. This will mimic the 0.7 producer behavior 
> and will enable tuning the producer for very high throughput.

--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira