[jira] [Comment Edited] (KAFKA-4739) KafkaConsumer poll going into an infinite loop

[Jason Gustafson (JIRA)]

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

Jason Gustafson edited comment on KAFKA-4739 at 2/10/17 5:48 AM:
-

[~neoeahit] Thanks for the update. These logs look much more like what I 
expect. There doesn't appear to be anything wrong with the consumer, but it 
would be nice to confirm it. 

1. Can you use the console consumer both with and without the 
{{--new-consumer}} option to consume from the same exact topics? 
2. You mentioned that you can reproduce the problem by restarting the process 
repeatedly, can you explain this a bit more? Can you get the same behavior 
using the console consumer? 
3. If you can get the consumer into this state again, can you get a thread dump 
from the broker? There have been a couple deadlocks fixed on the broker since 
0.9.0.1, so it would be nice to confirm that we're not hitting one of them.
4. What do you do currently to recover? Since you said restarting the consumers 
doesn't help, what does? Restarting the brokers?


was (Author: hachikuji):
[~neoeahit] Thanks for the update. These logs look much more like what I 
expect. There doesn't appear to be anything wrong with the consumer, but it 
would be nice to confirm it. 

1. Can you use the console consumer both with and without the 
{{--new-consumer}} option to consume from the same exact topics? 
2. You mentioned that you can reproduce the problem by restarting the process 
repeatedly, can you explain this a bit more? Can you get the same behavior 
using the console consumer? 
3. If you can get the consumer into this state again, can you get a thread dump 
from the broker? There have been a couple deadlocks fixed since 0.9.0.1, so it 
would be nice to confirm that we're not hitting one of them.
4. What do you do currently to recover? Since you said restarting the consumers 
doesn't help, what does? Restarting the brokers?

> KafkaConsumer poll going into an infinite loop
> --
>
> Key: KAFKA-4739
> URL: https://issues.apache.org/jira/browse/KAFKA-4739
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.1
>Reporter: Vipul Singh
>
> We are seeing an issue with our kafka consumer where it seems to go into an 
> infinite loop while polling, trying to fetch data from kafka. We are seeing 
> the heartbeat requests on the broker from the consumer, but nothing else from 
> the kafka consumer.
> We enabled debug level logging on the consumer, and see these logs: 
> https://gist.github.com/neoeahit/757bff7acdea62656f065f4dcb8974b4
> And this just goes on. The way we have been able to replicate this issue, is 
> by restarting the process in multiple successions.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4753) KafkaConsumer susceptible to FetchResponse starvation

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4753:


Could scenario 1 be mitigated with a new partition assignor which took into 
account partition leadership?

> KafkaConsumer susceptible to FetchResponse starvation
> -
>
> Key: KAFKA-4753
> URL: https://issues.apache.org/jira/browse/KAFKA-4753
> Project: Kafka
>  Issue Type: Bug
>Reporter: Onur Karaman
>Assignee: Onur Karaman
>
> FetchResponse starvation here means that the KafkaConsumer repeatedly fails 
> to fully form FetchResponses within the request timeout from a subset of the 
> brokers its fetching from while FetchResponses from the other brokers can get 
> fully formed and processed by the application.
> In other words, this ticket is concerned with scenarios where fetching from 
> some brokers hurts the progress of fetching from other brokers to the point 
> of repeatedly hitting a request timeout.
> Some FetchResponse starvation scenarios:
> 1. partition leadership of the consumer's assigned partitions is skewed 
> across brokers, causing uneven FetchResponse sizes across brokers.
> 2. the consumer seeks back on partitions on some brokers but not others, 
> causing uneven FetchResponse sizes across brokers.
> 3. the consumer's ability to keep up with various partitions across brokers 
> is skewed, causing uneven FetchResponse sizes across brokers.
> I've personally seen scenario 1 happen this past week to one of our users in 
> prod. They manually assigned partitions such that a few brokers led most of 
> the partitions while other brokers only led a single partition. When 
> NetworkClient sends out FetchRequests to different brokers in parallel with 
> an uneven partition distribution, FetchResponses from brokers who lead more 
> partitions will contain more data than FetchResponses from brokers who lead 
> few partitions. This means the small FetchResponses will get fully formed 
> quicker than larger FetchResponses. When the application eventually consumes 
> a smaller fully formed FetchResponses, the NetworkClient will send out a new 
> FetchRequest to the lightly-loaded broker. Their response will again come 
> back quickly while only marginal progress has been made on the larger 
> FetchResponse. Repeat this process several times and your application will 
> have potentially processed many smaller FetchResponses while the larger 
> FetchResponse made minimal progress and is forced to timeout, causing the 
> large FetchResponse to start all over again, which causes starvation.
> To mitigate the problem for the short term, I've suggested to our user that 
> they either:
> 1. bump up their "receive.buffer.bytes" beyond the current default of 64 KB 
> to something like 1 MB. This is the solution I short-term solution I 
> suggested they go with.
> 2. reduce the "max.partition.fetch.bytes" down from the current default of 1 
> MB to something like 100 KB. This solution wasn't advised as it could impact 
> broker performance.
> 3. ask our SRE's to run a partition reassignment to balance out the partition 
> leadership (partitions were already being led by their preferred leaders).
> 4. bump up their request timeout. It was set to open-source's former default 
> of 40 seconds.
> Contributing factors:
> 1. uneven FetchResponse sizes across brokers.
> 2. processing time of the polled ConsumerRecords.
> 3. "max.poll.records" increases the number of polls needed to consume a 
> FetchResponse, making constant-time overhead per poll magnified.
> 4. "max.poll.records" makes KafkaConsumer.poll bypass calls to 
> ConsumerNetworkClient.poll.
> 5. java.nio.channels.Selector.select, Selector.poll, NetworkClient.poll, and 
> ConsumerNetworkClient.poll can return before the poll timeout as soon as a 
> single channel is selected.
> 6. NetworkClient.poll is solely driven by the user thread with manual 
> partition assignment.
> So far I've only locally reproduced starvation scenario 1 and haven't even 
> attempted the other scenarios. Preventing the bypass of 
> ConsumerNetworkClient.poll (contributing factor 3) mitigates the issue, but 
> it seems starvation would still be possible.
> How to reproduce starvation scenario 1:
> 1. startup zookeeper
> 2. startup two brokers
> 3. create a topic t0 with two partitions led by broker 0 and create a topic 
> t1 with a single partition led by broker 1
> {code}
> > ./bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic t0 
> > --replica-assignment 0,0
> > ./bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic t1 
> > --replica-assignment 1
> {code}
> 4. Produce a lot of data into these topics
> {code}
> > ./bin/kafka-producer-perf-test.sh --topic t0 --num-records 2000 
> > 

[jira] [Commented] (KAFKA-4159) Allow overriding producer & consumer properties at the connector level

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4159:


[~sjdurfey] I've added you as a contributor. You should be able to assign this 
and other Kafka jiras.

> Allow overriding producer & consumer properties at the connector level
> --
>
> Key: KAFKA-4159
> URL: https://issues.apache.org/jira/browse/KAFKA-4159
> Project: Kafka
>  Issue Type: New Feature
>  Components: KafkaConnect
>Reporter: Shikhar Bhushan
>
> As an example use cases, overriding a sink connector's consumer's partition 
> assignment strategy.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-4758) Connect WorkerSinkTask is missing checks for NO_TIMESTAMP

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-4758.

   Resolution: Fixed
Fix Version/s: 0.10.3.0

Issue resolved by pull request 2533
[https://github.com/apache/kafka/pull/2533]

> Connect WorkerSinkTask is missing checks for NO_TIMESTAMP
> -
>
> Key: KAFKA-4758
> URL: https://issues.apache.org/jira/browse/KAFKA-4758
> Project: Kafka
>  Issue Type: Bug
>  Components: KafkaConnect
>Reporter: Jason Gustafson
>Assignee: Ryan P
> Fix For: 0.10.3.0
>
>
> The current check for NO_TIMESTAMP_TYPE is not sufficient. Upconverted 
> messages will have a timestamp type, but if the topic is set to use 
> CREAT_TIME, the timestamp will be NO_TIMESTAMP (-1). We should use {{null}} 
> in that case.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Work started] (KAFKA-4761) NullPointerException if batch.size=0 for producer config

[Jason Gustafson (JIRA)]

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

Work on KAFKA-4761 started by Jason Gustafson.
--
> NullPointerException if batch.size=0 for producer config
> 
>
> Key: KAFKA-4761
> URL: https://issues.apache.org/jira/browse/KAFKA-4761
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.2.0
> Environment: jdk1.8.0_40
>Reporter: Swen Moczarski
>Assignee: Jason Gustafson
>Priority: Minor
> Attachments: KafkaProducerTest.java
>
>
> When setting {{batch.size}} to {{0}} for producer I get the following 
> exception when sending a record:
> {noformat}
> java.lang.NullPointerException
>   at org.apache.kafka.common.utils.Utils.notNull(Utils.java:315)
>   at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:197)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:478)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:440)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:360)
>   at KafkaProducerTest.exposeNpeOnBatchSizeZero(KafkaProducerTest.java:21)
> {noformat}
> But documentation says:
> {quote}
>  ... (a batch size of zero will disable batching entirely).
> {quote}
> Please, see attached test.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Assigned] (KAFKA-4761) NullPointerException if batch.size=0 for producer config

[Jason Gustafson (JIRA)]

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

Jason Gustafson reassigned KAFKA-4761:
--

Assignee: Jason Gustafson

> NullPointerException if batch.size=0 for producer config
> 
>
> Key: KAFKA-4761
> URL: https://issues.apache.org/jira/browse/KAFKA-4761
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.2.0
> Environment: jdk1.8.0_40
>Reporter: Swen Moczarski
>Assignee: Jason Gustafson
>Priority: Minor
> Attachments: KafkaProducerTest.java
>
>
> When setting {{batch.size}} to {{0}} for producer I get the following 
> exception when sending a record:
> {noformat}
> java.lang.NullPointerException
>   at org.apache.kafka.common.utils.Utils.notNull(Utils.java:315)
>   at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:197)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:478)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:440)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:360)
>   at KafkaProducerTest.exposeNpeOnBatchSizeZero(KafkaProducerTest.java:21)
> {noformat}
> But documentation says:
> {quote}
>  ... (a batch size of zero will disable batching entirely).
> {quote}
> Please, see attached test.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4761) NullPointerException if batch.size=0 for producer config

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4761:
---
Status: Patch Available  (was: In Progress)

https://github.com/apache/kafka/pull/2545/files

> NullPointerException if batch.size=0 for producer config
> 
>
> Key: KAFKA-4761
> URL: https://issues.apache.org/jira/browse/KAFKA-4761
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.2.0
> Environment: jdk1.8.0_40
>Reporter: Swen Moczarski
>Assignee: Jason Gustafson
>Priority: Minor
> Attachments: KafkaProducerTest.java
>
>
> When setting {{batch.size}} to {{0}} for producer I get the following 
> exception when sending a record:
> {noformat}
> java.lang.NullPointerException
>   at org.apache.kafka.common.utils.Utils.notNull(Utils.java:315)
>   at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:197)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:478)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:440)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:360)
>   at KafkaProducerTest.exposeNpeOnBatchSizeZero(KafkaProducerTest.java:21)
> {noformat}
> But documentation says:
> {quote}
>  ... (a batch size of zero will disable batching entirely).
> {quote}
> Please, see attached test.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4761) NullPointerException if batch.size=0 for producer config

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4761:
---
Priority: Blocker  (was: Minor)

> NullPointerException if batch.size=0 for producer config
> 
>
> Key: KAFKA-4761
> URL: https://issues.apache.org/jira/browse/KAFKA-4761
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.2.0
> Environment: jdk1.8.0_40
>Reporter: Swen Moczarski
>Assignee: Jason Gustafson
>Priority: Blocker
> Fix For: 0.10.2.0
>
> Attachments: KafkaProducerTest.java
>
>
> When setting {{batch.size}} to {{0}} for producer I get the following 
> exception when sending a record:
> {noformat}
> java.lang.NullPointerException
>   at org.apache.kafka.common.utils.Utils.notNull(Utils.java:315)
>   at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:197)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:478)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:440)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:360)
>   at KafkaProducerTest.exposeNpeOnBatchSizeZero(KafkaProducerTest.java:21)
> {noformat}
> But documentation says:
> {quote}
>  ... (a batch size of zero will disable batching entirely).
> {quote}
> Please, see attached test.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4761) NullPointerException if batch.size=0 for producer config

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4761:
---
Fix Version/s: 0.10.2.0

> NullPointerException if batch.size=0 for producer config
> 
>
> Key: KAFKA-4761
> URL: https://issues.apache.org/jira/browse/KAFKA-4761
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.2.0
> Environment: jdk1.8.0_40
>Reporter: Swen Moczarski
>Assignee: Jason Gustafson
>Priority: Minor
> Fix For: 0.10.2.0
>
> Attachments: KafkaProducerTest.java
>
>
> When setting {{batch.size}} to {{0}} for producer I get the following 
> exception when sending a record:
> {noformat}
> java.lang.NullPointerException
>   at org.apache.kafka.common.utils.Utils.notNull(Utils.java:315)
>   at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:197)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:478)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:440)
>   at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:360)
>   at KafkaProducerTest.exposeNpeOnBatchSizeZero(KafkaProducerTest.java:21)
> {noformat}
> But documentation says:
> {quote}
>  ... (a batch size of zero will disable batching entirely).
> {quote}
> Please, see attached test.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Commented] (KAFKA-4686) Null Message payload is shutting down broker

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4686:


[~Ormod] Thanks. Do you recall if the topic experiencing this problem was using 
compaction or not? How were you able to resolve the issue?

> Null Message payload is shutting down broker
> 
>
> Key: KAFKA-4686
> URL: https://issues.apache.org/jira/browse/KAFKA-4686
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.1.1
> Environment: Amazon Linux AMI release 2016.03 kernel 
> 4.4.19-29.55.amzn1.x86_64
>Reporter: Rodrigo Queiroz Saramago
> Fix For: 0.10.3.0
>
> Attachments: KAFKA-4686-NullMessagePayloadError.tar.xz, 
> kafkaServer.out
>
>
> Hello, I have a test environment with 3 brokers and 1 zookeeper nodes, in 
> which clients connect using two-way ssl authentication. I use kafka version 
> 0.10.1.1, the system works as expected for a while, but if the broker goes 
> down and then is restarted, something got corrupted and is not possible start 
> broker again, it always fails with the same error. What this error mean? What 
> can I do in this case? Is this the expected behavior?
> [2017-01-23 07:03:28,927] ERROR There was an error in one of the threads 
> during logs loading: kafka.common.KafkaException: Message payload is null: 
> Message(magic = 0, attributes = 1, crc = 4122289508, key = null, payload = 
> null) (kafka.log.LogManager)
> [2017-01-23 07:03:28,929] FATAL Fatal error during KafkaServer startup. 
> Prepare to shutdown (kafka.server.KafkaServer)
> kafka.common.KafkaException: Message payload is null: Message(magic = 0, 
> attributes = 1, crc = 4122289508, key = null, payload = null)
> at 
> kafka.message.ByteBufferMessageSet$$anon$1.(ByteBufferMessageSet.scala:90)
> at 
> kafka.message.ByteBufferMessageSet$.deepIterator(ByteBufferMessageSet.scala:85)
> at kafka.message.MessageAndOffset.firstOffset(MessageAndOffset.scala:33)
> at kafka.log.LogSegment.recover(LogSegment.scala:223)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:218)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:179)
> at 
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
> at 
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
> at 
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
> at kafka.log.Log.loadSegments(Log.scala:179)
> at kafka.log.Log.(Log.scala:108)
> at 
> kafka.log.LogManager$$anonfun$loadLogs$2$$anonfun$3$$anonfun$apply$10$$anonfun$apply$1.apply$mcV$sp(LogManager.scala:151)
> at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:58)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> [2017-01-23 07:03:28,946] INFO shutting down (kafka.server.KafkaServer)
> [2017-01-23 07:03:28,949] INFO Terminate ZkClient event thread. 
> (org.I0Itec.zkclient.ZkEventThread)
> [2017-01-23 07:03:28,954] INFO EventThread shut down for session: 
> 0x159bd458ae70008 (org.apache.zookeeper.ClientCnxn)
> [2017-01-23 07:03:28,954] INFO Session: 0x159bd458ae70008 closed 
> (org.apache.zookeeper.ZooKeeper)
> [2017-01-23 07:03:28,957] INFO shut down completed (kafka.server.KafkaServer)
> [2017-01-23 07:03:28,959] FATAL Fatal error during KafkaServerStartable 
> startup. Prepare to shutdown (kafka.server.KafkaServerStartable)
> kafka.common.KafkaException: Message payload is null: Message(magic = 0, 
> attributes = 1, crc = 4122289508, key = null, payload = null)
> at 
> kafka.message.ByteBufferMessageSet$$anon$1.(ByteBufferMessageSet.scala:90)
> at 
> kafka.message.ByteBufferMessageSet$.deepIterator(ByteBufferMessageSet.scala:85)
> at kafka.message.MessageAndOffset.firstOffset(MessageAndOffset.scala:33)
> at kafka.log.LogSegment.recover(LogSegment.scala:223)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:218)
> at kafka.log.Log$$anonfun$loadSegments$4.apply(Log.scala:179)
> at 
> scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
> at 
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186)
> at 
> scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
> 

[jira] [Resolved] (KAFKA-4456) Offsets of deleted topics are not removed from consumer groups

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-4456.

Resolution: Duplicate

This looks like a duplicate of KAFKA-2000, which has a patch available.

> Offsets of deleted topics are not removed from consumer groups
> --
>
> Key: KAFKA-4456
> URL: https://issues.apache.org/jira/browse/KAFKA-4456
> Project: Kafka
>  Issue Type: Bug
>Reporter: Vahid Hashemian
>Assignee: Vahid Hashemian
>
> For Java consumer based consumer groups that have been consuming from a 
> topic, if the topic is removed, the offset information of that topic is not 
> removed from the consumer group (i.e. the latest committed offset remains 
> intact).
> [KAFKA-4095|https://issues.apache.org/jira/browse/KAFKA-4095] addresses the 
> same issue for ZK based consumers.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4665) Inconsistent handling of non-existing topics in offset fetch handling

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4665:
--

 Summary: Inconsistent handling of non-existing topics in offset 
fetch handling
 Key: KAFKA-4665
 URL: https://issues.apache.org/jira/browse/KAFKA-4665
 Project: Kafka
  Issue Type: Bug
  Components: consumer, core
Reporter: Jason Gustafson
 Fix For: 0.10.3.0


For version 0 of the offset fetch API, the broker returns 
UNKNOWN_TOPIC_OR_PARTITION for any topics/partitions which do not exist at the 
time of fetching. In later versions, we skip this check. We do, however, 
continue to return UNKNOWN_TOPIC_OR_PARTITION for authorization errors (i.e. if 
the principal does not have Describe access to the corresponding topic). We 
should probably make this behavior consistent across versions.

Note also that currently the consumer raises {{KafkaException}} when it 
encounters an UNKNOWN_TOPIC_OR_PARTITION error in the offset fetch response, 
which is inconsistent with how we usually handle this error. This probably 
doesn't cause any problems currently only because of the inconsistency 
mentioned in the first paragraph above.





--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3177) Kafka consumer can hang when position() is called on a non-existing partition.

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3177:
---
Fix Version/s: (was: 0.10.2.0)
   0.10.3.0

> Kafka consumer can hang when position() is called on a non-existing partition.
> --
>
> Key: KAFKA-3177
> URL: https://issues.apache.org/jira/browse/KAFKA-3177
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.0
>Reporter: Jiangjie Qin
>Assignee: Jason Gustafson
>Priority: Critical
> Fix For: 0.10.3.0
>
>
> This can be easily reproduced as following:
> {code}
> {
> ...
> consumer.assign(SomeNonExsitingTopicParition);
> consumer.position();
> ...
> }
> {code}
> It seems when position is called we will try to do the following:
> 1. Fetch committed offsets.
> 2. If there is no committed offsets, try to reset offset using reset 
> strategy. in sendListOffsetRequest(), if the consumer does not know the 
> TopicPartition, it will refresh its metadata and retry. In this case, because 
> the partition does not exist, we fall in to the infinite loop of refreshing 
> topic metadata.
> Another orthogonal issue is that if the topic in the above code piece does 
> not exist, position() call will actually create the topic due to the fact 
> that currently topic metadata request could automatically create the topic. 
> This is a known separate issue.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4635) Client Compatibility follow-up

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4635:


[~ijuma] I think I'm convinced we should not do #6. The problem is that a 
broker on an older version could still support the newer message format, so 
returning null in that case would lead to inconsistent results between those 
brokers and newer brokers which support the time index and the new ListOffsets 
API. It seems preferable to raise an exception.

> Client Compatibility follow-up
> --
>
> Key: KAFKA-4635
> URL: https://issues.apache.org/jira/browse/KAFKA-4635
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients
>Reporter: Ismael Juma
>Assignee: Colin P. McCabe
> Fix For: 0.10.2.0
>
>
> I collected a number of improvements that I think would be good to do before 
> the release. [~cmccabe], please correct if I got anything wrong and feel free 
> to move some items to separate JIRAs.
> 1. OffsetAndTimestamp is a public class and the javadoc should only include 
> the behaviour that users will see. The following (or part of it) should 
> probably be a non-javadoc comment as it only happens internally:
> "* The timestamp should never be negative, unless it is invalid.  This could 
> happen when handling a response from a broker that doesn't support KIP-79."
> 2. There was a bit of a discussion with regards to the name of the exception 
> that is thrown when a broker is too old. The current name is 
> ObsoleteBrokerException. We should decide on the name and then we should 
> update the relevant producer/consumer methods to mention it.
> 3. [~junrao] suggested that it would be a good idea log when downgrading 
> requests as the behaviour can be a little different. We should decide the 
> right logging level and add this.
> 4. We should have a system test against 0.9.0.1 brokers. We don't support it, 
> but we should ideally give a reasonable error message.
> 5. It seems like `Fetcher.listOffset` could use `retrieveOffsetsByTimes` 
> instead of calling `sendListOffsetRequests` directly. I think that would be a 
> little better, but not sure if others disagree.
> 6. [~hachikuji] suggested that a version mismatch in the `offsetsForTimes` 
> call should result in null entry in map instead of exception for consistency 
> with how we handle the unsupported message format case. I am adding this to 
> make sure we discuss it, but I am not actually sure that is what we should 
> do. Under normal circumstances, the brokers are either too old or not whereas 
> the message format is a topic level configuration and, strictly speaking, 
> independent of the broker version (there is a correlation in practice).
> 7. We log a warning in case of an error while doing an ApiVersions request. 
> Because it is the first request and we retry, the warning in the log is 
> useful. We have a similar warning for Metadata requests, but we only did it 
> for bootstrap brokers. Would it make sense to do the same for ApiVersions?
> 8. It would be good to add a few more tests for the usable versions 
> computation. We have a single simple one at the moment.
> 9. We should add a note to the upgrade notes specifying the change in 
> behaviour with regards to older broker versions.
> cc [~hachikuji].



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-3502) Build is killed during kafka streams tests due to `pure virtual method called` error

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-3502.

   Resolution: Fixed
Fix Version/s: 0.10.2.0

Issue resolved by pull request 2381
[https://github.com/apache/kafka/pull/2381]

> Build is killed during kafka streams tests due to `pure virtual method 
> called` error
> 
>
> Key: KAFKA-3502
> URL: https://issues.apache.org/jira/browse/KAFKA-3502
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Ashish K Singh
>Assignee: Guozhang Wang
>  Labels: transient-unit-test-failure
> Fix For: 0.10.2.0
>
>
> Build failed due to failure in streams' test. Not clear which test led to 
> this.
> Jenkins console: 
> https://builds.apache.org/job/kafka-trunk-git-pr-jdk7/3210/console
> {code}
> org.apache.kafka.streams.kstream.internals.KTableFilterTest > testValueGetter 
> PASSED
> org.apache.kafka.streams.kstream.internals.KStreamFlatMapTest > testFlatMap 
> PASSED
> org.apache.kafka.streams.kstream.internals.KTableAggregateTest > testAggBasic 
> PASSED
> org.apache.kafka.streams.kstream.internals.KStreamFlatMapValuesTest > 
> testFlatMapValues PASSED
> org.apache.kafka.streams.kstream.KStreamBuilderTest > testMerge PASSED
> org.apache.kafka.streams.kstream.KStreamBuilderTest > testFrom PASSED
> org.apache.kafka.streams.kstream.KStreamBuilderTest > testNewName PASSED
> pure virtual method called
> terminate called without an active exception
> :streams:test FAILED
> FAILURE: Build failed with an exception.
> * What went wrong:
> Execution failed for task ':streams:test'.
> > Process 'Gradle Test Executor 4' finished with non-zero exit value 134
> {code}
> Tried reproducing the issue locally, but could not.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-4491) OOME when Java client can not connect to brokers

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-4491.

Resolution: Duplicate

Resolving this in favor KAFKA-4493, which has a patch available.

> OOME when Java client can not connect to brokers
> 
>
> Key: KAFKA-4491
> URL: https://issues.apache.org/jira/browse/KAFKA-4491
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.0.1
> Environment: Any (Linux|Windows)
>Reporter: Gérald Quintana
>
> Scenario: The broker cluster switched to SSL protocol but not the clients, 
> this should have raised connection failures, but instead the client dies with 
> OOME.
> Sample code to easily reproce the problem:
> {code}
> public class Main {
> private static final Logger LOGGER = LoggerFactory.getLogger(Main.class);
> public static void main(String[] args) throws InterruptedException {
> int threads = 10;
> ExecutorService executorService = 
> Executors.newFixedThreadPool(threads);
> for (int i = 0; i < threads; i++) {
> executorService.execute(new PrintConsumer("testgroup"+i, 
> "testtopic"+i));
> }
> Thread.sleep(30L);
> executorService.shutdown();
> }
> private static class PrintConsumer implements Runnable {
> private final String groupId;
> private final String topic;
> private final AtomicBoolean running = new AtomicBoolean(true);
> public PrintConsumer(String groupId, String topic) {
> this.groupId = groupId;
> this.topic = topic;
> }
> @Override
> public void run() {
> Properties props = new Properties();
> props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
> "kafka1:9093,kafka2:9093");
> props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
> props.put(ConsumerConfig.CLIENT_ID_CONFIG,"testclient");
> props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
> "org.apache.kafka.common.serialization.StringDeserializer");
> props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
> "org.apache.kafka.common.serialization.StringDeserializer");
> //props.put("ssl.truststore.location","/opt/truststore.jks");
> //props.put("ssl.truststore.password", "localhost");
> //props.put("security.protocol", "SSL");
> while (running.get()) {
> LOGGER.info("Connecting "+topic);
> try (KafkaConsumer consumer = new 
> KafkaConsumer<>(props)) {
> consumer.subscribe(Collections.singleton(topic));
> while (running.get()) {
> ConsumerRecords records = 
> consumer.poll(100);
> for (ConsumerRecord record : records)
> LOGGER.info("offset = %d, key = %s, value = 
> %s%n", record.offset(), record.key(), record.value());
> }
> } catch (Exception e) {
> LOGGER.warn("Exception in "+groupId, e);
> }
> }
> }
> public void stop() {
> running.set(false);
> }
> }
> }
> {code}
> Thrown exception:
> {code}
> java.lang.OutOfMemoryError: Java heap space
>   at java.nio.HeapByteBuffer.(HeapByteBuffer.java:57)
>   at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:93)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:154)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:135)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:323)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:283)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:260)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:360)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:224)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:192)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:163)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:179)
>   at 
> 

[jira] [Resolved] (KAFKA-4670) Kafka Consumer should validate FetchResponse

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-4670.

Resolution: Duplicate

> Kafka Consumer should validate FetchResponse
> 
>
> Key: KAFKA-4670
> URL: https://issues.apache.org/jira/browse/KAFKA-4670
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.10.2.0
>Reporter: Roger Hoover
>Assignee: Jason Gustafson
>Priority: Minor
>
> As a negative test case, I purposefully configured a bad advertised listener 
> endpoint.  
> {code}
> advertised.listeners=PLAINTEXT://www.google.com:80
> {code}
> This causes the Consumer to over-allocate and run out of memory.
> {quote}
> [2017-01-18 10:03:03,866] DEBUG Sending metadata request 
> (type=MetadataRequest, topics=foo) to node -1 
> (org.apache.kafka.clients.NetworkClient)
> [2017-01-18 10:03:03,870] DEBUG Updated cluster metadata version 2 to 
> Cluster(id = oerqPfCuTCKYUUaWdFUSVQ, nodes = [www.google.com:80 (id: 0 rack: 
> null)], partitions = [Partition(topic = foo, partition = 0, leader = 0, 
> replicas = [0], isr = [0])]) (org.apache.kafka.clients.Metadata)
> [2017-01-18 10:03:03,871] DEBUG Received group coordinator response 
> ClientResponse(receivedTimeMs=1484762583870, latencyMs=88, 
> disconnected=false, 
> requestHeader={api_key=10,api_version=0,correlation_id=0,client_id=consumer-1},
>  
> responseBody={error_code=0,coordinator={node_id=0,host=www.google.com,port=80}})
>  (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-01-18 10:03:03,871] INFO Discovered coordinator www.google.com:80 (id: 
> 2147483647 rack: null) for group console-consumer-64535. 
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-01-18 10:03:03,871] DEBUG Initiating connection to node 2147483647 at 
> www.google.com:80. (org.apache.kafka.clients.NetworkClient)
> [2017-01-18 10:03:03,915] INFO Revoking previously assigned partitions [] for 
> group console-consumer-64535 
> (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)
> [2017-01-18 10:03:03,915] INFO (Re-)joining group console-consumer-64535 
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-01-18 10:03:03,917] DEBUG Sending JoinGroup ((type: JoinGroupRequest, 
> groupId=console-consumer-64535, sessionTimeout=1, 
> rebalanceTimeout=30, memberId=, protocolType=consumer, 
> groupProtocols=org.apache.kafka.common.requests.JoinGroupRequest$ProtocolMetadata@564fabc8))
>  to coordinator www.google.com:80 (id: 2147483647 rack: null) 
> (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
> [2017-01-18 10:03:03,932] DEBUG Created socket with SO_RCVBUF = 66646, 
> SO_SNDBUF = 131874, SO_TIMEOUT = 0 to node 2147483647 
> (org.apache.kafka.common.network.Selector)
> [2017-01-18 10:03:03,932] DEBUG Completed connection to node 2147483647.  
> Fetching API versions. (org.apache.kafka.clients.NetworkClient)
> [2017-01-18 10:03:03,932] DEBUG Initiating API versions fetch from node 
> 2147483647. (org.apache.kafka.clients.NetworkClient)
> [2017-01-18 10:03:03,990] ERROR Unknown error when running consumer:  
> (kafka.tools.ConsoleConsumer$)
> java.lang.OutOfMemoryError: Java heap space
>   at java.nio.HeapByteBuffer.(HeapByteBuffer.java:57)
>   at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:93)
>   at 
> org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
>   at 
> org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:169)
>   at 
> org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:150)
>   at 
> org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:355)
>   at org.apache.kafka.common.network.Selector.poll(Selector.java:303)
>   at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:346)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:226)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:172)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:331)
>   at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:300)
>   at 
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:261)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1025)
>   at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:990)
>   at 

[jira] [Updated] (KAFKA-4547) Consumer.position returns incorrect results for Kafka 0.10.1.0 client

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4547:
---
Resolution: Fixed
Status: Resolved  (was: Patch Available)

> Consumer.position returns incorrect results for Kafka 0.10.1.0 client
> -
>
> Key: KAFKA-4547
> URL: https://issues.apache.org/jira/browse/KAFKA-4547
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.10.1.0, 0.10.0.2, 0.10.1.1
> Environment: Windows Kafka 0.10.1.0
>Reporter: Pranav Nakhe
>Assignee: Vahid Hashemian
>Priority: Blocker
>  Labels: clients
> Fix For: 0.10.2.0
>
> Attachments: issuerep.zip
>
>
> Consider the following code -
>   KafkaConsumer consumer = new 
> KafkaConsumer(props);
>   List listOfPartitions = new ArrayList();
>   for (int i = 0; i < 
> consumer.partitionsFor("IssueTopic").size(); i++) {
>   listOfPartitions.add(new TopicPartition("IssueTopic", 
> i));
>   }
>   consumer.assign(listOfPartitions);  
>   consumer.pause(listOfPartitions);
>   consumer.seekToEnd(listOfPartitions);
> //consumer.resume(listOfPartitions); -- commented out
>   for(int i = 0; i < listOfPartitions.size(); i++) {
>   
> System.out.println(consumer.position(listOfPartitions.get(i)));
>   }
>   
> I have created a topic IssueTopic with 3 partitions with a single replica on 
> my single node kafka installation (0.10.1.0)
> The behavior noticed for Kafka client 0.10.1.0 as against Kafka client 
> 0.10.0.1
> A) Initially when there are no messages on IssueTopic running the above 
> program returns
> 0.10.1.0   
> 0  
> 0  
> 0   
> 0.10.0.1
> 0
> 0
> 0
> B) Next I send 6 messages and see that the messages have been evenly 
> distributed across the three partitions. Running the above program now 
> returns 
> 0.10.1.0   
> 0  
> 0  
> 2  
> 0.10.0.1
> 2
> 2
> 2
> Clearly there is a difference in behavior for the 2 clients.
> Now after seekToEnd call if I make a call to resume (uncomment the resume 
> call in code above) then the behavior is
> 0.10.1.0   
> 2  
> 2  
> 2  
> 0.10.0.1
> 2
> 2
> 2
> This is an issue I came across when using the spark kafka integration for 
> 0.10. When I use kafka 0.10.1.0 I started seeing this issue. I had raised a 
> pull request to resolve that issue [SPARK-18779] but when looking at the 
> kafka client implementation/documentation now it seems the issue is with 
> kafka and not with spark. There does not seem to be any documentation which 
> specifies/implies that we need to call resume after seekToEnd for position to 
> return the correct value. Also there is a clear difference in the behavior in 
> the two kafka client implementations. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4676) Kafka consumers gets stuck for some partitions

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4676:


Logs from the consumer would also be helpful. Thanks.

> Kafka consumers gets stuck for some partitions
> --
>
> Key: KAFKA-4676
> URL: https://issues.apache.org/jira/browse/KAFKA-4676
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.10.1.0
>Reporter: Vishal Shukla
>Priority: Critical
>  Labels: consumer, reliability
> Attachments: stuck-topic-thread-dump.log
>
>
> We recently upgraded to Kafka 0.10.1.0. We are frequently facing issue that 
> Kafka consumers get stuck suddenly for some partitions.
> Attached thread dump.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3117) Fail test at: PlaintextConsumerTest. testAutoCommitOnRebalance

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3117:
---
Resolution: Resolved
Status: Resolved  (was: Patch Available)

Marking this resolved since it hasn't recurred (as far as I know).

> Fail test at: PlaintextConsumerTest. testAutoCommitOnRebalance 
> ---
>
> Key: KAFKA-3117
> URL: https://issues.apache.org/jira/browse/KAFKA-3117
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Affects Versions: 0.9.0.0
> Environment: oracle java764bit
> ubuntu 13.10 
>Reporter: edwardt
>Assignee: Jason Gustafson
>  Labels: newbie, test, transient-unit-test-failure
>
> java.lang.AssertionError: Expected partitions [topic-0, topic-1, topic2-0, 
> topic2-1] but actually got [topic-0, topic-1]
>   at org.junit.Assert.fail(Assert.java:88)
>   at kafka.utils.TestUtils$.waitUntilTrue(TestUtils.scala:730)
>   at 
> kafka.api.BaseConsumerTest.testAutoCommitOnRebalance(BaseConsumerTest.scala:125)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:606)
>   at 
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47)
>   at 
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
>   at 
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44)
>   at 
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
>   at 
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
>   at 
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
>   at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:70)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
>   at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238)
>   at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63)
>   at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236)
>   at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53)
>   at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:22



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-1526) Producer performance tool should have an option to enable transactions

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1526.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> Producer performance tool should have an option to enable transactions
> --
>
> Key: KAFKA-1526
> URL: https://issues.apache.org/jira/browse/KAFKA-1526
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joel Koshy
>Assignee: Raul Castro Fernandez
>  Labels: transactions
> Attachments: KAFKA-1526_2014-08-19_10:54:51.patch, KAFKA-1526.patch
>
>
> If this flag is enabled the producer could start/commit/abort transactions 
> randomly - we could add more configs/parameters for more control on 
> transaction boundaries.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1525) DumpLogSegments should print transaction IDs

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1525.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> DumpLogSegments should print transaction IDs
> 
>
> Key: KAFKA-1525
> URL: https://issues.apache.org/jira/browse/KAFKA-1525
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joel Koshy
>Assignee: Dong Lin
>  Labels: transactions
> Attachments: KAFKA-1525_2014-07-22_16:48:45.patch, 
> KAFKA-1525_2014-08-15_11:49:25.patch, KAFKA-1525.patch
>
>
> This will help in some very basic integration testing of the transactional 
> producer and brokers (i.e., until we have a transactional simple consumer).
> We only need to print the txid's. There is no need to do transactional 
> buffering.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1527) SimpleConsumer should be transaction-aware

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1527.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> SimpleConsumer should be transaction-aware
> --
>
> Key: KAFKA-1527
> URL: https://issues.apache.org/jira/browse/KAFKA-1527
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joel Koshy
>Assignee: Raul Castro Fernandez
>  Labels: transactions
> Attachments: KAFKA-1527_2014-08-19_10:39:53.patch, 
> KAFKA-1527_2014-08-19_18:22:26.patch, KAFKA-1527.patch
>
>
> This will help in further integration testing of the transactional producer. 
> This could be implemented in the consumer-iterator level or at a higher level.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1601) ConsoleConsumer/SimpleConsumerPerformance should be transaction-aware

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1601.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> ConsoleConsumer/SimpleConsumerPerformance should be transaction-aware
> -
>
> Key: KAFKA-1601
> URL: https://issues.apache.org/jira/browse/KAFKA-1601
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Dong Lin
>Assignee: Dong Lin
>Priority: Minor
>  Labels: transactions
> Attachments: KAFKA-1601_2014-08-19_21:10:12.patch, 
> KAFKA-1601_2014-08-20_08:57:29.patch, KAFKA-1601.patch
>
>
> Implement buffered consumer logic in ConsumerTransactionBuffer class.
> The class takes as input messages from non-transactional consumer (e.g. 
> ConsoleConsumer, SimpleConsumer), recognizes transaction control requests 
> (e.g. commit, abort), and outputs transaction messages when their transaction 
> is committed.
> By default, the class outputs non-transactional messages immediately on input.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1541) Add transactional request definitions to clients package

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1541.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

>  Add transactional request definitions to clients package
> -
>
> Key: KAFKA-1541
> URL: https://issues.apache.org/jira/browse/KAFKA-1541
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joel Koshy
>Assignee: Raul Castro Fernandez
>  Labels: transactions
> Attachments: KAFKA-1541.patch, KAFKA-1541.patch, KAFKA-1541.patch, 
> KAFKA-1541.patch
>
>
> Separate jira for this since KAFKA-1522 only adds definitions to the core 
> package.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1524) Implement transactional producer

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1524.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> Implement transactional producer
> 
>
> Key: KAFKA-1524
> URL: https://issues.apache.org/jira/browse/KAFKA-1524
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joel Koshy
>Assignee: Raul Castro Fernandez
>  Labels: transactions
> Attachments: KAFKA-1524_2014-08-18_09:39:34.patch, 
> KAFKA-1524_2014-08-20_09:14:59.patch, KAFKA-1524.patch, KAFKA-1524.patch, 
> KAFKA-1524.patch
>
>
> Implement the basic transactional producer functionality as outlined in 
> https://cwiki.apache.org/confluence/display/KAFKA/Transactional+Messaging+in+Kafka
> The scope of this jira is basic functionality (i.e., to be able to begin and 
> commit or abort a transaction) without the failure scenarios.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1523) Implement transaction manager module

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1523.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> Implement transaction manager module
> 
>
> Key: KAFKA-1523
> URL: https://issues.apache.org/jira/browse/KAFKA-1523
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joel Koshy
>Assignee: Dong Lin
>  Labels: transactions
> Attachments: KAFKA-1523_2014-07-17_20:12:55.patch, 
> KAFKA-1523_2014-07-22_16:45:42.patch, KAFKA-1523_2014-08-05_21:25:55.patch, 
> KAFKA-1523_2014-08-08_21:36:52.patch
>
>
> * Entry point for transaction requests
> * Appends transaction control records to the transaction journal
> * Sends transaction control records to data brokers
> * Responsible for expiring transactions
> * Supports fail-over: for which it needs to maintain a transaction HW which 
> is the offset of the BEGIN control record of the earliest pending 
> transaction. It should checkpoint the HW periodically either to ZK/separate 
> topic/offset commit.
> We merge KAFKA-1565 transaction manager failover handling into this JIRA. 
> Transaction manager should guarantee that, once a pre-commit/pre-abort 
> request is acknowledged, commit/abort request will be delivered to partitions 
> involved in the transaction.
> This patch handles the following failover scenarios:
> 1) Transaction manager or its followers fail before txRequest is duplicated 
> on local log and followers.
> Solution: Transaction manager responds to request with error status. The 
> producer keeps trying to commit.
> 2) The txPartition’s leader is not available.
> Solution: Put txRequest on unSentTxRequestQueue. When metadataCache is 
> updated, check and re-send txRequest from unSentTxRequestQueue if possible.
> 3) The txPartition’s leader fails when txRequest is in channel manager.
> Solution: Retrieve all txRequests queued for transmission to this broker and 
> put them on unSentTxRequestQueue.
> 4) Transaction manage does not receive success response from txPartition’s 
> leaders within timeout period.
> Solution: Transaction manager expires the txRequest and re-send it.
> 5) Transaction manager fails.
> Solution: The new transaction manager reads transactionHW from zookeeper, and 
> sends txRequest starting from the transactionHW.
> This patch does not provide the following feature. These will be provided in 
> separate patches.
> 1) Producer offset commit.
> 2) Transaction expiration.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1604) System Test for Transaction Management

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1604.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> System Test for Transaction Management
> --
>
> Key: KAFKA-1604
> URL: https://issues.apache.org/jira/browse/KAFKA-1604
> Project: Kafka
>  Issue Type: Test
>Reporter: Dong Lin
>Assignee: Dong Lin
>  Labels: transactions
> Attachments: KAFKA-1604_2014-08-19_17:31:16.patch, 
> KAFKA-1604_2014-08-19_21:07:35.patch
>
>
> Perform end-to-end transaction management test in the following steps:
> 1) Start Zookeeper.
> 2) Start multiple brokers.
> 3) Create topic.
> 4) Start transaction-aware ProducerPerformance to generate transactional 
> messages to topic.
> 5) Start transaction-aware ConsoleConsumer to read messages from topic.
> 6) Bounce brokers (optional).
> 7) Verify that same number of messages are sent and received.
> This patch depends on KAFKA-1524, KAFKA-1526 and KAFKA-1601.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Resolved] (KAFKA-1569) Tool for performance and correctness of transactions end-to-end

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1569.

Resolution: Unresolved

This work has been superseded by KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.

> Tool for performance and correctness of transactions end-to-end
> ---
>
> Key: KAFKA-1569
> URL: https://issues.apache.org/jira/browse/KAFKA-1569
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Raul Castro Fernandez
>Assignee: Raul Castro Fernandez
>  Labels: transactions
> Attachments: KAFKA-1569.patch, KAFKA-1569.patch
>
>
> A producer tool that creates an input file, reads it and sends it to the 
> brokers according to some transaction configuration. And a consumer tool that 
> read data from brokers with transaction boundaries and writes it to a file.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4816) Message format changes for idempotent/transactional producer

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4816:
--

 Summary: Message format changes for idempotent/transactional 
producer
 Key: KAFKA-4816
 URL: https://issues.apache.org/jira/browse/KAFKA-4816
 Project: Kafka
  Issue Type: Sub-task
Reporter: Jason Gustafson
Assignee: Jason Gustafson


This task is for the implementation of the message format changes documented 
here: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging#KIP-98-ExactlyOnceDeliveryandTransactionalMessaging-MessageFormat.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Updated] (KAFKA-4817) Implement idempotent producer

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4817:
---
Summary: Implement idempotent producer  (was: Basic idempotent producer 
implementation)

> Implement idempotent producer
> -
>
> Key: KAFKA-4817
> URL: https://issues.apache.org/jira/browse/KAFKA-4817
> Project: Kafka
>  Issue Type: Sub-task
>  Components: clients, core, producer 
>Reporter: Jason Gustafson
>Assignee: Apurva Mehta
> Fix For: 0.11.0.0
>
>
> This task covers the implementation of the idempotent producer for KIP-98. 
> This covers both the necessary changes on the server-side and client-side 
> changes.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4818) Implement transactional producer

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4818:
--

 Summary: Implement transactional producer
 Key: KAFKA-4818
 URL: https://issues.apache.org/jira/browse/KAFKA-4818
 Project: Kafka
  Issue Type: Sub-task
Reporter: Jason Gustafson
Assignee: Guozhang Wang


This covers the implementation of the transaction coordinator and the changes 
to the producer and consumer to support transactions. 



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4815) Idempotent/transactional Producer Checklist (KIP-98)

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4815:
--

 Summary: Idempotent/transactional Producer Checklist (KIP-98)
 Key: KAFKA-4815
 URL: https://issues.apache.org/jira/browse/KAFKA-4815
 Project: Kafka
  Issue Type: New Feature
  Components: clients, core, producer 
Reporter: Jason Gustafson
Assignee: Jason Gustafson
 Fix For: 0.11.0.0


This issue tracks implementation progress for KIP-98: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Created] (KAFKA-4817) Basic idempotent producer implementation

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4817:
--

 Summary: Basic idempotent producer implementation
 Key: KAFKA-4817
 URL: https://issues.apache.org/jira/browse/KAFKA-4817
 Project: Kafka
  Issue Type: Sub-task
Reporter: Jason Gustafson
Assignee: Apurva Mehta


This task covers the implementation of the idempotent producer for KIP-98. This 
covers both the necessary changes on the server-side and client-side changes.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

[jira] [Assigned] (KAFKA-3807) OffsetValidationTest - transient failure on test_broker_rolling_bounce

[Jason Gustafson (JIRA)]

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

Jason Gustafson reassigned KAFKA-3807:
--

Assignee: Jason Gustafson

> OffsetValidationTest - transient failure on test_broker_rolling_bounce
> --
>
> Key: KAFKA-3807
> URL: https://issues.apache.org/jira/browse/KAFKA-3807
> Project: Kafka
>  Issue Type: Bug
>  Components: system tests
>Reporter: Geoff Anderson
>Assignee: Jason Gustafson
>
> {code}
> test_id:
> 2016-05-28--001.kafkatest.tests.client.consumer_test.OffsetValidationTest.test_broker_rolling_bounce
> status: FAIL
> run time:   3 minutes 8.042 seconds
> Broker rolling bounce caused 2 unexpected group rebalances
> Traceback (most recent call last):
>   File 
> "/var/lib/jenkins/workspace/system-test-kafka/kafka/venv/local/lib/python2.7/site-packages/ducktape-0.5.1-py2.7.egg/ducktape/tests/runner.py",
>  line 106, in run_all_tests
> data = self.run_single_test()
>   File 
> "/var/lib/jenkins/workspace/system-test-kafka/kafka/venv/local/lib/python2.7/site-packages/ducktape-0.5.1-py2.7.egg/ducktape/tests/runner.py",
>  line 162, in run_single_test
> return self.current_test_context.function(self.current_test)
>   File 
> "/var/lib/jenkins/workspace/system-test-kafka/kafka/tests/kafkatest/tests/client/consumer_test.py",
>  line 108, in test_broker_rolling_bounce
> "Broker rolling bounce caused %d unexpected group rebalances" % 
> unexpected_rebalances
> AssertionError: Broker rolling bounce caused 2 unexpected group rebalances
> {code}
> http://confluent-kafka-system-test-results.s3-us-west-2.amazonaws.com/2016-05-28--001.1464455059--apache--trunk--7b7c4a7/report.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4108) Improve DumpLogSegments offsets-decoder output format

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4108:
--

 Summary: Improve DumpLogSegments offsets-decoder output format
 Key: KAFKA-4108
 URL: https://issues.apache.org/jira/browse/KAFKA-4108
 Project: Kafka
  Issue Type: Improvement
  Components: tools
Reporter: Jason Gustafson


When using the DumpLogSegments with the "--offsets-decoder" option (for 
consuming __consumer_offsets), the encoding of group metadata makes it a little 
difficult to identify individual fields. In particular, we use the following 
formatted string for group metadata: 
{code}
${protocolType}:${groupMetadata.protocol}:${groupMetadata.generationId}:${assignment}
{code}
Keys have a similar formatting. Most users are probably not going to know which 
field is which based only on the output, so it would be helpful to include 
field names. Maybe we could just output a JSON object?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Work started] (KAFKA-3807) OffsetValidationTest - transient failure on test_broker_rolling_bounce

[Jason Gustafson (JIRA)]

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

Work on KAFKA-3807 started by Jason Gustafson.
--
> OffsetValidationTest - transient failure on test_broker_rolling_bounce
> --
>
> Key: KAFKA-3807
> URL: https://issues.apache.org/jira/browse/KAFKA-3807
> Project: Kafka
>  Issue Type: Bug
>  Components: system tests
>Reporter: Geoff Anderson
>Assignee: Jason Gustafson
>
> {code}
> test_id:
> 2016-05-28--001.kafkatest.tests.client.consumer_test.OffsetValidationTest.test_broker_rolling_bounce
> status: FAIL
> run time:   3 minutes 8.042 seconds
> Broker rolling bounce caused 2 unexpected group rebalances
> Traceback (most recent call last):
>   File 
> "/var/lib/jenkins/workspace/system-test-kafka/kafka/venv/local/lib/python2.7/site-packages/ducktape-0.5.1-py2.7.egg/ducktape/tests/runner.py",
>  line 106, in run_all_tests
> data = self.run_single_test()
>   File 
> "/var/lib/jenkins/workspace/system-test-kafka/kafka/venv/local/lib/python2.7/site-packages/ducktape-0.5.1-py2.7.egg/ducktape/tests/runner.py",
>  line 162, in run_single_test
> return self.current_test_context.function(self.current_test)
>   File 
> "/var/lib/jenkins/workspace/system-test-kafka/kafka/tests/kafkatest/tests/client/consumer_test.py",
>  line 108, in test_broker_rolling_bounce
> "Broker rolling bounce caused %d unexpected group rebalances" % 
> unexpected_rebalances
> AssertionError: Broker rolling bounce caused 2 unexpected group rebalances
> {code}
> http://confluent-kafka-system-test-results.s3-us-west-2.amazonaws.com/2016-05-28--001.1464455059--apache--trunk--7b7c4a7/report.html



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4107) Support offset reset capability in Kafka Connect

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4107:
--

 Summary: Support offset reset capability in Kafka Connect
 Key: KAFKA-4107
 URL: https://issues.apache.org/jira/browse/KAFKA-4107
 Project: Kafka
  Issue Type: Improvement
  Components: KafkaConnect
Reporter: Jason Gustafson
Assignee: Ewen Cheslack-Postava


It would be useful in some cases to be able to reset connector offsets. For 
example, if a topic in Kafka corresponding to a source database is accidentally 
deleted (or deleted because of corrupt data), an administrator may want to 
reset offsets and reproduce the log from the beginning. It may also be useful 
to have support for overriding offsets, but that seems like a less likely use 
case.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4118) StreamsSmokeTest.test_streams started failing since 18 August build

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4118:


[~guozhang] I have a patch available for KAFKA-3807 which fixes the transient 
failures (I think). Unfortunately, this smoke test looks like it's still 
failing: 
http://testing.confluent.io/confluent-kafka-branch-builder-system-test-results/?prefix=2016-09-03--001.1472861344--hachikuji--KAFKA-3807--50c8cfb/.
 Would you mind taking a look? 

> StreamsSmokeTest.test_streams started failing since 18 August build
> ---
>
> Key: KAFKA-4118
> URL: https://issues.apache.org/jira/browse/KAFKA-4118
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Ismael Juma
>Assignee: Guozhang Wang
> Fix For: 0.10.1.0
>
>
> Link to the first failure on 18 August: 
> http://confluent-kafka-system-test-results.s3-us-west-2.amazonaws.com/2016-08-18--001.1471540190--apache--trunk--40b1dd3/report.html
> The commit corresponding to the 18 August build was 
> https://github.com/apache/kafka/commit/40b1dd3f495a59ab, which is KIP-62 (and 
> before KIP-33)
> KAFKA-3807 tracks another test that started failing at the same time and 
> there's a possibility that the PR for that JIRA fixes this one too.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4133) Provide a configuration to control consumer max in-flight fetches

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4133:
--

 Summary: Provide a configuration to control consumer max in-flight 
fetches
 Key: KAFKA-4133
 URL: https://issues.apache.org/jira/browse/KAFKA-4133
 Project: Kafka
  Issue Type: Improvement
  Components: consumer
Reporter: Jason Gustafson


With KIP-74, we now have a good way to limit the size of fetch responses, but 
it may still be difficult for users to control overall memory since the 
consumer will send fetches in parallel to all the brokers which own partitions 
that it is subscribed to. To give users finer control, it might make sense to 
add a `max.in.flight.fetches` setting to limit the total number of concurrent 
fetches at any time. This would require a KIP since it's a new configuration.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4135) Inconsistent javadoc for KafkaConsumer.poll behavior when there is no subscription

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4135:
--

 Summary: Inconsistent javadoc for KafkaConsumer.poll behavior when 
there is no subscription
 Key: KAFKA-4135
 URL: https://issues.apache.org/jira/browse/KAFKA-4135
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Reporter: Jason Gustafson
Priority: Minor


Currently, the javadoc for {{KafkaConsumer.poll}} says the following: 
"It is an error to not have subscribed to any topics or partitions before 
polling for data." However, we don't actually raise an exception if this is the 
case. Perhaps we should?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4137) Refactor multi-threaded consumer for safer network layer access

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4137:
--

 Summary: Refactor multi-threaded consumer for safer network layer 
access
 Key: KAFKA-4137
 URL: https://issues.apache.org/jira/browse/KAFKA-4137
 Project: Kafka
  Issue Type: Improvement
  Components: consumer
Reporter: Jason Gustafson


In KIP-62, we added a background thread to send heartbeats while the user is 
processing fetched data from a call to poll(). In the implementation, we 
elected to share the instance of {{NetworkClient}} between the foreground 
thread and this background thread. After working with the system test failure 
in KAFKA-3807, we've realized that this probably wasn't a good decision. It is 
very tricky to get the synchronization correct with respect to response 
callbacks and reasoning about the multi-threaded behavior is very difficult. 
For example, a common pattern is to send a request and then call 
{{NetworkClient.poll()}} to await its return. With another thread also 
potentially calling poll(), the response can actually return before the sending 
thread itself invokes poll(). This can cause unnecessary (and potentially 
unbounded) blocking, and avoiding it is quite complex. 

A different approach we've discussed would be to use two instances of 
NetworkClient, one dedicated to fetching, and one dedicated to coordinator 
communication. The fetching NetworkClient can continue to work exclusively in 
the foreground thread and we can confine the coordinator NetworkClient to the 
background thread. This provides much better isolation and avoids all of the 
race conditions with calling poll() from two threads. The main complication is 
in how to expose blocking APIs to interact with the background thread. For 
example, in the current consumer API, rebalance are completed in the foreground 
thread, so we would need to coordinate with the background thread to preserve 
this (e.g. by using a Future abstraction).



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-4106) Consumer / add configure method to PartitionAssignor interface

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-4106.

Resolution: Resolved
  Assignee: Jason Gustafson

The requested functionality already exists.

> Consumer / add configure method to PartitionAssignor interface
> --
>
> Key: KAFKA-4106
> URL: https://issues.apache.org/jira/browse/KAFKA-4106
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, consumer
>Affects Versions: 0.10.0.1
>Reporter: Florian Hussonnois
>Assignee: Jason Gustafson
>Priority: Minor
>
> Currently, we can implement a custom PartitionAssignor which will forward 
> user data that will be used during the assignments protocol. For example, 
> data can be used to implement a rack-aware assignor
> However, currently we cannot dynamically configure a PartitionAssignor 
> instance.
> It would be nice to add a method configure(Map PartitionAssignor interface. Then, this method will be invoked by the 
> KafkaConsumer  on each assignor, as this is do for deserializers.
> The code modifications are pretty straight-forward but involve modifying the 
> public interface PartitionAssignor. Does that mean this JIRA needs a KIP ?
> I can contribute to that improvement.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4098) NetworkClient should not intercept all metadata requests on disconnect

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4098:
--

 Summary: NetworkClient should not intercept all metadata requests 
on disconnect 
 Key: KAFKA-4098
 URL: https://issues.apache.org/jira/browse/KAFKA-4098
 Project: Kafka
  Issue Type: Bug
Reporter: Jason Gustafson
Assignee: Jason Gustafson


It looks like we're missing a check in 
{{DefaultMetadataUpdater.maybeHandleDisconnecttion}} that the request was 
initiated by {{NetworkClient}}. We should do the same thing we do in 
{{maybeHandleCompletedReceive}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4101) java.lang.IllegalStateException in org.apache.kafka.common.network.Selector.channelOrFail

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4101:


[~andrey-sra] I think this might be fixed by KAFKA-3341 (in 0.10.0.0 
unfortunately). Was there any additional context in the log? I'm looking for an 
error messaging saying something like "Closing socket for {host} because of 
error."

> java.lang.IllegalStateException in 
> org.apache.kafka.common.network.Selector.channelOrFail
> -
>
> Key: KAFKA-4101
> URL: https://issues.apache.org/jira/browse/KAFKA-4101
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.1
> Environment: Ubuntu 14.04, AWS deployment, under heavy network load
>Reporter: Andrey Savov
>
> {code}
>  at org.apache.kafka.common.network.Selector.channelOrFail(Selector.java:467)
> at org.apache.kafka.common.network.Selector.mute(Selector.java:347)
> at 
> kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:434)
> at 
> kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:421)
> at scala.collection.Iterator$class.foreach(Iterator.scala:742)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at kafka.network.Processor.run(SocketServer.scala:421)
> at java.lang.Thread.run(Thread.java:745)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Comment Edited] (KAFKA-4101) java.lang.IllegalStateException in org.apache.kafka.common.network.Selector.channelOrFail

[Jason Gustafson (JIRA)]

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

Jason Gustafson edited comment on KAFKA-4101 at 8/29/16 11:03 PM:
--

[~andrey-sra] I think this might be fixed by KAFKA-3341 (in 0.10.0.0 
unfortunately). Was there any additional context in the log? I'm looking for an 
error messaging saying something like "Closing socket for [host] because of 
error."


was (Author: hachikuji):
[~andrey-sra] I think this might be fixed by KAFKA-3341 (in 0.10.0.0 
unfortunately). Was there any additional context in the log? I'm looking for an 
error messaging saying something like "Closing socket for {host} because of 
error."

> java.lang.IllegalStateException in 
> org.apache.kafka.common.network.Selector.channelOrFail
> -
>
> Key: KAFKA-4101
> URL: https://issues.apache.org/jira/browse/KAFKA-4101
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.1
> Environment: Ubuntu 14.04, AWS deployment, under heavy network load
>Reporter: Andrey Savov
>
> {code}
>  at org.apache.kafka.common.network.Selector.channelOrFail(Selector.java:467)
> at org.apache.kafka.common.network.Selector.mute(Selector.java:347)
> at 
> kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:434)
> at 
> kafka.network.Processor$$anonfun$run$11.apply(SocketServer.scala:421)
> at scala.collection.Iterator$class.foreach(Iterator.scala:742)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at kafka.network.Processor.run(SocketServer.scala:421)
> at java.lang.Thread.run(Thread.java:745)
> {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4081) Consumer API consumer new interface commitSyn does not verify the validity of offset

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4081:


I think it definitely makes sense to reject negative values (we currently use 
-1 to indicate an invalid offset which leads to inconsistent behavior as noted 
by [~mimaison]), but rejecting offsets greater than hw seems more challenging 
since neither the client (nor the server accepting the offset commit) will 
necessarily have an up-to-date value. We could use the last hw returned from a 
previous fetch, but it might be stale by the time the user attempts to commit 
offsets. Perhaps it would make more sense to reject _any_ offset which is 
greater than the current position? It would still be possible to commit an 
invalid offset, but not without an explicit seek to that offset. However, there 
is a compatibility concern for use cases which use the consumer only for access 
to the offset API, which could be seen in offset tooling.

> Consumer API consumer new interface commitSyn does not verify the validity of 
> offset
> 
>
> Key: KAFKA-4081
> URL: https://issues.apache.org/jira/browse/KAFKA-4081
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Affects Versions: 0.9.0.1
>Reporter: lifeng
>
> Consumer API consumer new interface commitSyn synchronization update offset, 
> for the illegal offset successful return, illegal offset<0 or offset>hw



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4106) Consumer / add configure method to PartitionAssignor interface

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4106:


[~fhussonnois] I think you might be able to do this already if you have your 
{{PartitionAssignor}} implement {{org.apache.kafka.common.Configurable}}.

> Consumer / add configure method to PartitionAssignor interface
> --
>
> Key: KAFKA-4106
> URL: https://issues.apache.org/jira/browse/KAFKA-4106
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, consumer
>Affects Versions: 0.10.0.1
>Reporter: Florian Hussonnois
>Priority: Minor
>
> Currently, we can implement a custom PartitionAssignor which will forward 
> user data that will be used during the assignments protocol. For example, 
> data can be used to implement a rack-aware assignor
> However, currently we cannot dynamically configure a PartitionAssignor 
> instance.
> It would be nice to add a method configure(Map PartitionAssignor interface. Then, this method will be invoked by the 
> KafkaConsumer  on each assignor, as this is do for deserializers.
> The code modifications are pretty straight-forward but involve modifying the 
> public interface PartitionAssignor. Does that mean this JIRA needs a KIP ?
> I can contribute to that improvement.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Assigned] (KAFKA-4103) DumpLogSegments cannot print data from offsets topic

[Jason Gustafson (JIRA)]

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

Jason Gustafson reassigned KAFKA-4103:
--

Assignee: Jason Gustafson

> DumpLogSegments cannot print data from offsets topic
> 
>
> Key: KAFKA-4103
> URL: https://issues.apache.org/jira/browse/KAFKA-4103
> Project: Kafka
>  Issue Type: Bug
>  Components: tools
>Reporter: Ewen Cheslack-Postava
>Assignee: Jason Gustafson
>Priority: Blocker
> Fix For: 0.10.1.0
>
>
> It looks like there's been a regression in the DumpLogSegments tool. I'm 
> marking it a blocker since it appears we can no longer dump offset 
> information from this tool, which makes it really hard to debug anything 
> related to __consumer_offsets.
> The 0.10.0 branch seems to work fine, but even with offsets log files 
> generated using only old formats (0.10.0 branch), the DumpLogSegments tool 
> from trunk (i.e. 0.10.1.0-SNAPSHOT with latest githash 
> b91eeac9438b8718c410045b0e9191296ebb536d as of reporting this) will cause the 
> exception below. This was found while doing some basic testing of KAFKA-4062.
> {quote}
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> offset: 0 position: 0 CreateTime: 1472615183913 isvalid: true payloadsize: 
> 199 magic: 1 compresscodec: NoCompressionCodec crc: 2036280914 keysize: 
> 26Exception in thread "main" java.util.IllegalFormatConversionException: x != 
> scala.math.BigInt
>   at 
> java.util.Formatter$FormatSpecifier.failConversion(Formatter.java:4045)
>   at 
> java.util.Formatter$FormatSpecifier.printInteger(Formatter.java:2748)
>   at 
> java.util.Formatter$FormatSpecifier.print(Formatter.java:2702)
>   at java.util.Formatter.format(Formatter.java:2488)
>   at java.util.Formatter.format(Formatter.java:2423)
>   at java.lang.String.format(String.java:2792)
>   at 
> kafka.tools.DumpLogSegments$OffsetsMessageParser.kafka$tools$DumpLogSegments$OffsetsMessageParser$$hex(DumpLogSegments.scala:240)
>   at 
> kafka.tools.DumpLogSegments$OffsetsMessageParser$$anonfun$3.apply(DumpLogSegments.scala:272)
>   at 
> kafka.tools.DumpLogSegments$OffsetsMessageParser$$anonfun$3.apply(DumpLogSegments.scala:262)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
>   at scala.collection.immutable.List.foreach(List.scala:318)
>   at 
> scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
>   at 
> scala.collection.AbstractTraversable.map(Traversable.scala:105)
>   at 
> kafka.tools.DumpLogSegments$OffsetsMessageParser.parseGroupMetadata(DumpLogSegments.scala:262)
>   at 
> kafka.tools.DumpLogSegments$OffsetsMessageParser.parse(DumpLogSegments.scala:290)
>   at 
> kafka.tools.DumpLogSegments$$anonfun$kafka$tools$DumpLogSegments$$dumpLog$1$$anonfun$apply$3.apply(DumpLogSegments.scala:332)
>   at 
> kafka.tools.DumpLogSegments$$anonfun$kafka$tools$DumpLogSegments$$dumpLog$1$$anonfun$apply$3.apply(DumpLogSegments.scala:312)
>   at scala.collection.Iterator$class.foreach(Iterator.scala:727)
>   at 
> kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:30)
>   at 
> kafka.tools.DumpLogSegments$$anonfun$kafka$tools$DumpLogSegments$$dumpLog$1.apply(DumpLogSegments.scala:312)
>   at 
> kafka.tools.DumpLogSegments$$anonfun$kafka$tools$DumpLogSegments$$dumpLog$1.apply(DumpLogSegments.scala:310)
>   at scala.collection.Iterator$class.foreach(Iterator.scala:727)
>   at 
> kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:30)
>   at 
> kafka.tools.DumpLogSegments$.kafka$tools$DumpLogSegments$$dumpLog(DumpLogSegments.scala:310)
>   at 
> kafka.tools.DumpLogSegments$$anonfun$main$1.apply(DumpLogSegments.scala:96)
>   at 
> kafka.tools.DumpLogSegments$$anonfun$main$1.apply(DumpLogSegments.scala:92)
>   at 
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
>   at 
> scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
>   at kafka.tools.DumpLogSegments$.main(DumpLogSegments.scala:92)
>   at kafka.tools.DumpLogSegments.main(DumpLogSegments.scala)
> {quote}
> I haven't really dug in, but the source of the error is confusing since the 
> relevant string formatting code doesn't seem to have changed anytime 
> recently. It seems it might be related to changes in the group metadata code. 
> I did the git bisect and 

[jira] [Updated] (KAFKA-4139) Kafka consumer stuck in ensureCoordinatorReady after broker failure

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4139:
---
Resolution: Fixed
Status: Resolved  (was: Patch Available)

> Kafka consumer stuck in ensureCoordinatorReady after broker failure
> ---
>
> Key: KAFKA-4139
> URL: https://issues.apache.org/jira/browse/KAFKA-4139
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.10.0.1
>Reporter: Rajini Sivaram
>Assignee: Rajini Sivaram
>
> In one of our tests with a single broker, consumer is stuck waiting to find 
> coordinator after the broker is restarted. {{findCoordinatorFuture}} is never 
> reset if {{sendGroupCoordinatorRequest()}} returns 
> {{RequestFuture.noBrokersAvailable()}}.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-4172) Fix masked test error in KafkaConsumerTest.testSubscriptionChangesWithAutoCommitEnabled

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-4172.

Resolution: Fixed

Issue resolved by pull request 1857
[https://github.com/apache/kafka/pull/1857]

> Fix masked test error in 
> KafkaConsumerTest.testSubscriptionChangesWithAutoCommitEnabled
> ---
>
> Key: KAFKA-4172
> URL: https://issues.apache.org/jira/browse/KAFKA-4172
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer, unit tests
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
> Fix For: 0.10.1.0
>
>
> This test case has an incorrectly matched mock fetch response, which was 
> silently raising an NPE, which was caught in NetworkClient. In general, aside 
> from fixing the test, we are probably missing a null check in the 
> FetchRespose to verify that the partitions included in the fetch response 
> were actually requested. This is usually not a problem because the broker 
> doesn't send us invalid fetch responses, but it's probably better to be a 
> little more defensive when handling responses.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4172) Fix masked test error in KafkaConsumerTest.testSubscriptionChangesWithAutoCommitEnabled

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4172:
--

 Summary: Fix masked test error in 
KafkaConsumerTest.testSubscriptionChangesWithAutoCommitEnabled
 Key: KAFKA-4172
 URL: https://issues.apache.org/jira/browse/KAFKA-4172
 Project: Kafka
  Issue Type: Bug
  Components: consumer, unit tests
Reporter: Jason Gustafson
Assignee: Jason Gustafson
 Fix For: 0.10.1.0


This test case has an incorrectly matched mock fetch response, which was 
silently raising an NPE, which was caught in NetworkClient. In general, aside 
from fixing the test, we are probably missing a null check in the FetchRespose 
to verify that the partitions included in the fetch response were actually 
requested. This is usually not a problem because the broker doesn't send us 
invalid fetch responses, but it's probably better to be a little more defensive 
when handling responses.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4145) Avoid redundant integration testing in ProducerSendTests

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4145:
--

 Summary: Avoid redundant integration testing in ProducerSendTests
 Key: KAFKA-4145
 URL: https://issues.apache.org/jira/browse/KAFKA-4145
 Project: Kafka
  Issue Type: Improvement
  Components: unit tests
Reporter: Jason Gustafson


We have a few test cases in {{BaseProducerSendTest}} which probably have little 
value being tested for both Plaintext and SSL. We can move them to 
{{PlaintextProducerSendTest}} and save a little bit on the build time. The 
following tests seem like possible candidates:

1. testSendCompressedMessageWithCreateTime
2. testSendNonCompressedMessageWithCreateTime
3. testSendCompressedMessageWithLogAppendTime
4. testSendNonCompressedMessageWithLogApendTime
5. testAutoCreateTopic
6. testFlush
7. testSendWithInvalidCreateTime
8. testCloseWithZeroTimeoutFromCallerThread
9. testCloseWithZeroTimeoutFromSenderThread



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-4133) Provide a configuration to control consumer max in-flight fetches

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-4133:


[~mimaison] You should have permission now.

> Provide a configuration to control consumer max in-flight fetches
> -
>
> Key: KAFKA-4133
> URL: https://issues.apache.org/jira/browse/KAFKA-4133
> Project: Kafka
>  Issue Type: Improvement
>  Components: consumer
>Reporter: Jason Gustafson
>Assignee: Mickael Maison
>
> With KIP-74, we now have a good way to limit the size of fetch responses, but 
> it may still be difficult for users to control overall memory since the 
> consumer will send fetches in parallel to all the brokers which own 
> partitions that it is subscribed to. To give users finer control, it might 
> make sense to add a `max.in.flight.fetches` setting to limit the total number 
> of concurrent fetches at any time. This would require a KIP since it's a new 
> configuration.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4147) Transient failure in ConsumerCoordinatorTest.testAutoCommitDynamicAssignment

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4147:
--

 Summary: Transient failure in 
ConsumerCoordinatorTest.testAutoCommitDynamicAssignment
 Key: KAFKA-4147
 URL: https://issues.apache.org/jira/browse/KAFKA-4147
 Project: Kafka
  Issue Type: Sub-task
  Components: unit tests
Reporter: Jason Gustafson
Assignee: Jason Gustafson


Seen recently: 
https://jenkins.confluent.io/job/kafka-trunk/1143/testReport/org.apache.kafka.clients.consumer.internals/ConsumerCoordinatorTest/testAutoCommitDynamicAssignment/.

{code}
java.lang.NullPointerException
at 
org.apache.kafka.clients.consumer.internals.ConsumerCoordinatorTest.testAutoCommitDynamicAssignment(ConsumerCoordinatorTest.java:821)
{code}

Looks like it's caused by a race condition with the heartbeat thread.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-4033) KIP-70: Revise Partition Assignment Semantics on New Consumer's Subscription Change

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4033:
---
   Resolution: Fixed
Fix Version/s: 0.10.1.0
   Status: Resolved  (was: Patch Available)

Issue resolved by pull request 1726
[https://github.com/apache/kafka/pull/1726]

> KIP-70: Revise Partition Assignment Semantics on New Consumer's Subscription 
> Change
> ---
>
> Key: KAFKA-4033
> URL: https://issues.apache.org/jira/browse/KAFKA-4033
> Project: Kafka
>  Issue Type: Bug
>Reporter: Vahid Hashemian
>Assignee: Vahid Hashemian
> Fix For: 0.10.1.0
>
>
> Modify the new consumer's implementation of topics subscribe and unsubscribe 
> interfaces so that they do not cause an immediate assignment update (this is 
> how the regex subscribe interface is implemented). Instead, the assignment 
> remains valid until it has been revoked in the next rebalance.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3333) Client Partitioner - Round Robin

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-:


Admittedly, the line may not be totally clear, but we have asked for KIPs for 
similar components on the consumer (see KIP-54 for example). In general, it's a 
good way to publicize the feature in order to gauge community interest and 
collect feedback. We've hesitated in the past to include additional 
partitioners because they can always be made available to users through a 
separate project. In this case, there might be some concern over the fact that 
a key can be sent to multiple partitions which breaks the usual binding of key 
and partition, but perhaps it makes sense for some use cases.

> Client Partitioner - Round Robin
> 
>
> Key: KAFKA-
> URL: https://issues.apache.org/jira/browse/KAFKA-
> Project: Kafka
>  Issue Type: New Feature
>  Components: clients
>Affects Versions: 0.10.0.0
>Reporter: Stephen Powis
> Fix For: 0.10.2.0
>
>   Original Estimate: 1h
>  Remaining Estimate: 1h
>
> The 
> [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java]
>  typically distributes using the hash of the keybytes, and falls back to 
> round robin if there is no key.  But there is currently no way to do Round 
> Robin partitioning if you have keys on your messages without writing your own 
> partitioning implementation.
> I think it'd be helpful to have an implementation of straight Round Robin 
> partitioning included with the library.  



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4160) Consumer onPartitionsRevoked should not be invoked while holding the coordinator lock

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4160:
--

 Summary: Consumer onPartitionsRevoked should not be invoked while 
holding the coordinator lock
 Key: KAFKA-4160
 URL: https://issues.apache.org/jira/browse/KAFKA-4160
 Project: Kafka
  Issue Type: Bug
  Components: consumer
Reporter: Jason Gustafson
Assignee: Jason Gustafson
Priority: Blocker
 Fix For: 0.10.1.0


We have a single lock which is used for protecting access to shared coordinator 
state between the foreground thread and the background heartbeat thread. 
Currently, the onPartitionsRevoked callback is invoked while holding this lock, 
which prevents the heartbeat thread from sending any heartbeats. If the 
heartbeat thread is blocked for longer than the session timeout, than the 
consumer is kicked out of the group. Typically this is not a problem because 
onPartitionsRevoked might only commit offsets, but for Kafka Streams, there is 
some expensive cleanup logic which can take longer than the session timeout.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-2307) Drop ConsumerOffsetChecker completely

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-2307:


Sounds good. I'll kick this to the next release then.

> Drop ConsumerOffsetChecker completely
> -
>
> Key: KAFKA-2307
> URL: https://issues.apache.org/jira/browse/KAFKA-2307
> Project: Kafka
>  Issue Type: Sub-task
>  Components: tools
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.2.0
>
>
> ConsumerOffsetChecker has been replaced by ConsumerGroupCommand and is 
> deprecated in 0.9.0. Should be dropped in 0.9.1.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-2307) Drop ConsumerOffsetChecker completely

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2307:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Drop ConsumerOffsetChecker completely
> -
>
> Key: KAFKA-2307
> URL: https://issues.apache.org/jira/browse/KAFKA-2307
> Project: Kafka
>  Issue Type: Sub-task
>  Components: tools
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.2.0
>
>
> ConsumerOffsetChecker has been replaced by ConsumerGroupCommand and is 
> deprecated in 0.9.0. Should be dropped in 0.9.1.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-740) Improve crash-safety of log segment swap

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-740:
--
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Improve crash-safety of log segment swap
> 
>
> Key: KAFKA-740
> URL: https://issues.apache.org/jira/browse/KAFKA-740
> Project: Kafka
>  Issue Type: Bug
>  Components: log
>Reporter: Jay Kreps
>Assignee: Jay Kreps
> Fix For: 0.10.2.0
>
>
> Currently Log.replaceSegments has a bug that can cause a swap containing 
> multiple segments to partially complete. This would lead to duplicate data in 
> the log.
> The proposed fix is to use a name like offset1_and_offset2.swap for a segment 
> meant to replace segments with base offsets offset1 and offset2.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3073) KafkaConnect should support regular expression for topics

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3073:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> KafkaConnect should support regular expression for topics
> -
>
> Key: KAFKA-3073
> URL: https://issues.apache.org/jira/browse/KAFKA-3073
> Project: Kafka
>  Issue Type: Improvement
>  Components: KafkaConnect
>Reporter: Gwen Shapira
>Assignee: Liquan Pei
>  Labels: needs-kip
> Fix For: 0.10.2.0
>
>
> KafkaConsumer supports both a list of topics or a pattern when subscribing. 
> KafkaConnect only supports a list of topics, which is not just more of a 
> hassle to configure - it also requires more maintenance.
> I suggest adding topics.pattern as a new configuration and letting users 
> choose between 'topics' or 'topics.pattern'.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-2045) Memory Management on the consumer

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2045:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Memory Management on the consumer
> -
>
> Key: KAFKA-2045
> URL: https://issues.apache.org/jira/browse/KAFKA-2045
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Guozhang Wang
> Fix For: 0.10.2.0
>
>
> We need to add the memory management on the new consumer like we did in the 
> new producer. This would probably include:
> 1. byte buffer re-usage for fetch response partition data.
> 2. byte buffer re-usage for on-the-fly de-compression.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3478) Finer Stream Flow Control

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3478:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Finer Stream Flow Control
> -
>
> Key: KAFKA-3478
> URL: https://issues.apache.org/jira/browse/KAFKA-3478
> Project: Kafka
>  Issue Type: Sub-task
>  Components: streams
>Reporter: Guozhang Wang
>  Labels: user-experience
> Fix For: 0.10.2.0
>
>
> Today we have a event-time based flow control mechanism in order to 
> synchronize multiple input streams in a best effort manner:
> http://docs.confluent.io/3.0.0/streams/architecture.html#flow-control-with-timestamps
> However, there are some use cases where users would like to have finer 
> control of the input streams, for example, with two input streams, one of 
> them always reading from offset 0 upon (re)-starting, and the other reading 
> for log end offset.
> Today we only have one consumer config "offset.auto.reset" to control that 
> behavior, which means all streams are read either from "earliest" or "latest".
> We should consider how to improve this settings to allow users have finer 
> control over these frameworks.
> =
> A finer flow control could also be used to allow for populating a {{KTable}} 
> (with an "initial" state) before starting the actual processing (this feature 
> was ask for in the mailing list multiple times already). Even if it is quite 
> hard to define, *when* the initial populating phase should end, this might 
> still be useful. There would be the following possibilities:
>  1) an initial fixed time period for populating
>(it might be hard for a user to estimate the correct value)
>  2) an "idle" period, ie, if no update to a KTable for a certain time is
> done, we consider it as populated
>  3) a timestamp cut off point, ie, all records with an older timestamp
> belong to the initial populating phase
>  4) a throughput threshold, ie, if the populating frequency falls below
> the threshold, the KTable is considered "finished"
>  5) maybe something else ??
> The API might look something like this
> {noformat}
> KTable table = builder.table("topic", 1000); // populate the table without 
> reading any other topics until see one record with timestamp 1000.
> {noformat}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-2000) Delete consumer offsets from kafka once the topic is deleted

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2000:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Delete consumer offsets from kafka once the topic is deleted
> 
>
> Key: KAFKA-2000
> URL: https://issues.apache.org/jira/browse/KAFKA-2000
> Project: Kafka
>  Issue Type: Bug
>Reporter: Sriharsha Chintalapani
>Assignee: Sriharsha Chintalapani
>  Labels: newbie++
> Fix For: 0.10.2.0
>
> Attachments: KAFKA-2000.patch, KAFKA-2000_2015-05-03_10:39:11.patch
>
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Created] (KAFKA-4155) Remove WorkerGroupMember from Connect

[Jason Gustafson (JIRA)]

Jason Gustafson created KAFKA-4155:
--

 Summary: Remove WorkerGroupMember from Connect
 Key: KAFKA-4155
 URL: https://issues.apache.org/jira/browse/KAFKA-4155
 Project: Kafka
  Issue Type: Improvement
  Components: KafkaConnect
Reporter: Jason Gustafson
Assignee: Ewen Cheslack-Postava


After a few refactors, it doesn't seem like {{WorkerGroupMember}} is bringing 
much to the table anymore since it does little more than delegate to 
{{WorkerCoordinator}}. Maybe we ought to just remove it?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3543) Allow a variant of transform() which can emit multiple values

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3543:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Allow a variant of transform() which can emit multiple values
> -
>
> Key: KAFKA-3543
> URL: https://issues.apache.org/jira/browse/KAFKA-3543
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Greg Fodor
>  Labels: api
> Fix For: 0.10.2.0
>
>
> Right now it seems that if you want to apply an arbitrary stateful 
> transformation to a stream, you either have to use a TransformerSupplier or 
> ProcessorSupplier sent to transform() or process(). The custom processor will 
> allow you to emit multiple new values, but the process() method currently 
> terminates that branch of the topology so you can't apply additional data 
> flow. transform() lets you continue the data flow, but forces you to emit a 
> single value for every input value.
> (It actually doesn't quite force you to do this, since you can hold onto the 
> ProcessorContext and emit multiple, but that's probably not the ideal way to 
> do it :))
> It seems desirable to somehow allow a transformation that emits multiple 
> values per input value. I'm not sure of the best way to factor this inside of 
> the current TransformerSupplier/Transformer architecture in a way that is 
> clean and efficient -- currently I'm doing the workaround above of just 
> calling forward() myself on the context and actually emitting dummy values 
> which are filtered out downstream.
> -
> It is worth considering adding a new flatTransofrm function as 
> {code}
>  KStream transform(TransformerSupplier Iterable>> transformerSupplier, String... stateStoreNames)
> {code}
> which is essentially the same as
> {code} transform().flatMap() {code}



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-2311) Consumer's ensureNotClosed method not thread safe

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2311:
---
Resolution: Fixed
Status: Resolved  (was: Patch Available)

> Consumer's ensureNotClosed method not thread safe
> -
>
> Key: KAFKA-2311
> URL: https://issues.apache.org/jira/browse/KAFKA-2311
> Project: Kafka
>  Issue Type: Bug
>  Components: clients
>Reporter: Tim Brooks
>Assignee: Tim Brooks
> Fix For: 0.10.1.0
>
> Attachments: KAFKA-2311.patch, KAFKA-2311.patch
>
>
> When a call is to the consumer is made, the first check is to see that the 
> consumer is not closed. This variable is not volatile so there is no 
> guarantee previous stores will be visible before a read.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3590) KafkaConsumer fails with "Messages are rejected since there are fewer in-sync replicas than required." when polling

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3590:
---
Fix Version/s: 0.10.1.0

> KafkaConsumer fails with "Messages are rejected since there are fewer in-sync 
> replicas than required." when polling
> ---
>
> Key: KAFKA-3590
> URL: https://issues.apache.org/jira/browse/KAFKA-3590
> Project: Kafka
>  Issue Type: Bug
>  Components: consumer
>Affects Versions: 0.9.0.1
> Environment: JDK1.8 Ubuntu 14.04
>Reporter: Sergey Alaev
>Assignee: Jason Gustafson
> Fix For: 0.10.1.0
>
>
> KafkaConsumer.poll() fails with "Messages are rejected since there are fewer 
> in-sync replicas than required.". Isn't this message about minimum number of 
> ISR's when *sending* messages?
> Stacktrace:
> org.apache.kafka.common.KafkaException: Unexpected error from SyncGroup: 
> Messages are rejected since there are fewer in-sync replicas than required.
> at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$SyncGroupRequestHandler.handle(AbstractCoordinator.java:444)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$SyncGroupRequestHandler.handle(AbstractCoordinator.java:411)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:665)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:644)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:167)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.onComplete(ConsumerNetworkClient.java:380)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:274) 
> ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:163)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:222)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.ensurePartitionAssignment(ConsumerCoordinator.java:311)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:890)
>  ~[kafka-clients-0.9.0.1.jar:na]
> at 
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853) 
> ~[kafka-clients-0.9.0.1.jar:na]



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1617) Move Metadata Cache to TopicManager and handling of Offset Request to LogManager

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1617:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Move Metadata Cache to TopicManager and handling of Offset Request to 
> LogManager
> 
>
> Key: KAFKA-1617
> URL: https://issues.apache.org/jira/browse/KAFKA-1617
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
> Fix For: 0.10.2.0
>
>
> This is a follow-up of KAFKA-1583. In order to make Kafka APIs a pure 
> stateless layer that just forwards different requests to the corresponding 
> managers, there are still two tasks left:
> 1. Move the metadata cache at KafkaApis to a separate manager, maybe called 
> TopicManager, which will be responsible for a) handle topic metadata request, 
> b) handle topic metadata update request by talking to replica manager if 
> necessary.
> 2. Move the handling logic of offset request to the LogManager, which should 
> contain all the information necessary to handle this request.
> Finally, the KafkaApis class should be stateless, meaning no inner variables 
> and no start()/shutdown() functions needed.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-4005) Add per topic compression ratio in broker and consumer.

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-4005:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add per topic compression ratio in broker and consumer.
> ---
>
> Key: KAFKA-4005
> URL: https://issues.apache.org/jira/browse/KAFKA-4005
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients, core
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.2.0
>
>
> Currently we only have compression ratio metric on the producer side. It 
> would be very useful to have that in the brokers and consumers for each topic 
> as well. 
> On the broker side, the compression ratio can be potentially depending on the 
> produce request or the final message written to the disk. It is probably more 
> useful to use the compression ratio of the messages we finally writes to 
> disk. If the messages are compressed in different compression type, we can 
> maintain the compression ratio for each type separately.
> The consumer side compression ratio would be for each topic and compression 
> type combination as well.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-334) Some tests fail when building on a Windows box

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-334.
---
   Resolution: Won't Fix
Fix Version/s: (was: 0.10.1.0)

Resolving this since it's been open forever and it seems there are no plans to 
fix it, especially since we have changed build systems.

> Some tests fail when building on a Windows box
> --
>
> Key: KAFKA-334
> URL: https://issues.apache.org/jira/browse/KAFKA-334
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.7
> Environment: Windows 7 - reproduces under command shell, cygwin, and 
> MINGW32 (Git Bash)
>Reporter: Roman Garcia
>Priority: Minor
>  Labels: build-failure, test-fail
>
> Trying to create a ZIP distro from sources failed.
> On Win7. On cygwin, command shell and git bash.
> Tried with incubator-src download from ASF download page, as well as fresh 
> checkout from latest trunk (r1329547).
> Once I tried the same on a Linux box, everything was working ok.
> svn co http://svn.apache.org/repos/asf/incubator/kafka/trunk kafka-0.7.0
> ./sbt update (OK)
> ./sbt package (OK)
> ./sbt release-zip (FAIL)
> Tests failing:
> [error] Error running kafka.integration.LazyInitProducerTest: Test FAILED
> [error] Error running kafka.zk.ZKLoadBalanceTest: Test FAILED
> [error] Error running kafka.javaapi.producer.ProducerTest: Test FAILED
> [error] Error running kafka.producer.ProducerTest: Test FAILED
> [error] Error running test: One or more subtasks failed
> [error] Error running doc: Scaladoc generation failed
> Stacks:
> [error] Test Failed: testZKSendWithDeadBroker
> junit.framework.AssertionFailedError: Message set should have another message
> at junit.framework.Assert.fail(Assert.java:47)
> at junit.framework.Assert.assertTrue(Assert.java:20)
> at 
> kafka.javaapi.producer.ProducerTest.testZKSendWithDeadBroker(ProducerTest.scala:448)
> [error] Test Failed: testZKSendToNewTopic
> junit.framework.AssertionFailedError: Message set should have 1 message
> at junit.framework.Assert.fail(Assert.java:47)
> at junit.framework.Assert.assertTrue(Assert.java:20)
> at 
> kafka.javaapi.producer.ProducerTest.testZKSendToNewTopic(ProducerTest.scala:416)
> [error] Test Failed: testLoadBalance(kafka.zk.ZKLoadBalanceTest)
> junit.framework.AssertionFailedError: expected:<5> but was:<0>
> at junit.framework.Assert.fail(Assert.java:47)
> at junit.framework.Assert.failNotEquals(Assert.java:277)
> at junit.framework.Assert.assertEquals(Assert.java:64)
> at junit.framework.Assert.assertEquals(Assert.java:195)
> at junit.framework.Assert.assertEquals(Assert.java:201)
> at 
> kafka.zk.ZKLoadBalanceTest.checkSetEqual(ZKLoadBalanceTest.scala:121)
> at 
> kafka.zk.ZKLoadBalanceTest.testLoadBalance(ZKLoadBalanceTest.scala:89)
> [error] Test Failed: testPartitionedSendToNewTopic
> java.lang.AssertionError:
>   Unexpected method call send("test-topic1", 0, 
> ByteBufferMessageSet(MessageAndOffset(message(magic = 1, attributes = 0, crc 
> = 2326977762, payload = java.nio.HeapByteBuffer[pos=0 lim=5 cap=5]),15), )):
> close(): expected: 1, actual: 0
> at 
> org.easymock.internal.MockInvocationHandler.invoke(MockInvocationHandler.java:45)
> at 
> org.easymock.internal.ObjectMethodsFilter.invoke(ObjectMethodsFilter.java:73)
> at 
> org.easymock.internal.ClassProxyFactory$MockMethodInterceptor.intercept(ClassProxyFactory.java:92)
> at 
> kafka.producer.SyncProducer$$EnhancerByCGLIB$$4385e618.send()
> at 
> kafka.producer.ProducerPool$$anonfun$send$1.apply$mcVI$sp(ProducerPool.scala:114)
> at 
> kafka.producer.ProducerPool$$anonfun$send$1.apply(ProducerPool.scala:100)
> at 
> kafka.producer.ProducerPool$$anonfun$send$1.apply(ProducerPool.scala:100)
> at 
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:57)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:43)
> at kafka.producer.ProducerPool.send(ProducerPool.scala:100)
> at kafka.producer.Producer.zkSend(Producer.scala:137)
> at kafka.producer.Producer.send(Producer.scala:99)
> at 
> kafka.producer.ProducerTest.testPartitionedSendToNewTopic(ProducerTest.scala:576)
> [error] Test Failed: testZKSendToNewTopic
> junit.framework.AssertionFailedError: Message set should have 1 message
> at junit.framework.Assert.fail(Assert.java:47)
> at junit.framework.Assert.assertTrue(Assert.java:20)
> at 
> kafka.producer.ProducerTest.testZKSendToNewTopic(ProducerTest.scala:429)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3701) Expose KafkaStreams metrics in public API

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3701:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Expose KafkaStreams metrics in public API
> -
>
> Key: KAFKA-3701
> URL: https://issues.apache.org/jira/browse/KAFKA-3701
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Jeff Klukas
>Priority: Minor
> Fix For: 0.10.2.0
>
>
> The Kafka clients expose their metrics registries through a `metrics` method 
> presenting an unmodifiable collection, but `KafkaStreams` does not expose its 
> registry. Currently, applications can access a StreamsMetrics instance via 
> the ProcessorContext within a Processor, but this limits flexibility.
> Having read-only access to a KafkaStreams.metrics() method would allow a 
> developer to define a health check for their application based on the metrics 
> that KafkaStreams is collecting. Or a developer might want to define a metric 
> in some other framework based on KafkaStreams' metrics.
> I am imagining that an application would build and register 
> KafkaStreams-based health checks after building a KafkaStreams instance but 
> before calling the start() method. Are metrics added to the registry at the 
> time a KafkaStreams instance is constructed, or only after calling the 
> start() method? If metrics are registered only after application startup, 
> then this approach may not be sufficient.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3209) Support single message transforms in Kafka Connect

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3209:
---
 Priority: Major  (was: Critical)
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Support single message transforms in Kafka Connect
> --
>
> Key: KAFKA-3209
> URL: https://issues.apache.org/jira/browse/KAFKA-3209
> Project: Kafka
>  Issue Type: Improvement
>  Components: KafkaConnect
>Reporter: Neha Narkhede
>  Labels: needs-kip
> Fix For: 0.10.2.0
>
>
> Users should be able to perform light transformations on messages between a 
> connector and Kafka. This is needed because some transformations must be 
> performed before the data hits Kafka (e.g. filtering certain types of events 
> or PII filtering). It's also useful for very light, single-message 
> modifications that are easier to perform inline with the data import/export.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3297) More optimally balanced partition assignment strategy (new consumer)

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3297:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> More optimally balanced partition assignment strategy (new consumer)
> 
>
> Key: KAFKA-3297
> URL: https://issues.apache.org/jira/browse/KAFKA-3297
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Andrew Olson
>Assignee: Andrew Olson
> Fix For: 0.10.2.0
>
>
> While the roundrobin partition assignment strategy is an improvement over the 
> range strategy, when the consumer topic subscriptions are not identical 
> (previously disallowed but will be possible as of KAFKA-2172) it can produce 
> heavily skewed assignments. As suggested 
> [here|https://issues.apache.org/jira/browse/KAFKA-2172?focusedCommentId=14530767=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14530767]
>  it would be nice to have a strategy that attempts to assign an equal number 
> of partitions to each consumer in a group, regardless of how similar their 
> individual topic subscriptions are. We can accomplish this by tracking the 
> number of partitions assigned to each consumer, and having the partition 
> assignment loop assign each partition to a consumer interested in that topic 
> with the least number of partitions assigned. 
> Additionally, we can optimize the distribution fairness by adjusting the 
> partition assignment order:
> * Topics with fewer consumers are assigned first.
> * In the event of a tie for least consumers, the topic with more partitions 
> is assigned first.
> The general idea behind these two rules is to keep the most flexible 
> assignment choices available as long as possible by starting with the most 
> constrained partitions/consumers.
> This JIRA addresses the new consumer. For the original high-level consumer, 
> see KAFKA-2435.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3522) Consider adding version information into rocksDB storage format

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3522:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Consider adding version information into rocksDB storage format
> ---
>
> Key: KAFKA-3522
> URL: https://issues.apache.org/jira/browse/KAFKA-3522
> Project: Kafka
>  Issue Type: Bug
>  Components: streams
>Reporter: Guozhang Wang
>Assignee: Ishita Mandhan
>  Labels: architecture
> Fix For: 0.10.2.0
>
>
> Kafka Streams does not introduce any modifications to the data format in the 
> underlying Kafka protocol, but it does use RocksDB for persistent state 
> storage, and currently its data format is fixed and hard-coded. We want to 
> consider the evolution path in the future we we change the data format, and 
> hence having some version info stored along with the storage file / directory 
> would be useful.
> And this information could be even out of the storage file; for example, we 
> can just use a small "version indicator" file in the rocksdb directory for 
> this purposes. Thoughts? [~enothereska] [~jkreps]



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3366) Find a way to auto-generate expected error codes

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3366:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Find a way to auto-generate expected error codes
> 
>
> Key: KAFKA-3366
> URL: https://issues.apache.org/jira/browse/KAFKA-3366
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Grant Henke
> Fix For: 0.10.2.0
>
>
> Currently we comment on the expected error codes in the client 
> Response/Request implementations. It would be nice to have this be a part of 
> the protocol documentation and auto generated in the docs. 
> The documentations should include the expected errors codes for a given 
> request and recommendations on how they should be handled.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1895) Investigate moving deserialization and decompression out of KafkaConsumer

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1895:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Investigate moving deserialization and decompression out of KafkaConsumer
> -
>
> Key: KAFKA-1895
> URL: https://issues.apache.org/jira/browse/KAFKA-1895
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Jay Kreps
> Fix For: 0.10.2.0
>
>
> The consumer implementation in KAFKA-1760 decompresses fetch responses and 
> deserializes them into ConsumerRecords which are then handed back as the 
> result of poll().
> There are several downsides to this:
> 1. It is impossible to scale serialization and decompression work beyond the 
> single thread running the KafkaConsumer.
> 2. The results can come back during the processing of other calls such as 
> commit() etc which can result in caching these records a little longer.
> An alternative would be to have ConsumerRecords wrap the actual compressed 
> serialized MemoryRecords chunks and do the deserialization during iteration. 
> This way you could scale this over a thread pool if needed.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3364) Centrallize doc generation

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3364:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Centrallize doc generation
> --
>
> Key: KAFKA-3364
> URL: https://issues.apache.org/jira/browse/KAFKA-3364
> Project: Kafka
>  Issue Type: Sub-task
>  Components: build
>Reporter: Grant Henke
> Fix For: 0.10.2.0
>
>
> Currently docs generation is scattered throughout the build file/process. 
> Centralizing doc generation into its own location/file would help make the 
> process more clear.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-853) Allow OffsetFetchRequest to initialize offsets

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-853:
--
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Allow OffsetFetchRequest to initialize offsets
> --
>
> Key: KAFKA-853
> URL: https://issues.apache.org/jira/browse/KAFKA-853
> Project: Kafka
>  Issue Type: Improvement
>  Components: core
>Affects Versions: 0.8.1
>Reporter: David Arthur
>Assignee: Balaji Seshadri
> Fix For: 0.10.2.0
>
>   Original Estimate: 24h
>  Remaining Estimate: 24h
>
> It would be nice for the OffsetFetchRequest API to have the option to 
> initialize offsets instead of returning unknown_topic_or_partition. It could 
> mimic the Offsets API by adding the "time" field and then follow the same 
> code path on the server as the Offset API. 
> In this case, the response would need to a boolean to indicate if the 
> returned offset was initialized or fetched from ZK.
> This would simplify the client logic when dealing with new topics.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Commented] (KAFKA-3333) Client Partitioner - Round Robin

[Jason Gustafson (JIRA)]

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

Jason Gustafson commented on KAFKA-:


I think this change probably calls for a KIP since it's a change to the public 
API. Moving to 0.10.2.0 for now.

> Client Partitioner - Round Robin
> 
>
> Key: KAFKA-
> URL: https://issues.apache.org/jira/browse/KAFKA-
> Project: Kafka
>  Issue Type: New Feature
>  Components: clients
>Affects Versions: 0.10.0.0
>Reporter: Stephen Powis
> Fix For: 0.10.2.0
>
>   Original Estimate: 1h
>  Remaining Estimate: 1h
>
> The 
> [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java]
>  typically distributes using the hash of the keybytes, and falls back to 
> round robin if there is no key.  But there is currently no way to do Round 
> Robin partitioning if you have keys on your messages without writing your own 
> partitioning implementation.
> I think it'd be helpful to have an implementation of straight Round Robin 
> partitioning included with the library.  



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3333) Client Partitioner - Round Robin

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Client Partitioner - Round Robin
> 
>
> Key: KAFKA-
> URL: https://issues.apache.org/jira/browse/KAFKA-
> Project: Kafka
>  Issue Type: New Feature
>  Components: clients
>Affects Versions: 0.10.0.0
>Reporter: Stephen Powis
> Fix For: 0.10.2.0
>
>   Original Estimate: 1h
>  Remaining Estimate: 1h
>
> The 
> [DefaultPartitioner|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java]
>  typically distributes using the hash of the keybytes, and falls back to 
> round robin if there is no key.  But there is currently no way to do Round 
> Robin partitioning if you have keys on your messages without writing your own 
> partitioning implementation.
> I think it'd be helpful to have an implementation of straight Round Robin 
> partitioning included with the library.  



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3370) Add options to auto.offset.reset to reset offsets upon initialization only

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3370:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add options to auto.offset.reset to reset offsets upon initialization only
> --
>
> Key: KAFKA-3370
> URL: https://issues.apache.org/jira/browse/KAFKA-3370
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
>Assignee: Vahid Hashemian
> Fix For: 0.10.2.0
>
>
> Currently "auto.offset.reset" is applied in the following two cases:
> 1) upon starting the consumer for the first time (hence no committed offsets 
> before);
> 2) upon fetching offsets out-of-range.
> For scenarios where case 2) needs to be avoid (i.e. people need to be 
> notified upon offsets out-of-range rather than silently offset reset), 
> "auto.offset.reset" need to be set to "none". However for case 1) setting 
> "auto.offset.reset" to "none" will cause NoOffsetForPartitionException upon 
> polling. And in this case, seekToBeginning/seekToEnd is mistakenly applied 
> trying to set the offset at initialization, which are actually designed for 
> during the life time of the consumer (in rebalance callback, for example).
> The fix proposal is to add two more options to "auto.offset.reset", 
> "earliest-on-start", and "latest-on-start", whose semantics are "earliest" 
> and "latest" for case 1) only, and "none" for case 2).



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Resolved] (KAFKA-1332) Add functionality to the offsetsBeforeTime() API

[Jason Gustafson (JIRA)]

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

Jason Gustafson resolved KAFKA-1332.

   Resolution: Duplicate
 Assignee: Jason Gustafson
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

Resolving this as a duplicate of KAFKA-4148, which is for KIP-79.

> Add functionality to the offsetsBeforeTime() API
> 
>
> Key: KAFKA-1332
> URL: https://issues.apache.org/jira/browse/KAFKA-1332
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Affects Versions: 0.9.0.0
>Reporter: Neha Narkhede
>Assignee: Jason Gustafson
> Fix For: 0.10.2.0
>
>
> Add functionality to the offsetsBeforeTime() API to load offsets 
> corresponding to a particular timestamp, including earliest and latest offsets



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1776) Re-factor out existing tools that have been implemented behind the CLI

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1776:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Re-factor out existing tools that have been implemented behind the CLI
> --
>
> Key: KAFKA-1776
> URL: https://issues.apache.org/jira/browse/KAFKA-1776
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Joe Stein
>Priority: Minor
> Fix For: 0.10.2.0
>
>




--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3714) Allow users greater access to register custom streams metrics

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3714:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Allow users greater access to register custom streams metrics
> -
>
> Key: KAFKA-3714
> URL: https://issues.apache.org/jira/browse/KAFKA-3714
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Jeff Klukas
>Assignee: Guozhang Wang
>Priority: Minor
>  Labels: api
> Fix For: 0.10.2.0
>
>
> Copying in some discussion that originally appeared in 
> https://github.com/apache/kafka/pull/1362#issuecomment-219064302
> Kafka Streams is largely a higher-level abstraction on top of producers and 
> consumers, and it seems sensible to match the KafkaStreams interface to that 
> of KafkaProducer and KafkaConsumer where possible. For producers and 
> consumers, the metric registry is internal and metrics are only exposed as an 
> unmodifiable map. This allows users to access client metric values for use in 
> application health checks, etc., but doesn't allow them to register new 
> metrics.
> That approach seems reasonable if we assume that a user interested in 
> defining custom metrics is already going to be using a separate metrics 
> library. In such a case, users will likely find it easier to define metrics 
> using whatever library they're familiar with rather than learning the API for 
> Kafka's Metrics class. Is this a reasonable assumption?
> If we want to expose the Metrics instance so that users can define arbitrary 
> metrics, I'd argue that there's need for documentation updates. In 
> particular, I find the notion of metric tags confusing. Tags can be defined 
> in a MetricConfig when the Metrics instance is constructed, 
> StreamsMetricsImpl is maintaining its own set of tags, and users can set tag 
> overrides.
> If a user were to get access to the Metrics instance, they would be missing 
> the tags defined in StreamsMetricsImpl. I'm imagining that users would want 
> their custom metrics to sit alongside the predefined metrics with the same 
> tags, and users shouldn't be expected to manage those additional tags 
> themselves.
> So, why are we allowing users to define their own metrics via the 
> StreamsMetrics interface in the first place? Is it that we'd like to be able 
> to provide a built-in latency metric, but the definition depends on the 
> details of the use case so there's no generic solution? That would be 
> sufficient motivation for this special case of addLatencySensor. If we want 
> to continue down that path and give users access to define a wider range of 
> custom metrics, I'd prefer to extend the StreamsMetrics interface so that 
> users can call methods on that object, automatically getting the tags 
> appropriate for that instance rather than interacting with the raw Metrics 
> instance.
> ---
> Guozhang had the following comments:
> 1) For the producer/consumer cases, all internal metrics are provided and 
> abstracted from users, and they just need to read the documentation to poll 
> whatever provided metrics that they are interested; and if they want to 
> define more metrics, they are likely to be outside the clients themselves and 
> they can use whatever methods they like, so Metrics do not need to be exposed 
> to users.
> 2) For streams, things are a bit different: users define the computational 
> logic, which becomes part of the "Streams Client" processing and may be of 
> interests to be monitored by user themselves; think of a customized processor 
> that sends an email to some address based on a condition, and users want to 
> monitor the average rate of emails sent. Hence it is worth considering 
> whether or not they should be able to access the Metrics instance to define 
> their own along side the pre-defined metrics provided by the library.
> 3) Now, since the Metrics class was not previously designed for public usage, 
> it is not designed to be very user-friendly for defining sensors, especially 
> the semantics differences between name / scope / tags. StreamsMetrics tries 
> to hide some of these semantics confusion from users, but it still expose 
> tags and hence is not perfect in doing so. We need to think of a better 
> approach so that: 1) user defined metrics will be "aligned" (i.e. with the 
> same name prefix within a single application, with similar scope hierarchy 
> definition, etc) with library provided metrics, 2) natural APIs to do so.
> I do not have concrete ideas about 3) above on top of my head, comments are 
> more than welcomed.
> ---
> I'm not sure that I agree that 1) and 2) are truly different situations. A 
> user might choose to send email messages within a bare 

[jira] [Updated] (KAFKA-2525) Update ConsumerPerformance.scala to report join group time (new consumer)

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2525:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Update ConsumerPerformance.scala to report join group time (new consumer)
> -
>
> Key: KAFKA-2525
> URL: https://issues.apache.org/jira/browse/KAFKA-2525
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Geoff Anderson
>Assignee: Geoff Anderson
>Priority: Minor
> Fix For: 0.10.2.0
>
>
> The patch for https://issues.apache.org/jira/browse/KAFKA-2489 adds a small 
> amount of logic to ConsumerPerformance.scala which triggers consumer offsets 
> topic creation/group join etc. before consumer performance is measured.
> In https://github.com/apache/kafka/pull/179, [~gwenshap] pointed out that for 
> detecting regressions, it may be useful with the new consumer to report group 
> join time in addition to the other stats.
> The only tricky bit here is that this "consumer preparation" time also 
> includes time required to create the consumer offsets topic, so we'd like to 
> trigger creation of consumer offsets topic and make sure it's complete before 
> moving on the the "join group" phase.
> The solution here may be as simple as calling 
> consumer.partitionsFor(consumerOffsetTopic) to trigger a metadata request 
> (and hence topic creation)



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3596) Kafka Streams: Window expiration needs to consider more than event time

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3596:
---
Affects Version/s: (was: 0.10.1.0)
   0.10.0.0
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Kafka Streams: Window expiration needs to consider more than event time
> ---
>
> Key: KAFKA-3596
> URL: https://issues.apache.org/jira/browse/KAFKA-3596
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Affects Versions: 0.10.0.0
>Reporter: Henry Cai
>Assignee: Guozhang Wang
>Priority: Minor
>  Labels: architecture
> Fix For: 0.10.2.0
>
>
> Currently in Kafka Streams, the way the windows are expired in RocksDB is 
> triggered by new event insertion.  When a window is created at T0 with 10 
> minutes retention, when we saw a new record coming with event timestamp T0 + 
> 10 +1, we will expire that window (remove it) out of RocksDB.
> In the real world, it's very easy to see event coming with future timestamp 
> (or out-of-order events coming with big time gaps between events), this way 
> of retiring a window based on one event's event timestamp is dangerous.  I 
> think at least we need to consider both the event's event time and 
> server/stream time elapse.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3736) Add http metrics reporter

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3736:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add http metrics reporter
> -
>
> Key: KAFKA-3736
> URL: https://issues.apache.org/jira/browse/KAFKA-3736
> Project: Kafka
>  Issue Type: New Feature
>  Components: core
>Reporter: Adrian Muraru
> Fix For: 0.10.2.0
>
>
> The current builtin JMX metrics reporter is pretty heavy in terms of load and 
> collection. A new http lightweight reporter is proposed to expose the metrics 
> via a local http port.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-2787) Refactor gradle build

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2787:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Refactor gradle build
> -
>
> Key: KAFKA-2787
> URL: https://issues.apache.org/jira/browse/KAFKA-2787
> Project: Kafka
>  Issue Type: Task
>  Components: build
>Affects Versions: 0.8.2.2
>Reporter: Grant Henke
>Assignee: Grant Henke
> Fix For: 0.10.2.0
>
>
> The build files are quite large with a lot of duplication and overlap. This 
> could lead to mistakes, reduce readability and functionality, and hinder 
> future changes.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-2758) Improve Offset Commit Behavior

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2758:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Improve Offset Commit Behavior
> --
>
> Key: KAFKA-2758
> URL: https://issues.apache.org/jira/browse/KAFKA-2758
> Project: Kafka
>  Issue Type: Sub-task
>  Components: consumer
>Reporter: Guozhang Wang
>  Labels: newbiee
> Fix For: 0.10.2.0
>
>
> There are two scenarios of offset committing that we can improve:
> 1) we can filter the partitions whose committed offset is equal to the 
> consumed offset, meaning there is no new consumed messages from this 
> partition and hence we do not need to include this partition in the commit 
> request.
> 2) we can make a commit request right after resetting to a fetch / consume 
> position either according to the reset policy (e.g. on consumer starting up, 
> or handling of out of range offset, etc), or through the {code} seek {code} 
> so that if the consumer fails right after these event, upon recovery it can 
> restarts from the reset position instead of resetting again: this can lead 
> to, for example, data loss if we use "largest" as reset policy while there 
> are new messages coming to the fetching partitions.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3203) Add UnknownCodecException and UnknownMagicByteException to error mapping

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3203:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add UnknownCodecException and UnknownMagicByteException to error mapping
> 
>
> Key: KAFKA-3203
> URL: https://issues.apache.org/jira/browse/KAFKA-3203
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, core
>Affects Versions: 0.9.0.0
>Reporter: Jiangjie Qin
>Assignee: Grant Henke
> Fix For: 0.10.2.0
>
>
> Currently most of the exceptions to user have an error code. While 
> UnknownCodecException and UnknownMagicByteException can also be thrown to 
> client, broker does not have error mapping for them, so clients will only 
> receive UnknownServerException, which is vague.
> We should create those two exceptions in client package and add them to error 
> mapping.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1206) allow Kafka to start from a resource negotiator system

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1206:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> allow Kafka to start from a resource negotiator system
> --
>
> Key: KAFKA-1206
> URL: https://issues.apache.org/jira/browse/KAFKA-1206
> Project: Kafka
>  Issue Type: Bug
>Reporter: Joe Stein
>  Labels: mesos
> Fix For: 0.10.2.0
>
> Attachments: KAFKA-1206_2014-01-16_00:40:30.patch
>
>
> We need a generic implementation to hold the property information for 
> brokers, producers and consumers.  We want the resource negotiator to store 
> this information however it wants and give it respond with a 
> java.util.Properties.  This can get used then in the Kafka.scala as 
> serverConfigs for the KafkaServerStartable.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-3575) Use console consumer access topic that does not exist, can not use "Control + C" to exit process

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3575:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Use console consumer access topic that does not exist, can not use "Control + 
> C" to exit process
> 
>
> Key: KAFKA-3575
> URL: https://issues.apache.org/jira/browse/KAFKA-3575
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.9.0.0
> Environment: SUSE Linux Enterprise Server 11 SP3
>Reporter: NieWang
>Priority: Minor
> Fix For: 0.10.2.0
>
>
> 1.  use "sh kafka-console-consumer.sh --zookeeper 10.252.23.133:2181 --topic 
> topic_02"  start console consumer. topic_02 does not exist.
> 2. you can not use "Control + C" to exit console consumer process. The 
> process is blocked.
> 3. use jstack check process stack, as follows:
> linux:~ # jstack 122967
> 2016-04-18 15:46:06
> Full thread dump Java HotSpot(TM) 64-Bit Server VM (25.66-b17 mixed mode):
> "Attach Listener" #29 daemon prio=9 os_prio=0 tid=0x01781800 
> nid=0x1e0c8 waiting on condition [0x]
>java.lang.Thread.State: RUNNABLE
> "Thread-4" #27 prio=5 os_prio=0 tid=0x018a4000 nid=0x1e08a waiting on 
> condition [0x7ffbe5ac]
>java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for  <0xe00ed3b8> (a 
> java.util.concurrent.CountDownLatch$Sync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at 
> java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at 
> java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
> at 
> java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
> at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
> at kafka.tools.ConsoleConsumer$$anon$1.run(ConsoleConsumer.scala:101)
> "SIGINT handler" #28 daemon prio=9 os_prio=0 tid=0x019d5800 
> nid=0x1e089 in Object.wait() [0x7ffbe5bc1000]
>java.lang.Thread.State: WAITING (on object monitor)
> at java.lang.Object.$$YJP$$wait(Native Method)
> at java.lang.Object.wait(Object.java)
> at java.lang.Thread.join(Thread.java:1245)
> - locked <0xe71fd4e8> (a kafka.tools.ConsoleConsumer$$anon$1)
> at java.lang.Thread.join(Thread.java:1319)
> at 
> java.lang.ApplicationShutdownHooks.runHooks(ApplicationShutdownHooks.java:106)
> at 
> java.lang.ApplicationShutdownHooks$1.run(ApplicationShutdownHooks.java:46)
> at java.lang.Shutdown.runHooks(Shutdown.java:123)
> at java.lang.Shutdown.sequence(Shutdown.java:167)
> at java.lang.Shutdown.exit(Shutdown.java:212)
> - locked <0xe00abfd8> (a java.lang.Class for 
> java.lang.Shutdown)
> at java.lang.Terminator$1.handle(Terminator.java:52)
> at sun.misc.Signal$1.run(Signal.java:212)
> at java.lang.Thread.run(Thread.java:745)
> "metrics-meter-tick-thread-2" #20 daemon prio=5 os_prio=0 
> tid=0x7ffbec77a800 nid=0x1e079 waiting on condition [0x7ffbe66c8000]
>java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for  <0xe6fa6438> (a 
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at 
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
> at 
> java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:1088)
> at 
> java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.take(ScheduledThreadPoolExecutor.java:809)
> at 
> java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1067)
> at 
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1127)
> at 
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> "metrics-meter-tick-thread-1" #19 daemon prio=5 os_prio=0 
> tid=0x7ffbec783000 nid=0x1e078 waiting on condition [0x7ffbe67c9000]
>java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for  <0xe6fa6438> (a 
> java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
>  

[jira] [Updated] (KAFKA-3677) Add a producer performance tuning tool

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-3677:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add a producer performance tuning tool
> --
>
> Key: KAFKA-3677
> URL: https://issues.apache.org/jira/browse/KAFKA-3677
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.2.0
>
>
> In general, the producer of Kafka needs to be tuned based on the user traffic 
> pattern in order to get the optimal performance. It would be useful to 
> provide a tool that helps user explore different settings based on the user 
> traffic pattern (message size, compression type and ratio). 
> This ticket will use ProducerPerformance with synthetic traffic of the data 
> pattern specified by user to to explore different producer configurations and 
> offer performance tuning suggestions to the users.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1704) Add PartitionConfig besides LogConfig

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1704:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add PartitionConfig besides LogConfig
> -
>
> Key: KAFKA-1704
> URL: https://issues.apache.org/jira/browse/KAFKA-1704
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
>Assignee: Guozhang Wang
> Fix For: 0.10.2.0
>
>
> Today we only two places to store configs: server configs which is used to 
> store server side global configs, and log configs to store others. However, 
> many topic / partition level configs would be better stored in a partition 
> config such that they do not need to require accessing the underlying logs, 
> for example:
> 1. uncleanLeaderElectionEnable
> 2. minInSyncReplicas
> 3. compact [? this is defined per-topic / partition but maybe ok to store as 
> log configs]



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-566) Add last modified time to the TopicMetadataRequest

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-566:
--
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Add last modified time to the TopicMetadataRequest
> --
>
> Key: KAFKA-566
> URL: https://issues.apache.org/jira/browse/KAFKA-566
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.8.0
>Reporter: Jay Kreps
>Assignee: Sriharsha Chintalapani
> Fix For: 0.10.2.0
>
>
> To support KAFKA-560 it would be nice to have a last modified time in the 
> TopicMetadataRequest. This would be the timestamp of the last append to the 
> log as taken from stat on the final log segment.
> Implementation would involve
> 1. Adding a new field to TopicMetadataRequest
> 2. Adding a method Log.lastModified: Long to get the last modified time from 
> a log
> This timestamp would, of course, be subject to error in the event that the 
> file was touched without modification, but I think that is actually okay 
> since it provides a manual way to avoid gc'ing a topic that you  know you 
> will want.
> It is debatable whether this should go in 0.8. It would be nice to add the 
> field to the metadata request, at least, as that change should be easy and 
> would avoid needing to bump the version in the future.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1935) Consumer should use a separate socket for Coordinator connection

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1935:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> Consumer should use a separate socket for Coordinator connection
> 
>
> Key: KAFKA-1935
> URL: https://issues.apache.org/jira/browse/KAFKA-1935
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Guozhang Wang
>  Labels: newbiee
> Fix For: 0.10.2.0
>
>
> KAFKA-1925 is just a quick-fix of this issue, we need to let consumer to be 
> able to create separate sockets for the same server for coordinator / broker 
> roles.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-1694) KIP-4: Command line and centralized operations

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-1694:
---
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0
  Description: 
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Command+Line+and+Related+Improvements
  (was: 
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Command+Line+and+Related+Improvements)

> KIP-4: Command line and centralized operations
> --
>
> Key: KAFKA-1694
> URL: https://issues.apache.org/jira/browse/KAFKA-1694
> Project: Kafka
>  Issue Type: New Feature
>Reporter: Joe Stein
>Assignee: Grant Henke
>Priority: Critical
> Fix For: 0.10.2.0
>
> Attachments: KAFKA-1694.patch, KAFKA-1694_2014-12-24_21:21:51.patch, 
> KAFKA-1694_2015-01-12_15:28:41.patch, KAFKA-1694_2015-01-12_18:54:48.patch, 
> KAFKA-1694_2015-01-13_19:30:11.patch, KAFKA-1694_2015-01-14_15:42:12.patch, 
> KAFKA-1694_2015-01-14_18:07:39.patch, KAFKA-1694_2015-03-12_13:04:37.patch, 
> KAFKA-1772_1802_1775_1774_v2.patch
>
>
> https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Command+Line+and+Related+Improvements



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

[jira] [Updated] (KAFKA-747) RequestChannel re-design

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-747:
--
Fix Version/s: (was: 0.10.1.0)
   0.10.2.0

> RequestChannel re-design
> 
>
> Key: KAFKA-747
> URL: https://issues.apache.org/jira/browse/KAFKA-747
> Project: Kafka
>  Issue Type: New Feature
>  Components: network
>Reporter: Jay Kreps
>Assignee: Neha Narkhede
> Fix For: 0.10.2.0
>
>
> We have had some discussion around how to handle queuing requests. There are 
> two competing concerns:
> 1. We need to maintain request order on a per-socket basis.
> 2. We want to be able to balance load flexibly over a pool of threads so that 
> if one thread blocks on I/O request processing continues.
> Two Approaches We Have Considered
> 1. Have a global queue of unprocessed requests. All I/O threads read requests 
> off this global queue and process them. To avoid re-ordering have the network 
> layer only read one request at a time.
> 2. Have a queue per I/O thread and have the network threads statically map 
> sockets to I/O thread request queues.
> Problems With These Approaches
> In the first case you are not able to get any per-producer parallelism. That 
> is you can't read the next request while the current one is being handled. 
> This seems like it would not be a big deal, but preliminary benchmarks show 
> that it might be. 
> In the second case there are two problems. The first is that when an I/O 
> thread gets blocked all request processing for sockets attached to that I/O 
> thread will grind to a halt. If you have 10,000 connections, and  10 I/O 
> threads, then each blockage will stop 1,000 producers. If there is one topic 
> that has long synchronous flush times enabled (or is experiencing fsync 
> locking) this will cause big latency blips for all producers using that I/O 
> thread. The next problem is around backpressure and memory management. Say we 
> use BlockingQueues to feed the I/O threads. And say that one I/O thread 
> stalls. It's request queue will fill up and it will then block ALL network 
> threads, since they will block on inserting into that queue, even though the 
> other I/O threads are unused and have empty queues.
> A Proposed Better Solution
> The problem with the first solution is that we are not pipelining requests. 
> The problem with the second approach is that we are too constrained in moving 
> work from one I/O thread to another.
> Instead we should have a single request queue-like structure, but internally 
> enforce the condition that requests are not re-ordered.
> Here are the details. We retain RequestChannel but refactor its internals. 
> Internally we replace the blocking queue with a linked list. We also keep an 
> in-flight-keys array with one entry per I/O thread. When removing a work item 
> from the list we can't just take the first thing. Instead we need to walk the 
> list and look for something with a request key not in the in-flight-keys 
> array. When a response is sent, we remove that key from the in-flight array.
> This guarantees that requests for a socket with key K are ordered, but that 
> processing for K can only block requests made by K.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)