[jira] [Commented] (KAFKA-3289) Update Kafka protocol guide wiki for KIP-31 / KIP-32

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3289:
-

[~eapache] Thanks for the comments on the wiki page. I just added the request 
version for all the ApiKeys on the protocol guide wiki. 
Personally I prefer not to comment on that wiki page directly to avoid having a 
long comment history on the page. Please feel free to either reply on this 
ticket or shoot me an email directly if you have any further questions.

> Update Kafka protocol guide wiki for KIP-31 / KIP-32
> 
>
> Key: KAFKA-3289
> URL: https://issues.apache.org/jira/browse/KAFKA-3289
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.0.0
>
>




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

[jira] [Commented] (KAFKA-3289) Update Kafka protocol guide wiki for KIP-31 / KIP-32

[Evan Huus (JIRA)]

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

Evan Huus commented on KAFKA-3289:
--

OK, I commented there because I didn't know this ticket existed and had no 
other obvious way to contact you.

It would be helpful if for the new versions of produce/consume requests you 
indicated which fields had changed somehow. Offset Commit Request does this by 
just listing the format three times, but since I think the changes this time 
are in the *message* format, I'm not sure the best way to show that.

Thanks!

> Update Kafka protocol guide wiki for KIP-31 / KIP-32
> 
>
> Key: KAFKA-3289
> URL: https://issues.apache.org/jira/browse/KAFKA-3289
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.0.0
>
>




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

[jira] [Commented] (KAFKA-2073) Replace TopicMetadata request/response with o.a.k.requests.metadata

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-2073:
---

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/988

KAFKA-2073: migrate to client-side topic metadata request/response



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-2073

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/988.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #988


commit 765a9fcebf46c3f05826a60eba323bd3e25cd7e0
Author: Jason Gustafson 
Date:   2016-03-01T01:43:30Z

KAFKA-2073: migrate to client-side topic metadata request/response




> Replace TopicMetadata request/response with o.a.k.requests.metadata
> ---
>
> Key: KAFKA-2073
> URL: https://issues.apache.org/jira/browse/KAFKA-2073
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Gwen Shapira
>Assignee: Jason Gustafson
>
> Replace TopicMetadata request/response with o.a.k.requests.metadata.
> Note, this is more challenging that it appears because while the wire 
> protocol is identical, the objects are completely different.



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

[GitHub] kafka pull request: KAFKA-2073: migrate to client-side topic metad...

[hachikuji]

GitHub user hachikuji opened a pull request:

https://github.com/apache/kafka/pull/988

KAFKA-2073: migrate to client-side topic metadata request/response



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/hachikuji/kafka KAFKA-2073

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/988.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #988


commit 765a9fcebf46c3f05826a60eba323bd3e25cd7e0
Author: Jason Gustafson 
Date:   2016-03-01T01:43:30Z

KAFKA-2073: migrate to client-side topic metadata request/response




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-3289) Update Kafka protocol guide wiki for KIP-31 / KIP-32

[Kim Christensen (JIRA)]

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

Kim Christensen commented on KAFKA-3289:


Hi Jiangjie,

I think it would be nice if it was clearly visible, that the changes to Message 
format and Produce Response isn't available in the current version of Kafka.

> Update Kafka protocol guide wiki for KIP-31 / KIP-32
> 
>
> Key: KAFKA-3289
> URL: https://issues.apache.org/jira/browse/KAFKA-3289
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.0.0
>
>




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

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

[Andrew Olson (JIRA)]

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

Andrew Olson updated KAFKA-3297:

Status: Patch Available  (was: In Progress)

> 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
>
> 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] [Commented] (KAFKA-2435) More optimally balanced partition assignment strategy

[Andrew Olson (JIRA)]

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

Andrew Olson commented on KAFKA-2435:
-

KIP link: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-49+-+Fair+Partition+Assignment+Strategy

> More optimally balanced partition assignment strategy
> -
>
> Key: KAFKA-2435
> URL: https://issues.apache.org/jira/browse/KAFKA-2435
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Andrew Olson
>Assignee: Andrew Olson
> Attachments: KAFKA-2435.patch
>
>
> 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 original high-level consumer. For the new consumer, 
> see KAFKA-3297.



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

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

[Andrew Olson (JIRA)]

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

Andrew Olson commented on KAFKA-3297:
-

KIP link: 
https://cwiki.apache.org/confluence/display/KAFKA/KIP-49+-+Fair+Partition+Assignment+Strategy

> 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
>
> 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)

Re: [VOTE] KIP-33 - Add a time based log index to Kafka

[Jun Rao]

Jiangjie,

Currently, we roll a new log segment if the index is full. We can probably
just do the same on the time index. This will bound the index size.

1. Sounds good.

2. I was wondering an edge case where the largest timestamp is in the
oldest segment and the time index is empty is in all newer segments. At
some point, we delete the oldest segment since it has expired. Then, we
delete all but the active segment. Now, what should the largest timestamp
be? Should it be the previous largest timestamp that we have seen or should
we dig out the largest timestamp in the active segment?

Thanks,

Jun


On Mon, Feb 29, 2016 at 7:29 PM, Becket Qin  wrote:

> Hi Jun,
>
> I think index.interval.bytes is used to control the density of the offset
> index. The counterpart of index.interval.bytes for time index is
> time.index.interval.ms. If we did not change the semantic of log.roll.ms,
> log.roll.ms/time.index.interval.ms and
> log.segment.bytes/index.interval.bytes are a perfect mapping from bytes to
> time. However, because we changed the behavior of log.roll.ms, we need to
> guard against a potentially excessively large time index. We can either
> reuse index.interval.bytes or introduce time.index.interval.bytes, but I
> cannot think of additional usage for time.index.interval.bytes other than
> limiting the time index size.
>
> I agree that the memory mapped file is probably not a big issue here and we
> can change the default index size to 2MB.
>
> For the two cases you mentioned.
> 1. Because the message offset in the time index is also monotonically
> increasing, truncating should be straightforward. i.e. only keep the
> entries that are pointing to the offsets earlier than the truncated to
> offsets.
>
> 2. The current assumption is that if the time index of a segment is empty
> and there are no previous time index entry, we will assume that segment
> should be removed - because all the older segment with even larger
> timestamp have been removed. So in the case you mentioned, during startup
> we will remove all the segments and roll out a new empty segment.
>
> Thanks,
>
> Jiangjie (Becket) Qin
>
>
>
> On Mon, Feb 29, 2016 at 6:09 PM, Jun Rao  wrote:
>
> > Hi, Becket,
> >
> > I thought that your proposal to build time-based index just based off
> > index.interval.bytes
> > is reasonable. Is there a particular need to also add time.
> > index.interval.bytes?
> >
> > Compute the pre-allocated index file size based on log segment file size
> > can be useful. However, the tricky thing is that log segment size can be
> > changed dynamically. Also, for mmap files, they don't use heap space,
> just
> > virtual memory, which will be paged in on demand. So, I am not sure if
> > memory space is a big concern there. The simplest thing is probably to
> > change the default index size to 2MB to match the default log segment
> size.
> >
> > A couple of other things to think through.
> >
> > 1. Currently, LogSegment supports truncating to an offset. How do we do
> > that on a time-based index?
> >
> > 2. Since it's possible to have a empty time-based index (if all message
> > timestamps are smaller than the largest timestamp in previous segment),
> we
> > need to figure out what timestamp to use for retaining such log segment.
> In
> > the extreme case, it can happen that after we delete an old log segment,
> > all of the new log segments have an empty time-based index, in this case,
> > how do we avoid losing track of the latest timestamp?
> >
> > Thanks,
> >
> > Jun
> >
> > On Sun, Feb 28, 2016 at 3:26 PM, Becket Qin 
> wrote:
> >
> > > Hi Guozhang,
> > >
> > > The size of memory mapped index file was also our concern as well. That
> > is
> > > why we are suggesting minute level time indexing instead of second
> level.
> > > There are a few thoughts on the extra memory cost of time index.
> > >
> > > 1. Currently all the index files are loaded as memory mapped files.
> > Notice
> > > that only the index of the active segment is of the default size 10MB.
> > > Typically the index of the old segments are much smaller than 10MB. So
> if
> > > we use the same initial size for time index files, the total amount of
> > > memory won't be doubled, but the memory cost of active segments will be
> > > doubled. (However, the 10MB value itself seems problematic, see later
> > > reasoning).
> > >
> > > 2. It is likely that the time index is much smaller than the offset
> index
> > > because user would adjust the time index interval ms depending on the
> > topic
> > > volume. i.e for a low volume topic the time index interval ms will be
> > much
> > > longer so that we can avoid inserting one time index entry for each
> > message
> > > in the extreme case.
> > >
> > > 3. To further guard against the unnecessary frequent insertion of time
> > > index entry, we used the index.interval.bytes as a restriction for time
> > > index entry as well. Such that even for a newly 

[jira] [Commented] (KAFKA-3147) Memory records is not writable in MirrorMaker

[Kyle Kavanagh (JIRA)]

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

Kyle Kavanagh commented on KAFKA-3147:
--

Any possibility this makes it into a 0.9.x build rather than 0.10?

> Memory records is not writable in MirrorMaker
> -
>
> Key: KAFKA-3147
> URL: https://issues.apache.org/jira/browse/KAFKA-3147
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.0
>Reporter: Meghana Narasimhan
>Assignee: Mayuresh Gharat
> Fix For: 0.10.0.0
>
>
> Hi,
> We are running a 3 node cluster (kafka version 0.9) and Node 0 also has a few 
> mirror makers running. 
> When we do a rolling restart of the cluster, the mirror maker shuts down with 
> the following errors.
> [2016-01-11 20:16:00,348] WARN Got error produce response with correlation id 
> 12491674 on topic-partition test-99, retrying (2147483646 attempts left). 
> Error: NOT_LEADER_FOR_PARTITION 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:00,853] FATAL [mirrormaker-thread-0] Mirror maker thread 
> failure due to  (kafka.tools.MirrorMaker$MirrorMakerThread)
> java.lang.IllegalStateException: Memory records is not writable
> at 
> org.apache.kafka.common.record.MemoryRecords.append(MemoryRecords.java:93)
> at 
> org.apache.kafka.clients.producer.internals.RecordBatch.tryAppend(RecordBatch.java:69)
> at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:168)
> at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:435)
> at 
> kafka.tools.MirrorMaker$MirrorMakerProducer.send(MirrorMaker.scala:593)
> at 
> kafka.tools.MirrorMaker$MirrorMakerThread$$anonfun$run$3.apply(MirrorMaker.scala:398)
> at 
> kafka.tools.MirrorMaker$MirrorMakerThread$$anonfun$run$3.apply(MirrorMaker.scala:398)
> 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.tools.MirrorMaker$MirrorMakerThread.run(MirrorMaker.scala:398)
> [2016-01-11 20:16:01,072] WARN Got error produce response with correlation id 
> 12491679 on topic-partition test-75, retrying (2147483646 attempts left). 
> Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:01,073] WARN Got error produce response with correlation id 
> 12491679 on topic-partition test-93, retrying (2147483646 attempts left). 
> Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:01,073] WARN Got error produce response with correlation id 
> 12491679 on topic-partition test-24, retrying (2147483646 attempts left). 
> Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:20,479] FATAL [mirrormaker-thread-0] Mirror maker thread 
> exited abnormally, stopping the whole mirror maker. 
> (kafka.tools.MirrorMaker$MirrorMakerThread)
> Curious if the NOT_LEADER_FOR_PARTITION is because of a potential bug hinted 
> at in the thread , 
> http://mail-archives.apache.org/mod_mbox/kafka-users/201505.mbox/%3ccajs3ho_u8s1xou_kudnfjamypjtmrjlw10qvkngn2yqkdan...@mail.gmail.com%3E
>
> And I think the mirror maker shuts down because of the 
> "abort.on.send.failure" which is set to true in our case. 



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

[GitHub] kafka pull request: KAFKA-3310: Fix for NPEs observed when throttl...

[auradkar]

GitHub user auradkar opened a pull request:

https://github.com/apache/kafka/pull/989

KAFKA-3310: Fix for NPEs observed when throttling clients.

The fix basically ensures that the throttleTimeSensor is non-null before 
handing off to record the metric value. We also record the throttle time to 0 
so that we don't recreate the sensor always.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/auradkar/kafka KAFKA-3310

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/989.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #989


commit cd5007eb3c94ae2d1983cc6a4b9a9fe4e96ff1b1
Author: Aditya Auradkar 
Date:   2016-03-01T20:18:59Z

KAFKA-3310: Fix for NPEs observed when throttling clients.

The fix basically ensures that the throttleTimeSensor is non-null before 
handing off to record the metric value. We also record the throttle time to 0 
so that we don't recreate the sensor always.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-3310) fetch requests can trigger repeated NPE when quota is enabled

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-3310:
---

GitHub user auradkar opened a pull request:

https://github.com/apache/kafka/pull/989

KAFKA-3310: Fix for NPEs observed when throttling clients.

The fix basically ensures that the throttleTimeSensor is non-null before 
handing off to record the metric value. We also record the throttle time to 0 
so that we don't recreate the sensor always.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/auradkar/kafka KAFKA-3310

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/989.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #989


commit cd5007eb3c94ae2d1983cc6a4b9a9fe4e96ff1b1
Author: Aditya Auradkar 
Date:   2016-03-01T20:18:59Z

KAFKA-3310: Fix for NPEs observed when throttling clients.

The fix basically ensures that the throttleTimeSensor is non-null before 
handing off to record the metric value. We also record the throttle time to 0 
so that we don't recreate the sensor always.




> fetch requests can trigger repeated NPE when quota is enabled
> -
>
> Key: KAFKA-3310
> URL: https://issues.apache.org/jira/browse/KAFKA-3310
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.1
>Reporter: Jun Rao
>
> We saw the following NPE when consumer quota is enabled. NPE is triggered on 
> every fetch request from the client.
> java.lang.NullPointerException
> at 
> kafka.server.ClientQuotaManager.recordAndMaybeThrottle(ClientQuotaManager.scala:122)
> at 
> kafka.server.KafkaApis.kafka$server$KafkaApis$$sendResponseCallback$3(KafkaApis.scala:419)
> at 
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$1.apply(KafkaApis.scala:436)
> at 
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$1.apply(KafkaApis.scala:436)
> at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:481)
> at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:431)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:69)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:60)
> at java.lang.Thread.run(Thread.java:745)
> One possible cause of this is the logic of removing inactive sensors. 
> Currently, in ClientQuotaManager, we create two sensors per clientId: a 
> throttleTimeSensor and a quotaSensor. Each sensor expires if it's not 
> actively updated for 1 hour. What can happen is that initially, the quota is 
> not exceeded. So, quotaSensor is being updated actively, but 
> throttleTimeSensor is not. At some point, throttleTimeSensor is removed by 
> the expiring thread. Now, we are in a situation that quotaSensor is 
> registered, but throttleTimeSensor is not. Later on, if the quota is 
> exceeded, we will hit the above NPE when trying to update throttleTimeSensor.



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

[jira] [Commented] (KAFKA-3310) fetch requests can trigger repeated NPE when quota is enabled

[Aditya Auradkar (JIRA)]

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

Aditya Auradkar commented on KAFKA-3310:


[~junrao] - can you take a look?

> fetch requests can trigger repeated NPE when quota is enabled
> -
>
> Key: KAFKA-3310
> URL: https://issues.apache.org/jira/browse/KAFKA-3310
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.1
>Reporter: Jun Rao
>
> We saw the following NPE when consumer quota is enabled. NPE is triggered on 
> every fetch request from the client.
> java.lang.NullPointerException
> at 
> kafka.server.ClientQuotaManager.recordAndMaybeThrottle(ClientQuotaManager.scala:122)
> at 
> kafka.server.KafkaApis.kafka$server$KafkaApis$$sendResponseCallback$3(KafkaApis.scala:419)
> at 
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$1.apply(KafkaApis.scala:436)
> at 
> kafka.server.KafkaApis$$anonfun$handleFetchRequest$1.apply(KafkaApis.scala:436)
> at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:481)
> at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:431)
> at kafka.server.KafkaApis.handle(KafkaApis.scala:69)
> at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:60)
> at java.lang.Thread.run(Thread.java:745)
> One possible cause of this is the logic of removing inactive sensors. 
> Currently, in ClientQuotaManager, we create two sensors per clientId: a 
> throttleTimeSensor and a quotaSensor. Each sensor expires if it's not 
> actively updated for 1 hour. What can happen is that initially, the quota is 
> not exceeded. So, quotaSensor is being updated actively, but 
> throttleTimeSensor is not. At some point, throttleTimeSensor is removed by 
> the expiring thread. Now, we are in a situation that quotaSensor is 
> registered, but throttleTimeSensor is not. Later on, if the quota is 
> exceeded, we will hit the above NPE when trying to update throttleTimeSensor.



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

[GitHub] kafka pull request: MINOR: Move streams-exmaples source files unde...

[guozhangwang]

GitHub user guozhangwang opened a pull request:

https://github.com/apache/kafka/pull/992

MINOR: Move streams-exmaples source files under src folder

Also remove some unused imports.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/guozhangwang/kafka KSExamples

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/992.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #992


commit c5b6855c395ff8d941feddc86cd2cc1152cb6408
Author: Guozhang Wang 
Date:   2016-03-02T00:22:23Z

Move streams-examples source files under src

commit 702e5baa46adeacd61c30a28ab3016c8086597a5
Author: Guozhang Wang 
Date:   2016-03-02T00:25:16Z

remove unused imports




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Gwen Shapira]

I don't see a use for the name - clients should be able to translate
ApiKey to name for any API they support, and I'm not sure why would a
client need to log anything about APIs it does not support. Am I
missing something?

On a related note, Magnus is currently on vacation, but he should be
back at the end of next week. I'd like to hold off on the vote until
he gets back since his experience in implementing clients  and his
opinions will be very valuable for this discussion.

Gwen

On Tue, Mar 1, 2016 at 4:24 PM, Ashish Singh  wrote:
> Works with me. I will update PR to remove this.
>
> Also, "api_name" have been pointed out as a concern. However, it can be
> handy for logging and similar purposes. Any take on that?
>
> On Tue, Mar 1, 2016 at 3:46 PM, Gwen Shapira  wrote:
>
>> Jay also mentioned:
>> "Or, alternately, since deprecation has no functional impact and is
>> just a message
>> to developers, we could just leave it out of the protocol and just have it
>> in release notes etc."
>>
>> I'm in favor of leaving it out of the protocol. I can't really see a
>> use-case.
>>
>> Gwen
>>
>> On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh  wrote:
>>
>> > I hope it is OK for me to make some progress here. I have made the
>> > following changes.
>> >
>> > 1. Updated KIP-35, to adopt Jay's suggestion on maintaining separate list
>> > of deprecated versions, instead of using a version of -1.
>> > 2. Added information on required permissions, Describe action on Cluster
>> > resource, to be able to retrieve protocol versions from a auth enabled
>> > Kafka cluster.
>> >
>> > Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary patch
>> is
>> > available to review, https://github.com/apache/kafka/pull/986
>> >
>> > On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh 
>> wrote:
>> >
>> > > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found it
>> > really
>> > > difficult to cope up with Kafka releases as they want to support users
>> on
>> > > different Kafka versions. Capability to retrieve protocol version will
>> > go a
>> > > long way to ease out those pain points. I will be happy to help out
>> with
>> > > the work on this KIP. @Magnus, thanks for driving this, is it OK if I
>> > carry
>> > > forward the work from here. It will be ideal to have this in 0.10.0.0.
>> > >
>> > > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps  wrote:
>> > >
>> > >> I wonder if we need to solve the error problem? I think this KIP
>> gives a
>> > >> descent work around.
>> > >>
>> > >> Probably we should have included an error in the response header, but
>> we
>> > >> debated it at the time decided not to and now it is pretty hard to add
>> > >> because the headers aren't versioned (d'oh).
>> > >>
>> > >> It seems like any other solution is going to be kind of a hack, right?
>> > >> Sending malformed responses back seems like not a clean solution...
>> > >>
>> > >> (Not sure if I was pro- having a top-level error or not, but in any
>> case
>> > >> the rationale for the decision was that so many of the requests were
>> > >> per-partition or per-topic or whatever and hence fail or succeed at
>> that
>> > >> level and this makes it hard to know what the right top-level error
>> code
>> > >> is
>> > >> and hard for the client to figure out what to do with the top level
>> > error
>> > >> if some of the partitions succeed but there is a top-level error).
>> > >>
>> > >> I think actually this new API actually gives a way to handle this
>> > >> gracefully on the client side by just having clients that want to be
>> > >> graceful check for support for their version. Clients that do that
>> will
>> > >> have a graceful message.
>> > >>
>> > >> At some point if we're ever reworking the headers we should really
>> > >> consider
>> > >> (a) versioning them and (b) adding a top-level error code in the
>> > response.
>> > >> But given this would be a big breaking change and this is really just
>> to
>> > >> give a nicer error message seems like it probably isn't worth it to
>> try
>> > to
>> > >> do something now.
>> > >>
>> > >> -Jay
>> > >>
>> > >>
>> > >>
>> > >> On Mon, Oct 12, 2015 at 8:11 PM, Jiangjie Qin
>> > > >
>> > >> wrote:
>> > >>
>> > >> > I am thinking instead of returning an empty response, it would be
>> > >> better to
>> > >> > return an explicit UnsupportedVersionException code.
>> > >> >
>> > >> > Today KafkaApis handles the error in the following way:
>> > >> > 1. For requests/responses using old Scala classes, KafkaApis uses
>> > >> > RequestOrResponse.handleError() to return an error response.
>> > >> > 2. For requests/response using Java classes (only JoinGroupRequest
>> and
>> > >> > Heartbeat now), KafkaApis calls AbstractRequest.getErrorResponse()
>> to
>> > >> > return an error response.
>> > >> >
>> > >> > In KAFKA-2512, I am returning an UnsupportedVersionException for
>> case

Re: [VOTE] KIP-33 - Add a time based log index to Kafka

[Jun Rao]

Hi, Jiangjie,

I was thinking perhaps just reusing index.interval.bytes is enough. Not
sure if there is much value in adding an additional time.index.interval.ms.

For 1, the timestamp index has entries of timestamp -> file position. So,
there is actually no offset in the index, right?

For 2, what you said makes sense for time-based retention. Does that apply
if the retention is trigged by size? The difference here is that we can't
assume all segments with messages of timestamp smaller than the latest
timestamp will be deleted after the message with the latest timestamp is
deleted.

Thanks,

Jun

On Tue, Mar 1, 2016 at 1:00 PM, Becket Qin  wrote:

> Hi Jun,
>
> Rolling out a new segment when the time index is full sounds good. So both
> time index and offset index will be sharing the configuration of max index
> size.
> If we do that, do you think we still want to reuse index.interval.bytes? If
> we don't, the risk is that in some corner cases, we might end up with many
> small segments. (e.g. small time.index.interval.ms with small max index
> size). But this is probably more of a misconfiguration.
>
> 2. If the broker is still running when all the segments except the active
> segment is deleted, we will have an in memory latest timestamp. So that is
> not a problem.
>
> In another case, if a broker boots up and sees only one segment with an
> empty time index file, we can scan the active segment and rebuild the time
> index.  i.e. we do not need to care about the previous largest timestamp
> but simply start over. (We need to scan the active segment because it is
> possible that the last message appended to the log has a timestamp not
> expired, but the broker died before inserting the time index entry for
> it.). If all the messages in the active segment has expired, we should roll
> out a new segment and reset the latest timetamp to -1.
> The principal here is that we will try to build the time indices for the
> existing segments that have not expired. If the message with previously
> latest timestamp has already been deleted, there is no need to remember
> that any more.
>
> That said, I believe this corner case is really because user is not
> configuring the acceptable time difference threshold appropriately.
>
> Thanks,
>
> Jiangjie (Becket) Qin
>
>
> On Tue, Mar 1, 2016 at 11:55 AM, Jun Rao  wrote:
>
> > Jiangjie,
> >
> > Currently, we roll a new log segment if the index is full. We can
> probably
> > just do the same on the time index. This will bound the index size.
> >
> > 1. Sounds good.
> >
> > 2. I was wondering an edge case where the largest timestamp is in the
> > oldest segment and the time index is empty is in all newer segments. At
> > some point, we delete the oldest segment since it has expired. Then, we
> > delete all but the active segment. Now, what should the largest timestamp
> > be? Should it be the previous largest timestamp that we have seen or
> should
> > we dig out the largest timestamp in the active segment?
> >
> > Thanks,
> >
> > Jun
> >
> >
> > On Mon, Feb 29, 2016 at 7:29 PM, Becket Qin 
> wrote:
> >
> > > Hi Jun,
> > >
> > > I think index.interval.bytes is used to control the density of the
> offset
> > > index. The counterpart of index.interval.bytes for time index is
> > > time.index.interval.ms. If we did not change the semantic of
> log.roll.ms
> > ,
> > > log.roll.ms/time.index.interval.ms and
> > > log.segment.bytes/index.interval.bytes are a perfect mapping from bytes
> > to
> > > time. However, because we changed the behavior of log.roll.ms, we need
> > to
> > > guard against a potentially excessively large time index. We can either
> > > reuse index.interval.bytes or introduce time.index.interval.bytes, but
> I
> > > cannot think of additional usage for time.index.interval.bytes other
> than
> > > limiting the time index size.
> > >
> > > I agree that the memory mapped file is probably not a big issue here
> and
> > we
> > > can change the default index size to 2MB.
> > >
> > > For the two cases you mentioned.
> > > 1. Because the message offset in the time index is also monotonically
> > > increasing, truncating should be straightforward. i.e. only keep the
> > > entries that are pointing to the offsets earlier than the truncated to
> > > offsets.
> > >
> > > 2. The current assumption is that if the time index of a segment is
> empty
> > > and there are no previous time index entry, we will assume that segment
> > > should be removed - because all the older segment with even larger
> > > timestamp have been removed. So in the case you mentioned, during
> startup
> > > we will remove all the segments and roll out a new empty segment.
> > >
> > > Thanks,
> > >
> > > Jiangjie (Becket) Qin
> > >
> > >
> > >
> > > On Mon, Feb 29, 2016 at 6:09 PM, Jun Rao  wrote:
> > >
> > > > Hi, Becket,
> > > >
> > > > I thought that your proposal to build time-based index just based off
> > > > 

[jira] [Updated] (KAFKA-3221) kafka-acls.sh must verify if a user has sufficient privileges to perform acls CRUD

[Ashish K Singh (JIRA)]

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

Ashish K Singh updated KAFKA-3221:
--
Resolution: Won't Fix
Status: Resolved  (was: Patch Available)

> kafka-acls.sh must verify if a user has sufficient privileges to perform acls 
> CRUD
> --
>
> Key: KAFKA-3221
> URL: https://issues.apache.org/jira/browse/KAFKA-3221
> Project: Kafka
>  Issue Type: Improvement
>Affects Versions: 0.9.0.0
>Reporter: Ashish K Singh
>Assignee: Ashish K Singh
> Fix For: 0.10.0.0
>
>
> kafka-acls.sh provides an insecure entry point to Kafka's authorization. No 
> checks are performed or no user information is provided to authorizer to 
> validate a user, before the user performs CRUD of acls. This is a security 
> hole that must be addressed.
> As Kafka supports pluggable authorization, we need to look at this issue from 
> two aspects.
> 1. Default zk based authorizer, SimpleAclAuthorizer
> For SimpleAclAuthorizer, one could rely on Zookeeper authentication to check 
> if a user can really perform CRUD on Kafka acls. However, this check relies 
> on the assumption, which is usually true, that non-admin users won't have 
> access to Kafka service's user account.
> 2. Custom Authorizer
> Custom authorizer that gets executed in same address space as of Kafka 
> broker, does not have any way of determining which user is really trying to 
> perform CRUD of acls. For authorize requests, authorizer gets user 
> information, KafkaPrincipal, from session, however for CRUD of acls, i.e., 
> addAcls, removeAcls and getAcls, authorizer does not have requestor's info to 
> validate if it should allow or deny the request.



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

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Ashish Singh]

On Tue, Mar 1, 2016 at 5:11 PM, Gwen Shapira  wrote:

> I don't see a use for the name - clients should be able to translate
> ApiKey to name for any API they support, and I'm not sure why would a
> client need to log anything about APIs it does not support. Am I
> missing something?
>
Yea, it is a fair assumption that client would know about APIs it supports.
It could have been helpful for client users to see new APIs though, however
users can always refer to protocol doc of new version to find corresponding
names of the new APIs.

>
> On a related note, Magnus is currently on vacation, but he should be
> back at the end of next week. I'd like to hold off on the vote until
> he gets back since his experience in implementing clients  and his
> opinions will be very valuable for this discussion.
>
That is great. It will be valuable to have his feedback. I will hold off on
removing "api_name" and "api_deprecated_versions" or adding release version.

>
> Gwen
>
> On Tue, Mar 1, 2016 at 4:24 PM, Ashish Singh  wrote:
> > Works with me. I will update PR to remove this.
> >
> > Also, "api_name" have been pointed out as a concern. However, it can be
> > handy for logging and similar purposes. Any take on that?
> >
> > On Tue, Mar 1, 2016 at 3:46 PM, Gwen Shapira  wrote:
> >
> >> Jay also mentioned:
> >> "Or, alternately, since deprecation has no functional impact and is
> >> just a message
> >> to developers, we could just leave it out of the protocol and just have
> it
> >> in release notes etc."
> >>
> >> I'm in favor of leaving it out of the protocol. I can't really see a
> >> use-case.
> >>
> >> Gwen
> >>
> >> On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh 
> wrote:
> >>
> >> > I hope it is OK for me to make some progress here. I have made the
> >> > following changes.
> >> >
> >> > 1. Updated KIP-35, to adopt Jay's suggestion on maintaining separate
> list
> >> > of deprecated versions, instead of using a version of -1.
> >> > 2. Added information on required permissions, Describe action on
> Cluster
> >> > resource, to be able to retrieve protocol versions from a auth enabled
> >> > Kafka cluster.
> >> >
> >> > Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary
> patch
> >> is
> >> > available to review, https://github.com/apache/kafka/pull/986
> >> >
> >> > On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh 
> >> wrote:
> >> >
> >> > > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found it
> >> > really
> >> > > difficult to cope up with Kafka releases as they want to support
> users
> >> on
> >> > > different Kafka versions. Capability to retrieve protocol version
> will
> >> > go a
> >> > > long way to ease out those pain points. I will be happy to help out
> >> with
> >> > > the work on this KIP. @Magnus, thanks for driving this, is it OK if
> I
> >> > carry
> >> > > forward the work from here. It will be ideal to have this in
> 0.10.0.0.
> >> > >
> >> > > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps 
> wrote:
> >> > >
> >> > >> I wonder if we need to solve the error problem? I think this KIP
> >> gives a
> >> > >> descent work around.
> >> > >>
> >> > >> Probably we should have included an error in the response header,
> but
> >> we
> >> > >> debated it at the time decided not to and now it is pretty hard to
> add
> >> > >> because the headers aren't versioned (d'oh).
> >> > >>
> >> > >> It seems like any other solution is going to be kind of a hack,
> right?
> >> > >> Sending malformed responses back seems like not a clean solution...
> >> > >>
> >> > >> (Not sure if I was pro- having a top-level error or not, but in any
> >> case
> >> > >> the rationale for the decision was that so many of the requests
> were
> >> > >> per-partition or per-topic or whatever and hence fail or succeed at
> >> that
> >> > >> level and this makes it hard to know what the right top-level error
> >> code
> >> > >> is
> >> > >> and hard for the client to figure out what to do with the top level
> >> > error
> >> > >> if some of the partitions succeed but there is a top-level error).
> >> > >>
> >> > >> I think actually this new API actually gives a way to handle this
> >> > >> gracefully on the client side by just having clients that want to
> be
> >> > >> graceful check for support for their version. Clients that do that
> >> will
> >> > >> have a graceful message.
> >> > >>
> >> > >> At some point if we're ever reworking the headers we should really
> >> > >> consider
> >> > >> (a) versioning them and (b) adding a top-level error code in the
> >> > response.
> >> > >> But given this would be a big breaking change and this is really
> just
> >> to
> >> > >> give a nicer error message seems like it probably isn't worth it to
> >> try
> >> > to
> >> > >> do something now.
> >> > >>
> >> > >> -Jay
> >> > >>
> >> > >>
> >> > >>
> >> > >> On Mon, Oct 12, 2015 at 8:11 PM, Jiangjie Qin
> >> 

[jira] [Commented] (KAFKA-3147) Memory records is not writable in MirrorMaker

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3147:
-

[~kdkavanagh] The next Kafka release will be 0.10.0. :)  So this fix will be in 
the next release.

> Memory records is not writable in MirrorMaker
> -
>
> Key: KAFKA-3147
> URL: https://issues.apache.org/jira/browse/KAFKA-3147
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.0
>Reporter: Meghana Narasimhan
>Assignee: Mayuresh Gharat
> Fix For: 0.10.0.0
>
>
> Hi,
> We are running a 3 node cluster (kafka version 0.9) and Node 0 also has a few 
> mirror makers running. 
> When we do a rolling restart of the cluster, the mirror maker shuts down with 
> the following errors.
> [2016-01-11 20:16:00,348] WARN Got error produce response with correlation id 
> 12491674 on topic-partition test-99, retrying (2147483646 attempts left). 
> Error: NOT_LEADER_FOR_PARTITION 
> (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:00,853] FATAL [mirrormaker-thread-0] Mirror maker thread 
> failure due to  (kafka.tools.MirrorMaker$MirrorMakerThread)
> java.lang.IllegalStateException: Memory records is not writable
> at 
> org.apache.kafka.common.record.MemoryRecords.append(MemoryRecords.java:93)
> at 
> org.apache.kafka.clients.producer.internals.RecordBatch.tryAppend(RecordBatch.java:69)
> at 
> org.apache.kafka.clients.producer.internals.RecordAccumulator.append(RecordAccumulator.java:168)
> at 
> org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:435)
> at 
> kafka.tools.MirrorMaker$MirrorMakerProducer.send(MirrorMaker.scala:593)
> at 
> kafka.tools.MirrorMaker$MirrorMakerThread$$anonfun$run$3.apply(MirrorMaker.scala:398)
> at 
> kafka.tools.MirrorMaker$MirrorMakerThread$$anonfun$run$3.apply(MirrorMaker.scala:398)
> 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.tools.MirrorMaker$MirrorMakerThread.run(MirrorMaker.scala:398)
> [2016-01-11 20:16:01,072] WARN Got error produce response with correlation id 
> 12491679 on topic-partition test-75, retrying (2147483646 attempts left). 
> Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:01,073] WARN Got error produce response with correlation id 
> 12491679 on topic-partition test-93, retrying (2147483646 attempts left). 
> Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:01,073] WARN Got error produce response with correlation id 
> 12491679 on topic-partition test-24, retrying (2147483646 attempts left). 
> Error: NETWORK_EXCEPTION (org.apache.kafka.clients.producer.internals.Sender)
> [2016-01-11 20:16:20,479] FATAL [mirrormaker-thread-0] Mirror maker thread 
> exited abnormally, stopping the whole mirror maker. 
> (kafka.tools.MirrorMaker$MirrorMakerThread)
> Curious if the NOT_LEADER_FOR_PARTITION is because of a potential bug hinted 
> at in the thread , 
> http://mail-archives.apache.org/mod_mbox/kafka-users/201505.mbox/%3ccajs3ho_u8s1xou_kudnfjamypjtmrjlw10qvkngn2yqkdan...@mail.gmail.com%3E
>
> And I think the mirror maker shuts down because of the 
> "abort.on.send.failure" which is set to true in our case. 



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

[GitHub] kafka pull request: HOTFIX: Use the correct serde classes

[guozhangwang]

GitHub user guozhangwang opened a pull request:

https://github.com/apache/kafka/pull/991

HOTFIX: Use the correct serde classes



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/guozhangwang/kafka KSerde

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/991.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #991


commit d4f0068517ab5b866ee4a8fe08cfab5bd3ee56cb
Author: Guozhang Wang 
Date:   2016-03-01T22:44:09Z

hotfix: use the correct serde classes




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Gwen Shapira]

Jay also mentioned:
"Or, alternately, since deprecation has no functional impact and is
just a message
to developers, we could just leave it out of the protocol and just have it
in release notes etc."

I'm in favor of leaving it out of the protocol. I can't really see a
use-case.

Gwen

On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh  wrote:

> I hope it is OK for me to make some progress here. I have made the
> following changes.
>
> 1. Updated KIP-35, to adopt Jay's suggestion on maintaining separate list
> of deprecated versions, instead of using a version of -1.
> 2. Added information on required permissions, Describe action on Cluster
> resource, to be able to retrieve protocol versions from a auth enabled
> Kafka cluster.
>
> Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary patch is
> available to review, https://github.com/apache/kafka/pull/986
>
> On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh  wrote:
>
> > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found it
> really
> > difficult to cope up with Kafka releases as they want to support users on
> > different Kafka versions. Capability to retrieve protocol version will
> go a
> > long way to ease out those pain points. I will be happy to help out with
> > the work on this KIP. @Magnus, thanks for driving this, is it OK if I
> carry
> > forward the work from here. It will be ideal to have this in 0.10.0.0.
> >
> > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps  wrote:
> >
> >> I wonder if we need to solve the error problem? I think this KIP gives a
> >> descent work around.
> >>
> >> Probably we should have included an error in the response header, but we
> >> debated it at the time decided not to and now it is pretty hard to add
> >> because the headers aren't versioned (d'oh).
> >>
> >> It seems like any other solution is going to be kind of a hack, right?
> >> Sending malformed responses back seems like not a clean solution...
> >>
> >> (Not sure if I was pro- having a top-level error or not, but in any case
> >> the rationale for the decision was that so many of the requests were
> >> per-partition or per-topic or whatever and hence fail or succeed at that
> >> level and this makes it hard to know what the right top-level error code
> >> is
> >> and hard for the client to figure out what to do with the top level
> error
> >> if some of the partitions succeed but there is a top-level error).
> >>
> >> I think actually this new API actually gives a way to handle this
> >> gracefully on the client side by just having clients that want to be
> >> graceful check for support for their version. Clients that do that will
> >> have a graceful message.
> >>
> >> At some point if we're ever reworking the headers we should really
> >> consider
> >> (a) versioning them and (b) adding a top-level error code in the
> response.
> >> But given this would be a big breaking change and this is really just to
> >> give a nicer error message seems like it probably isn't worth it to try
> to
> >> do something now.
> >>
> >> -Jay
> >>
> >>
> >>
> >> On Mon, Oct 12, 2015 at 8:11 PM, Jiangjie Qin  >
> >> wrote:
> >>
> >> > I am thinking instead of returning an empty response, it would be
> >> better to
> >> > return an explicit UnsupportedVersionException code.
> >> >
> >> > Today KafkaApis handles the error in the following way:
> >> > 1. For requests/responses using old Scala classes, KafkaApis uses
> >> > RequestOrResponse.handleError() to return an error response.
> >> > 2. For requests/response using Java classes (only JoinGroupRequest and
> >> > Heartbeat now), KafkaApis calls AbstractRequest.getErrorResponse() to
> >> > return an error response.
> >> >
> >> > In KAFKA-2512, I am returning an UnsupportedVersionException for case
> >> [1]
> >> > when see an unsupported version. This will put the error code per
> topic
> >> or
> >> > partition for most of the requests, but might not work all the time.
> >> e.g.
> >> > TopicMetadataRequest with an empty topic set.
> >> >
> >> > Case [2] does not quite work for unsupported version, because we will
> >> > thrown an uncaught exception when version is not recognized (BTW this
> >> is a
> >> > bug). Part of the reason is that for some response types, error code
> is
> >> not
> >> > part of the response level field.
> >> >
> >> > Maybe it worth checking how each response is dealing with error code
> >> today.
> >> > A scan of the response formats gives the following result:
> >> > 1. TopicMetadataResponse - per topic error code, does not work when
> the
> >> > topic set is empty in the request.
> >> > 2. ProduceResonse - per partition error code.
> >> > 3. OffsetCommitResponse - per partition.
> >> > 4. OffsetFetchResponse - per partition.
> >> > 5. OffsetResponse - per partition.
> >> > 6. FetchResponse - per partition
> >> > 7. ConsumerMetadataResponse - response level
> >> > 8. ControlledShutdownResponse - 

[GitHub] kafka pull request: HOTFIX: Use the correct serde classes

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/991


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Ashish Singh]

Works with me. I will update PR to remove this.

Also, "api_name" have been pointed out as a concern. However, it can be
handy for logging and similar purposes. Any take on that?

On Tue, Mar 1, 2016 at 3:46 PM, Gwen Shapira  wrote:

> Jay also mentioned:
> "Or, alternately, since deprecation has no functional impact and is
> just a message
> to developers, we could just leave it out of the protocol and just have it
> in release notes etc."
>
> I'm in favor of leaving it out of the protocol. I can't really see a
> use-case.
>
> Gwen
>
> On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh  wrote:
>
> > I hope it is OK for me to make some progress here. I have made the
> > following changes.
> >
> > 1. Updated KIP-35, to adopt Jay's suggestion on maintaining separate list
> > of deprecated versions, instead of using a version of -1.
> > 2. Added information on required permissions, Describe action on Cluster
> > resource, to be able to retrieve protocol versions from a auth enabled
> > Kafka cluster.
> >
> > Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary patch
> is
> > available to review, https://github.com/apache/kafka/pull/986
> >
> > On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh 
> wrote:
> >
> > > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found it
> > really
> > > difficult to cope up with Kafka releases as they want to support users
> on
> > > different Kafka versions. Capability to retrieve protocol version will
> > go a
> > > long way to ease out those pain points. I will be happy to help out
> with
> > > the work on this KIP. @Magnus, thanks for driving this, is it OK if I
> > carry
> > > forward the work from here. It will be ideal to have this in 0.10.0.0.
> > >
> > > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps  wrote:
> > >
> > >> I wonder if we need to solve the error problem? I think this KIP
> gives a
> > >> descent work around.
> > >>
> > >> Probably we should have included an error in the response header, but
> we
> > >> debated it at the time decided not to and now it is pretty hard to add
> > >> because the headers aren't versioned (d'oh).
> > >>
> > >> It seems like any other solution is going to be kind of a hack, right?
> > >> Sending malformed responses back seems like not a clean solution...
> > >>
> > >> (Not sure if I was pro- having a top-level error or not, but in any
> case
> > >> the rationale for the decision was that so many of the requests were
> > >> per-partition or per-topic or whatever and hence fail or succeed at
> that
> > >> level and this makes it hard to know what the right top-level error
> code
> > >> is
> > >> and hard for the client to figure out what to do with the top level
> > error
> > >> if some of the partitions succeed but there is a top-level error).
> > >>
> > >> I think actually this new API actually gives a way to handle this
> > >> gracefully on the client side by just having clients that want to be
> > >> graceful check for support for their version. Clients that do that
> will
> > >> have a graceful message.
> > >>
> > >> At some point if we're ever reworking the headers we should really
> > >> consider
> > >> (a) versioning them and (b) adding a top-level error code in the
> > response.
> > >> But given this would be a big breaking change and this is really just
> to
> > >> give a nicer error message seems like it probably isn't worth it to
> try
> > to
> > >> do something now.
> > >>
> > >> -Jay
> > >>
> > >>
> > >>
> > >> On Mon, Oct 12, 2015 at 8:11 PM, Jiangjie Qin
>  > >
> > >> wrote:
> > >>
> > >> > I am thinking instead of returning an empty response, it would be
> > >> better to
> > >> > return an explicit UnsupportedVersionException code.
> > >> >
> > >> > Today KafkaApis handles the error in the following way:
> > >> > 1. For requests/responses using old Scala classes, KafkaApis uses
> > >> > RequestOrResponse.handleError() to return an error response.
> > >> > 2. For requests/response using Java classes (only JoinGroupRequest
> and
> > >> > Heartbeat now), KafkaApis calls AbstractRequest.getErrorResponse()
> to
> > >> > return an error response.
> > >> >
> > >> > In KAFKA-2512, I am returning an UnsupportedVersionException for
> case
> > >> [1]
> > >> > when see an unsupported version. This will put the error code per
> > topic
> > >> or
> > >> > partition for most of the requests, but might not work all the time.
> > >> e.g.
> > >> > TopicMetadataRequest with an empty topic set.
> > >> >
> > >> > Case [2] does not quite work for unsupported version, because we
> will
> > >> > thrown an uncaught exception when version is not recognized (BTW
> this
> > >> is a
> > >> > bug). Part of the reason is that for some response types, error code
> > is
> > >> not
> > >> > part of the response level field.
> > >> >
> > >> > Maybe it worth checking how each response is dealing with error code
> > >> today.
> > >> 

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Ashish Singh]

Hello Dana,

On Mon, Feb 29, 2016 at 4:11 PM, Dana Powers  wrote:

> This is a fantastic and much-needed KIP. All third-party clients have had
> to deal with this issue. In my experience most clients are either declaring
> they only support version broker version X, or are spending a lot of time
> hacking around the issue. I think the community of non-java drivers would
> see significant benefit from this proposal.
>
> My specific thought is that for kafka-python it has been easier to manage
> compatibility using broker release version to gate various features by
> api-protocol version. For example, only enable group coordination apis if
> >= (0, 9), kafka-backed offsets >= (0, 8, 2), etc. As an example, here are
> some backwards compatibility issues that I think are difficult to capture
> w/ just the protocol versions:
>
> - LZ4 compression only supported in brokers >= 0.8.2, but no protocol
> change.
> - kafka-backed offset storage, in additional to requiring new offset
> commit/fetch protocol versions, also requires adding support for tracking
> the group coordinator.
> - 0.8.2-beta OffsetCommit api [different than 0.8.X release]
>
>
> Could release version be added to the api response in this proposal?
> Perhaps include a KafkaRelease string in the Response before the array of
> api versions?
>
I think that will be useful, however we should discuss this in next KIP
meeting to check if others see similar value. There are a few questions
that are raised on PR, we can briefly touch upon them as well.

>
> Thanks for the great KIP, Magnus. And thanks for restarting the discussion,
> Ashish. I also would like to see this addressed in 0.10
>
> -Dana
>
>
> On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh  wrote:
>
> > I hope it is OK for me to make some progress here. I have made the
> > following changes.
> >
> > 1. Updated KIP-35, to adopt Jay's suggestion on maintaining separate list
> > of deprecated versions, instead of using a version of -1.
> > 2. Added information on required permissions, Describe action on Cluster
> > resource, to be able to retrieve protocol versions from a auth enabled
> > Kafka cluster.
> >
> > Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary patch
> is
> > available to review, https://github.com/apache/kafka/pull/986
> >
> > On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh 
> wrote:
> >
> > > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found it
> > really
> > > difficult to cope up with Kafka releases as they want to support users
> on
> > > different Kafka versions. Capability to retrieve protocol version will
> > go a
> > > long way to ease out those pain points. I will be happy to help out
> with
> > > the work on this KIP. @Magnus, thanks for driving this, is it OK if I
> > carry
> > > forward the work from here. It will be ideal to have this in 0.10.0.0.
> > >
> > > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps  wrote:
> > >
> > >> I wonder if we need to solve the error problem? I think this KIP
> gives a
> > >> descent work around.
> > >>
> > >> Probably we should have included an error in the response header, but
> we
> > >> debated it at the time decided not to and now it is pretty hard to add
> > >> because the headers aren't versioned (d'oh).
> > >>
> > >> It seems like any other solution is going to be kind of a hack, right?
> > >> Sending malformed responses back seems like not a clean solution...
> > >>
> > >> (Not sure if I was pro- having a top-level error or not, but in any
> case
> > >> the rationale for the decision was that so many of the requests were
> > >> per-partition or per-topic or whatever and hence fail or succeed at
> that
> > >> level and this makes it hard to know what the right top-level error
> code
> > >> is
> > >> and hard for the client to figure out what to do with the top level
> > error
> > >> if some of the partitions succeed but there is a top-level error).
> > >>
> > >> I think actually this new API actually gives a way to handle this
> > >> gracefully on the client side by just having clients that want to be
> > >> graceful check for support for their version. Clients that do that
> will
> > >> have a graceful message.
> > >>
> > >> At some point if we're ever reworking the headers we should really
> > >> consider
> > >> (a) versioning them and (b) adding a top-level error code in the
> > response.
> > >> But given this would be a big breaking change and this is really just
> to
> > >> give a nicer error message seems like it probably isn't worth it to
> try
> > to
> > >> do something now.
> > >>
> > >> -Jay
> > >>
> > >>
> > >>
> > >> On Mon, Oct 12, 2015 at 8:11 PM, Jiangjie Qin
>  > >
> > >> wrote:
> > >>
> > >> > I am thinking instead of returning an empty response, it would be
> > >> better to
> > >> > return an explicit UnsupportedVersionException code.
> > >> >
> > >> > Today KafkaApis handles the 

[jira] [Commented] (KAFKA-2970) Both UpdateMetadataRequest.java and LeaderAndIsrRequest.java have an Endpoint class

[Guozhang Wang (JIRA)]

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

Guozhang Wang commented on KAFKA-2970:
--

[~ijuma] Makes sense, we probably should make them in `o.a.k.common.internals`.

> Both UpdateMetadataRequest.java and LeaderAndIsrRequest.java have an Endpoint 
> class
> ---
>
> Key: KAFKA-2970
> URL: https://issues.apache.org/jira/browse/KAFKA-2970
> Project: Kafka
>  Issue Type: Improvement
>  Components: clients
>Affects Versions: 0.9.0.0
>Reporter: Grant Henke
>Assignee: chen zhu
>
> Both UpdateMetadataRequest.java and LeaderAndIsrRequest.java have an Endpoint 
> class which contain the same information. These should be consolidated for 
> simplicity and inter-opt. 



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

[jira] [Commented] (KAFKA-3229) Root statestore is not registered with ProcessorStateManager, inner state store is registered instead

[Tom Dearman (JIRA)]

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

Tom Dearman commented on KAFKA-3229:


[~guozhang] not sure what you mean, there were no new parameters added to 
register(), a StateStore was always passed through just the wrong one. My code 
added a parameter to init() but is this a method that a user would call?

> Root statestore is not registered with ProcessorStateManager, inner state 
> store is registered instead
> -
>
> Key: KAFKA-3229
> URL: https://issues.apache.org/jira/browse/KAFKA-3229
> Project: Kafka
>  Issue Type: Bug
>  Components: kafka streams
>Affects Versions: 0.10.0.0
> Environment: MacOS El Capitan
>Reporter: Tom Dearman
>Assignee: Tom Dearman
> Fix For: 0.10.0.0
>
>
> When the hierarchy of nested StateStores are created, init is called on the 
> root store, but parent StateStores such as  MeteredKeyValueStore just call 
> the contained StateStore until a store such as MemoryStore calls 
> ProcessorContext.register, but it passes 'this' to the method so only that 
> inner state store (MemoryStore in this case) is registered with 
> ProcessorStateManager.  As state is added to the store none of the parent 
> stores code will be called, metering, or even StoreChangeLogger to put the 
> state on the kafka topic.



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

[jira] [Commented] (KAFKA-3296) All consumer reads hang indefinately

[Simon Cooper (JIRA)]

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

Simon Cooper commented on KAFKA-3296:
-

It's definitely a single broker, and a standard single-node zookeeper setup. 
Occasionally we do see warnings from zookeeper in the logs when everything is 
setup - something along the lines of 'fsync took 12000ms, this may adversely 
affect performance' - if ZK takes a long time to send updates/watch 
notifications, could this cause the controller to think it isn't the leader?

We'll turn on controller debug logging for the next time this happens

> All consumer reads hang indefinately
> 
>
> Key: KAFKA-3296
> URL: https://issues.apache.org/jira/browse/KAFKA-3296
> Project: Kafka
>  Issue Type: Bug
>Affects Versions: 0.9.0.0, 0.9.0.1
>Reporter: Simon Cooper
>Priority: Critical
> Attachments: kafkalogs.zip
>
>
> We've got several integration tests that bring up systems on VMs for testing. 
> We've recently upgraded to 0.9, and very occasionally we occasionally see an 
> issue where every consumer that tries to read from the broker hangs, spamming 
> the following in their logs:
> {code}2016-02-26T12:25:37,856 | DEBUG | o.a.k.c.NetworkClient 
> [pool-10-thread-1] | Sending metadata request 
> ClientRequest(expectResponse=true, callback=null, 
> request=RequestSend(header={api_key=3,api_version=0,correlation_id=21905,client_id=consumer-1},
>  body={topics=[Topic1]}), isInitiatedByNetworkClient, 
> createdTimeMs=1456489537856, sendTimeMs=0) to node 1
> 2016-02-26T12:25:37,856 | DEBUG | o.a.k.c.Metadata [pool-10-thread-1] | 
> Updated cluster metadata version 10954 to Cluster(nodes = [Node(1, 
> server.internal, 9092)], partitions = [Partition(topic = Topic1, partition = 
> 0, leader = 1, replicas = [1,], isr = [1,]])
> 2016-02-26T12:25:37,856 | DEBUG | o.a.k.c.c.i.AbstractCoordinator 
> [pool-10-thread-1] | Issuing group metadata request to broker 1
> 2016-02-26T12:25:37,857 | DEBUG | o.a.k.c.c.i.AbstractCoordinator 
> [pool-10-thread-1] | Group metadata response 
> ClientResponse(receivedTimeMs=1456489537857, disconnected=false, 
> request=ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@28edb273,
>  
> request=RequestSend(header={api_key=10,api_version=0,correlation_id=21906,client_id=consumer-1},
>  body={group_id=}), createdTimeMs=1456489537856, sendTimeMs=1456489537856), 
> responseBody={error_code=15,coordinator={node_id=-1,host=,port=-1}})
> 2016-02-26T12:25:37,956 | DEBUG | o.a.k.c.NetworkClient [pool-10-thread-1] | 
> Sending metadata request ClientRequest(expectResponse=true, callback=null, 
> request=RequestSend(header={api_key=3,api_version=0,correlation_id=21907,client_id=consumer-1},
>  body={topics=[Topic1]}), isInitiatedByNetworkClient, 
> createdTimeMs=1456489537956, sendTimeMs=0) to node 1
> 2016-02-26T12:25:37,956 | DEBUG | o.a.k.c.Metadata [pool-10-thread-1] | 
> Updated cluster metadata version 10955 to Cluster(nodes = [Node(1, 
> server.internal, 9092)], partitions = [Partition(topic = Topic1, partition = 
> 0, leader = 1, replicas = [1,], isr = [1,]])
> 2016-02-26T12:25:37,956 | DEBUG | o.a.k.c.c.i.AbstractCoordinator 
> [pool-10-thread-1] | Issuing group metadata request to broker 1
> 2016-02-26T12:25:37,957 | DEBUG | o.a.k.c.c.i.AbstractCoordinator 
> [pool-10-thread-1] | Group metadata response 
> ClientResponse(receivedTimeMs=1456489537957, disconnected=false, 
> request=ClientRequest(expectResponse=true, 
> callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@40cee8cc,
>  
> request=RequestSend(header={api_key=10,api_version=0,correlation_id=21908,client_id=consumer-1},
>  body={group_id=}), createdTimeMs=1456489537956, sendTimeMs=1456489537956), 
> responseBody={error_code=15,coordinator={node_id=-1,host=,port=-1}})
> 2016-02-26T12:25:38,056 | DEBUG | o.a.k.c.NetworkClient [pool-10-thread-1] | 
> Sending metadata request ClientRequest(expectResponse=true, callback=null, 
> request=RequestSend(header={api_key=3,api_version=0,correlation_id=21909,client_id=consumer-1},
>  body={topics=[Topic1]}), isInitiatedByNetworkClient, 
> createdTimeMs=1456489538056, sendTimeMs=0) to node 1
> 2016-02-26T12:25:38,056 | DEBUG | o.a.k.c.Metadata [pool-10-thread-1] | 
> Updated cluster metadata version 10956 to Cluster(nodes = [Node(1, 
> server.internal, 9092)], partitions = [Partition(topic = Topic1, partition = 
> 0, leader = 1, replicas = [1,], isr = [1,]])
> 2016-02-26T12:25:38,056 | DEBUG | o.a.k.c.c.i.AbstractCoordinator 
> [pool-10-thread-1] | Issuing group metadata request to broker 1
> 2016-02-26T12:25:38,057 | DEBUG | o.a.k.c.c.i.AbstractCoordinator 
> [pool-10-thread-1] | Group metadata 

Re: [VOTE] KIP-33 - Add a time based log index to Kafka

[Becket Qin]

Hi Jun,

Rolling out a new segment when the time index is full sounds good. So both
time index and offset index will be sharing the configuration of max index
size.
If we do that, do you think we still want to reuse index.interval.bytes? If
we don't, the risk is that in some corner cases, we might end up with many
small segments. (e.g. small time.index.interval.ms with small max index
size). But this is probably more of a misconfiguration.

2. If the broker is still running when all the segments except the active
segment is deleted, we will have an in memory latest timestamp. So that is
not a problem.

In another case, if a broker boots up and sees only one segment with an
empty time index file, we can scan the active segment and rebuild the time
index.  i.e. we do not need to care about the previous largest timestamp
but simply start over. (We need to scan the active segment because it is
possible that the last message appended to the log has a timestamp not
expired, but the broker died before inserting the time index entry for
it.). If all the messages in the active segment has expired, we should roll
out a new segment and reset the latest timetamp to -1.
The principal here is that we will try to build the time indices for the
existing segments that have not expired. If the message with previously
latest timestamp has already been deleted, there is no need to remember
that any more.

That said, I believe this corner case is really because user is not
configuring the acceptable time difference threshold appropriately.

Thanks,

Jiangjie (Becket) Qin


On Tue, Mar 1, 2016 at 11:55 AM, Jun Rao  wrote:

> Jiangjie,
>
> Currently, we roll a new log segment if the index is full. We can probably
> just do the same on the time index. This will bound the index size.
>
> 1. Sounds good.
>
> 2. I was wondering an edge case where the largest timestamp is in the
> oldest segment and the time index is empty is in all newer segments. At
> some point, we delete the oldest segment since it has expired. Then, we
> delete all but the active segment. Now, what should the largest timestamp
> be? Should it be the previous largest timestamp that we have seen or should
> we dig out the largest timestamp in the active segment?
>
> Thanks,
>
> Jun
>
>
> On Mon, Feb 29, 2016 at 7:29 PM, Becket Qin  wrote:
>
> > Hi Jun,
> >
> > I think index.interval.bytes is used to control the density of the offset
> > index. The counterpart of index.interval.bytes for time index is
> > time.index.interval.ms. If we did not change the semantic of log.roll.ms
> ,
> > log.roll.ms/time.index.interval.ms and
> > log.segment.bytes/index.interval.bytes are a perfect mapping from bytes
> to
> > time. However, because we changed the behavior of log.roll.ms, we need
> to
> > guard against a potentially excessively large time index. We can either
> > reuse index.interval.bytes or introduce time.index.interval.bytes, but I
> > cannot think of additional usage for time.index.interval.bytes other than
> > limiting the time index size.
> >
> > I agree that the memory mapped file is probably not a big issue here and
> we
> > can change the default index size to 2MB.
> >
> > For the two cases you mentioned.
> > 1. Because the message offset in the time index is also monotonically
> > increasing, truncating should be straightforward. i.e. only keep the
> > entries that are pointing to the offsets earlier than the truncated to
> > offsets.
> >
> > 2. The current assumption is that if the time index of a segment is empty
> > and there are no previous time index entry, we will assume that segment
> > should be removed - because all the older segment with even larger
> > timestamp have been removed. So in the case you mentioned, during startup
> > we will remove all the segments and roll out a new empty segment.
> >
> > Thanks,
> >
> > Jiangjie (Becket) Qin
> >
> >
> >
> > On Mon, Feb 29, 2016 at 6:09 PM, Jun Rao  wrote:
> >
> > > Hi, Becket,
> > >
> > > I thought that your proposal to build time-based index just based off
> > > index.interval.bytes
> > > is reasonable. Is there a particular need to also add time.
> > > index.interval.bytes?
> > >
> > > Compute the pre-allocated index file size based on log segment file
> size
> > > can be useful. However, the tricky thing is that log segment size can
> be
> > > changed dynamically. Also, for mmap files, they don't use heap space,
> > just
> > > virtual memory, which will be paged in on demand. So, I am not sure if
> > > memory space is a big concern there. The simplest thing is probably to
> > > change the default index size to 2MB to match the default log segment
> > size.
> > >
> > > A couple of other things to think through.
> > >
> > > 1. Currently, LogSegment supports truncating to an offset. How do we do
> > > that on a time-based index?
> > >
> > > 2. Since it's possible to have a empty time-based index (if all message
> > > timestamps are smaller 

[GitHub] kafka pull request: KAFKA-3311: Prepare internal source topics bef...

[guozhangwang]

GitHub user guozhangwang opened a pull request:

https://github.com/apache/kafka/pull/990

KAFKA-3311: Prepare internal source topics before calling partition grouper



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/guozhangwang/kafka K3311

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/990.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #990


commit b4c45b3d43cb0bdeb144e60c42b3e2d88afbe4f9
Author: Guozhang Wang 
Date:   2016-03-01T22:39:01Z

KAFKA-3311 v1




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-3311) Move co-partition checking to PartitionAssignor and auto-create internal topics

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-3311:
---

GitHub user guozhangwang opened a pull request:

https://github.com/apache/kafka/pull/990

KAFKA-3311: Prepare internal source topics before calling partition grouper



You can merge this pull request into a Git repository by running:

$ git pull https://github.com/guozhangwang/kafka K3311

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/990.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #990


commit b4c45b3d43cb0bdeb144e60c42b3e2d88afbe4f9
Author: Guozhang Wang 
Date:   2016-03-01T22:39:01Z

KAFKA-3311 v1




> Move co-partition checking to PartitionAssignor and auto-create internal 
> topics
> ---
>
> Key: KAFKA-3311
> URL: https://issues.apache.org/jira/browse/KAFKA-3311
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Guozhang Wang
>Assignee: Guozhang Wang
>
> Right now the internal topics management (i.e. re-partition topics for now) 
> is buggy such that it was not auto-created by the assignor. We need to fix 
> this by:
> 1) moving co-paritition info into the assignor.
> 2) let assignor create the internal topics with the right number of 
> partitions as: if co-partitioned, equal to the other partitions; otherwise if 
> there are other source topics in the sub-topology, equal to the maximum of 
> other partitions; otherwise equal to the number of tasks writing to this 
> topic.



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

[jira] [Updated] (KAFKA-2073) Replace TopicMetadata request/response with o.a.k.requests.metadata

[Jason Gustafson (JIRA)]

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

Jason Gustafson updated KAFKA-2073:
---
Status: Patch Available  (was: Open)

> Replace TopicMetadata request/response with o.a.k.requests.metadata
> ---
>
> Key: KAFKA-2073
> URL: https://issues.apache.org/jira/browse/KAFKA-2073
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Gwen Shapira
>Assignee: Jason Gustafson
>
> Replace TopicMetadata request/response with o.a.k.requests.metadata.
> Note, this is more challenging that it appears because while the wire 
> protocol is identical, the objects are completely different.



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

[jira] [Commented] (KAFKA-3289) Update Kafka protocol guide wiki for KIP-31 / KIP-32

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3289:
-

[~kichristensen][~eapache] Yes, it looks the protocol does not have a clear 
change history for all the requests and responses. That is indeed a problem. I 
will update the page.

> Update Kafka protocol guide wiki for KIP-31 / KIP-32
> 
>
> Key: KAFKA-3289
> URL: https://issues.apache.org/jira/browse/KAFKA-3289
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.0.0
>
>




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

Re: Possible bug in Streams → KStreamImpl.java

[Guozhang Wang]

Filed hotfix PR https://github.com/apache/kafka/pull/991.

On Mon, Feb 29, 2016 at 3:34 PM, Guozhang Wang  wrote:

> Hi Avi,
>
> Thanks for pointing it out! And yes it is indeed a bug in the code. Could
> you file a HOTFIX PR fixing this and also modify the existing unit test to
> cover this case?
>
> Thanks,
> Guozhang
>
> On Mon, Feb 29, 2016 at 2:15 PM, Avi Flax  wrote:
>
>> I was just playing around with Streams’ join features, just to get a
>> feel for them, and I think I may have noticed a bug in the code, in
>> KStreamImpl.java on line 310:
>>
>>
>> https://github.com/apache/kafka/blob/845c6eae1f6c6bcf117f5baa53bb19b4611c0528/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamImpl.java#L310
>>
>> (I’m linking to the latest commit that changed this file so that the
>> link will be stable, but line 310 is currently identical in this
>> commit and trunk.)
>>
>> the line reads:
>>
>> .withValues(otherValueSerializer, otherValueDeserializer)
>>
>> but I think maybe it’s supposed to read:
>>
>> .withValues(thisValueSerializer, thisValueDeserializer)
>>
>> I took a look at the tests and it seems they’re not catching this
>> because in the current tests, the serdes for both streams are the same
>> — it might be a good idea to add a test wherein they’re different.
>>
>> If Streams was stable I’d offer to prepare a PR but given that it’s a
>> WIP I figured it would be better to just share this observation.
>>
>> HTH!
>>
>> Avi
>>
>
>
>
> --
> -- Guozhang
>



-- 
-- Guozhang

[jira] [Created] (KAFKA-3313) Update Kafka protocol guide wiki page to reflect the change history of wire protocol and the corresponding Kafka version.

[Jiangjie Qin (JIRA)]

Jiangjie Qin created KAFKA-3313:
---

 Summary: Update Kafka protocol guide wiki page to reflect the 
change history of wire protocol and the corresponding Kafka version.
 Key: KAFKA-3313
 URL: https://issues.apache.org/jira/browse/KAFKA-3313
 Project: Kafka
  Issue Type: Task
Reporter: Jiangjie Qin
Assignee: Jiangjie Qin


Currently the protocol guide does not clearly document the versions and change 
history of the wire protocols. We should make the following thing clear:
1. The change between versions.
2. The supporting Kafka version for different request / response / message 
format versions.



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

[jira] [Created] (KAFKA-3314) Add CDDL license to LICENSE and NOTICE file

[Jun Rao (JIRA)]

Jun Rao created KAFKA-3314:
--

 Summary: Add CDDL license to LICENSE and NOTICE file
 Key: KAFKA-3314
 URL: https://issues.apache.org/jira/browse/KAFKA-3314
 Project: Kafka
  Issue Type: Improvement
Reporter: Jun Rao
Assignee: Jun Rao
 Fix For: 0.10.0.0


Kafka now has a binary dependency on jersey, which is dual licensed under CDDL. 
According to http://www.apache.org/legal/resolved.html#category-a , we need to 
add CDDL to our LICENSE and NOTICE file. 

The discussion on this can be found at 
https://mail-archives.apache.org/mod_mbox/www-legal-discuss/201602.mbox/%3ccafbh0q1u33uog1+xsntens7rzaa5f6hgujczwx03xotug1c...@mail.gmail.com%3E



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

[jira] [Resolved] (KAFKA-2923) Improve 0.9.0 Upgrade Documents

[Gwen Shapira (JIRA)]

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

Gwen Shapira resolved KAFKA-2923.
-
Resolution: Fixed

> Improve 0.9.0 Upgrade Documents 
> 
>
> Key: KAFKA-2923
> URL: https://issues.apache.org/jira/browse/KAFKA-2923
> Project: Kafka
>  Issue Type: Bug
>Reporter: Guozhang Wang
>Assignee: Grant Henke
>  Labels: newbie
> Fix For: 0.10.0.0
>
>
> A couple of places we can improve the upgrade docs:
> 1) Explanation about replica.lag.time.max.ms and how it relates to the old 
> configs.
> 2) Default quota configs.
> 3) Client-server compatibility: old clients working with new servers and new 
> clients working with old servers?



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

Build failed in Jenkins: kafka-trunk-jdk7 #1077

[Apache Jenkins Server]

See 

Changes:

[wangguoz] HOTFIX: Use the correct serde classes

--
[...truncated 5623 lines...]

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testMissingTopic 
PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > readTaskState 
PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > putTaskState 
PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorStateNonRetriableFailure PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorStateShouldOverride PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorStateRetriableFailure PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putSafeOverridesValueSetBySameWorker PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
readConnectorState PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putSafeConnectorIgnoresStaleStatus PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorState PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
putAndGetConnectorStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
putAndGetTaskStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
deleteTaskStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
deleteConnectorStatus PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testWriteFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testWriteNullValueFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testWriteNullKeyFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testNoOffsetsToFlush 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testFlushFailureReplacesOffsets PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testAlreadyFlushing 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelBeforeAwaitFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelAfterAwaitFlush PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testSaveRestore 
PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testGetSet PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testListConnectors PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testListConnectorsNotLeader PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testListConnectorsNotSynced PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testCreateConnector PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testCreateConnectorNotLeader PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testCreateConnectorExists PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testDeleteConnector PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testDeleteConnectorNotLeader PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testDeleteConnectorNotFound PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testGetConnector PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testGetConnectorConfig PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testGetConnectorConfigConnectorNotFound PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testGetConnectorTaskConfigs PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testGetConnectorTaskConfigsConnectorNotFound PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testPutConnectorTaskConfigs PASSED

org.apache.kafka.connect.runtime.rest.resources.ConnectorsResourceTest > 
testPutConnectorTaskConfigsConnectorNotFound PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testCreateConnector PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testJoinAssignment PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testRebalance PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testHaltCleansUpWorker PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 
testCreateConnectorAlreadyExists PASSED

org.apache.kafka.connect.runtime.distributed.DistributedHerderTest > 

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Ashish Singh]

On Tue, Mar 1, 2016 at 6:30 PM, Gwen Shapira  wrote:

> One more thing, the KIP actually had 3 parts:
> 1. The version protocol
> 2. New response on messages of wrong API key or wrong version
> 3. Protocol documentation
>
There is a WIP patch for adding protocol docs,
https://github.com/apache/kafka/pull/970 . By protocol documentation, you
mean updating this, right?

>
> I understand that you are offering to only implement part 1?
> But the KIP discussion and vote should still cover all three parts,
> they will just be implemented in separate JIRA?
>
The patch for KAFKA-3307, https://github.com/apache/kafka/pull/986, covers
1 and 2. KAFKA-3309 tracks documentation part. Yes, we should include all
the three points you mentioned while discussing or voting for KIP-35.

>
> On Tue, Mar 1, 2016 at 5:25 PM, Ashish Singh  wrote:
> > On Tue, Mar 1, 2016 at 5:11 PM, Gwen Shapira  wrote:
> >
> >> I don't see a use for the name - clients should be able to translate
> >> ApiKey to name for any API they support, and I'm not sure why would a
> >> client need to log anything about APIs it does not support. Am I
> >> missing something?
> >>
> > Yea, it is a fair assumption that client would know about APIs it
> supports.
> > It could have been helpful for client users to see new APIs though,
> however
> > users can always refer to protocol doc of new version to find
> corresponding
> > names of the new APIs.
> >
> >>
> >> On a related note, Magnus is currently on vacation, but he should be
> >> back at the end of next week. I'd like to hold off on the vote until
> >> he gets back since his experience in implementing clients  and his
> >> opinions will be very valuable for this discussion.
> >>
> > That is great. It will be valuable to have his feedback. I will hold off
> on
> > removing "api_name" and "api_deprecated_versions" or adding release
> version.
> >
> >>
> >> Gwen
> >>
> >> On Tue, Mar 1, 2016 at 4:24 PM, Ashish Singh 
> wrote:
> >> > Works with me. I will update PR to remove this.
> >> >
> >> > Also, "api_name" have been pointed out as a concern. However, it can
> be
> >> > handy for logging and similar purposes. Any take on that?
> >> >
> >> > On Tue, Mar 1, 2016 at 3:46 PM, Gwen Shapira 
> wrote:
> >> >
> >> >> Jay also mentioned:
> >> >> "Or, alternately, since deprecation has no functional impact and is
> >> >> just a message
> >> >> to developers, we could just leave it out of the protocol and just
> have
> >> it
> >> >> in release notes etc."
> >> >>
> >> >> I'm in favor of leaving it out of the protocol. I can't really see a
> >> >> use-case.
> >> >>
> >> >> Gwen
> >> >>
> >> >> On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh 
> >> wrote:
> >> >>
> >> >> > I hope it is OK for me to make some progress here. I have made the
> >> >> > following changes.
> >> >> >
> >> >> > 1. Updated KIP-35, to adopt Jay's suggestion on maintaining
> separate
> >> list
> >> >> > of deprecated versions, instead of using a version of -1.
> >> >> > 2. Added information on required permissions, Describe action on
> >> Cluster
> >> >> > resource, to be able to retrieve protocol versions from a auth
> enabled
> >> >> > Kafka cluster.
> >> >> >
> >> >> > Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary
> >> patch
> >> >> is
> >> >> > available to review, https://github.com/apache/kafka/pull/986
> >> >> >
> >> >> > On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh  >
> >> >> wrote:
> >> >> >
> >> >> > > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found
> it
> >> >> > really
> >> >> > > difficult to cope up with Kafka releases as they want to support
> >> users
> >> >> on
> >> >> > > different Kafka versions. Capability to retrieve protocol version
> >> will
> >> >> > go a
> >> >> > > long way to ease out those pain points. I will be happy to help
> out
> >> >> with
> >> >> > > the work on this KIP. @Magnus, thanks for driving this, is it OK
> if
> >> I
> >> >> > carry
> >> >> > > forward the work from here. It will be ideal to have this in
> >> 0.10.0.0.
> >> >> > >
> >> >> > > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps 
> >> wrote:
> >> >> > >
> >> >> > >> I wonder if we need to solve the error problem? I think this KIP
> >> >> gives a
> >> >> > >> descent work around.
> >> >> > >>
> >> >> > >> Probably we should have included an error in the response
> header,
> >> but
> >> >> we
> >> >> > >> debated it at the time decided not to and now it is pretty hard
> to
> >> add
> >> >> > >> because the headers aren't versioned (d'oh).
> >> >> > >>
> >> >> > >> It seems like any other solution is going to be kind of a hack,
> >> right?
> >> >> > >> Sending malformed responses back seems like not a clean
> solution...
> >> >> > >>
> >> >> > >> (Not sure if I was pro- having a top-level error or not, but in
> any
> >> >> case
> >> >> > >> the rationale for 

[jira] [Updated] (KAFKA-3314) Add CDDL license to LICENSE and NOTICE file

[Gwen Shapira (JIRA)]

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

Gwen Shapira updated KAFKA-3314:

Priority: Blocker  (was: Major)

> Add CDDL license to LICENSE and NOTICE file
> ---
>
> Key: KAFKA-3314
> URL: https://issues.apache.org/jira/browse/KAFKA-3314
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Jun Rao
>Assignee: Jun Rao
>Priority: Blocker
> Fix For: 0.10.0.0
>
>
> Kafka now has a binary dependency on jersey, which is dual licensed under 
> CDDL. According to http://www.apache.org/legal/resolved.html#category-a , we 
> need to add CDDL to our LICENSE and NOTICE file. 
> The discussion on this can be found at 
> https://mail-archives.apache.org/mod_mbox/www-legal-discuss/201602.mbox/%3ccafbh0q1u33uog1+xsntens7rzaa5f6hgujczwx03xotug1c...@mail.gmail.com%3E



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

[GitHub] kafka pull request: MINOR: Fixed error in test by moving commit va...

[gwenshap]

GitHub user gwenshap opened a pull request:

https://github.com/apache/kafka/pull/994

MINOR: Fixed error in test by moving commit validation

While looking at the failure here:

https://builds.apache.org/job/kafka-trunk-jdk7/ws/connect/runtime/build/reports/tests/classes/org.apache.kafka.connect.runtime.WorkerSourceTaskTest.html

I noticed some stray errors of "Unexpected method call 
SourceTask.commit()", so I moved the location where we expect commits to inside 
the expectedFlush method since this is where the commit happens (on successful 
flush).


You can merge this pull request into a Git repository by running:

$ git pull https://github.com/gwenshap/kafka sourcetask-test-fix

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/994.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #994


commit f0642eea6b67a41dfd2c107cdfcd8a3744ffdb33
Author: Gwen Shapira 
Date:   2016-03-02T06:12:44Z

MINOR: Fixed error in test by moving commit validation




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS] KIP-35 - Retrieve protocol version

[Gwen Shapira]

One more thing, the KIP actually had 3 parts:
1. The version protocol
2. New response on messages of wrong API key or wrong version
3. Protocol documentation

I understand that you are offering to only implement part 1?
But the KIP discussion and vote should still cover all three parts,
they will just be implemented in separate JIRA?

On Tue, Mar 1, 2016 at 5:25 PM, Ashish Singh  wrote:
> On Tue, Mar 1, 2016 at 5:11 PM, Gwen Shapira  wrote:
>
>> I don't see a use for the name - clients should be able to translate
>> ApiKey to name for any API they support, and I'm not sure why would a
>> client need to log anything about APIs it does not support. Am I
>> missing something?
>>
> Yea, it is a fair assumption that client would know about APIs it supports.
> It could have been helpful for client users to see new APIs though, however
> users can always refer to protocol doc of new version to find corresponding
> names of the new APIs.
>
>>
>> On a related note, Magnus is currently on vacation, but he should be
>> back at the end of next week. I'd like to hold off on the vote until
>> he gets back since his experience in implementing clients  and his
>> opinions will be very valuable for this discussion.
>>
> That is great. It will be valuable to have his feedback. I will hold off on
> removing "api_name" and "api_deprecated_versions" or adding release version.
>
>>
>> Gwen
>>
>> On Tue, Mar 1, 2016 at 4:24 PM, Ashish Singh  wrote:
>> > Works with me. I will update PR to remove this.
>> >
>> > Also, "api_name" have been pointed out as a concern. However, it can be
>> > handy for logging and similar purposes. Any take on that?
>> >
>> > On Tue, Mar 1, 2016 at 3:46 PM, Gwen Shapira  wrote:
>> >
>> >> Jay also mentioned:
>> >> "Or, alternately, since deprecation has no functional impact and is
>> >> just a message
>> >> to developers, we could just leave it out of the protocol and just have
>> it
>> >> in release notes etc."
>> >>
>> >> I'm in favor of leaving it out of the protocol. I can't really see a
>> >> use-case.
>> >>
>> >> Gwen
>> >>
>> >> On Mon, Feb 29, 2016 at 3:55 PM, Ashish Singh 
>> wrote:
>> >>
>> >> > I hope it is OK for me to make some progress here. I have made the
>> >> > following changes.
>> >> >
>> >> > 1. Updated KIP-35, to adopt Jay's suggestion on maintaining separate
>> list
>> >> > of deprecated versions, instead of using a version of -1.
>> >> > 2. Added information on required permissions, Describe action on
>> Cluster
>> >> > resource, to be able to retrieve protocol versions from a auth enabled
>> >> > Kafka cluster.
>> >> >
>> >> > Created https://issues.apache.org/jira/browse/KAFKA-3304. Primary
>> patch
>> >> is
>> >> > available to review, https://github.com/apache/kafka/pull/986
>> >> >
>> >> > On Thu, Feb 25, 2016 at 1:27 PM, Ashish Singh 
>> >> wrote:
>> >> >
>> >> > > Kafka clients in Hadoop ecosystem, Flume, Spark, etc, have found it
>> >> > really
>> >> > > difficult to cope up with Kafka releases as they want to support
>> users
>> >> on
>> >> > > different Kafka versions. Capability to retrieve protocol version
>> will
>> >> > go a
>> >> > > long way to ease out those pain points. I will be happy to help out
>> >> with
>> >> > > the work on this KIP. @Magnus, thanks for driving this, is it OK if
>> I
>> >> > carry
>> >> > > forward the work from here. It will be ideal to have this in
>> 0.10.0.0.
>> >> > >
>> >> > > On Mon, Oct 12, 2015 at 9:29 PM, Jay Kreps 
>> wrote:
>> >> > >
>> >> > >> I wonder if we need to solve the error problem? I think this KIP
>> >> gives a
>> >> > >> descent work around.
>> >> > >>
>> >> > >> Probably we should have included an error in the response header,
>> but
>> >> we
>> >> > >> debated it at the time decided not to and now it is pretty hard to
>> add
>> >> > >> because the headers aren't versioned (d'oh).
>> >> > >>
>> >> > >> It seems like any other solution is going to be kind of a hack,
>> right?
>> >> > >> Sending malformed responses back seems like not a clean solution...
>> >> > >>
>> >> > >> (Not sure if I was pro- having a top-level error or not, but in any
>> >> case
>> >> > >> the rationale for the decision was that so many of the requests
>> were
>> >> > >> per-partition or per-topic or whatever and hence fail or succeed at
>> >> that
>> >> > >> level and this makes it hard to know what the right top-level error
>> >> code
>> >> > >> is
>> >> > >> and hard for the client to figure out what to do with the top level
>> >> > error
>> >> > >> if some of the partitions succeed but there is a top-level error).
>> >> > >>
>> >> > >> I think actually this new API actually gives a way to handle this
>> >> > >> gracefully on the client side by just having clients that want to
>> be
>> >> > >> graceful check for support for their version. Clients that do that
>> >> will
>> >> > >> have a graceful 

[GitHub] kafka pull request: MINOR: Move streams-exmaples source files unde...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/992


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-2944) NullPointerException in KafkaConfigStorage when config storage starts right before shutdown request

[Gwen Shapira (JIRA)]

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

Gwen Shapira commented on KAFKA-2944:
-

Created PR https://github.com/apache/kafka/pull/993 for partial improvement 
here.

> NullPointerException in KafkaConfigStorage when config storage starts right 
> before shutdown request
> ---
>
> Key: KAFKA-2944
> URL: https://issues.apache.org/jira/browse/KAFKA-2944
> Project: Kafka
>  Issue Type: Bug
>  Components: copycat
>Affects Versions: 0.9.0.0
>Reporter: Ewen Cheslack-Postava
>Assignee: Ewen Cheslack-Postava
>
> Relevant log where you can see a config update starting, then the request to 
> shutdown happens and we end up with a NullPointerException:
> {quote}
> [2015-12-03 09:12:55,712] DEBUG Change in connector task count from 2 to 3, 
> writing updated task configurations 
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
> [2015-12-03 09:12:56,224] INFO Kafka Connect stopping 
> (org.apache.kafka.connect.runtime.Connect)
> [2015-12-03 09:12:56,224] INFO Stopping REST server 
> (org.apache.kafka.connect.runtime.rest.RestServer)
> [2015-12-03 09:12:56,227] INFO Stopped 
> ServerConnector@10cb550e{HTTP/1.1}{0.0.0.0:8083} 
> (org.eclipse.jetty.server.ServerConnector)
> [2015-12-03 09:12:56,234] INFO Stopped 
> o.e.j.s.ServletContextHandler@3f8a24d5{/,null,UNAVAILABLE} 
> (org.eclipse.jetty.server.handler.ContextHandler)
> [2015-12-03 09:12:56,235] INFO REST server stopped 
> (org.apache.kafka.connect.runtime.rest.RestServer)
> [2015-12-03 09:12:56,235] INFO Herder stopping 
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
> [2015-12-03 09:12:58,209] ERROR Unexpected exception in KafkaBasedLog's work 
> thread (org.apache.kafka.connect.util.KafkaBasedLog)
> java.lang.NullPointerException
>   at 
> org.apache.kafka.connect.storage.KafkaConfigStorage.completeTaskIdSet(KafkaConfigStorage.java:558)
>   at 
> org.apache.kafka.connect.storage.KafkaConfigStorage.access$1200(KafkaConfigStorage.java:143)
>   at 
> org.apache.kafka.connect.storage.KafkaConfigStorage$1.onCompletion(KafkaConfigStorage.java:476)
>   at 
> org.apache.kafka.connect.storage.KafkaConfigStorage$1.onCompletion(KafkaConfigStorage.java:372)
>   at 
> org.apache.kafka.connect.util.KafkaBasedLog.poll(KafkaBasedLog.java:235)
>   at 
> org.apache.kafka.connect.util.KafkaBasedLog.readToLogEnd(KafkaBasedLog.java:275)
>   at 
> org.apache.kafka.connect.util.KafkaBasedLog.access$300(KafkaBasedLog.java:70)
>   at 
> org.apache.kafka.connect.util.KafkaBasedLog$WorkThread.run(KafkaBasedLog.java:307)
> [2015-12-03 09:13:26,704] ERROR Failed to write root configuration to Kafka:  
> (org.apache.kafka.connect.storage.KafkaConfigStorage)
> java.util.concurrent.TimeoutException: Timed out waiting for future
>   at 
> org.apache.kafka.connect.util.ConvertingFutureCallback.get(ConvertingFutureCallback.java:74)
>   at 
> org.apache.kafka.connect.storage.KafkaConfigStorage.putTaskConfigs(KafkaConfigStorage.java:352)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnector(DistributedHerder.java:737)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnectorTasksWithRetry(DistributedHerder.java:677)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:673)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startWork(DistributedHerder.java:640)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.handleRebalanceCompleted(DistributedHerder.java:598)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:184)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:159)
>   at java.lang.Thread.run(Thread.java:745)
> [2015-12-03 09:13:26,704] ERROR Failed to reconfigure connector's tasks, 
> retrying after backoff: 
> (org.apache.kafka.connect.runtime.distributed.DistributedHerder)
> org.apache.kafka.connect.errors.ConnectException: Error writing root 
> configuration to Kafka
>   at 
> org.apache.kafka.connect.storage.KafkaConfigStorage.putTaskConfigs(KafkaConfigStorage.java:355)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnector(DistributedHerder.java:737)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.reconfigureConnectorTasksWithRetry(DistributedHerder.java:677)
>   at 
> org.apache.kafka.connect.runtime.distributed.DistributedHerder.startConnector(DistributedHerder.java:673)
>   at 
> 

Re: [VOTE] KIP-33 - Add a time based log index to Kafka

[Becket Qin]

Hi Jun,

I see. If we only use index.interval.bytes, the time index entry will be
inserted when (1) the largest timestamp is in this segment AND (2) at least
index.interval.bytes have been appended since last time index entry
insertion.
In this case (1) becomes implicit instead of having an explicit threshold
of time.index.interval.ms. This should work fine.

For 1, the current proposal is actually intended to use the offsets of the
messages instead of the file position. The reason is that
OffsetRequest(ListOffsetRequest) gives back a list of offsets. Having
offsets instead of file position is more convenient in that case. So far we
don't have an interface for consumer to directly consume from a given
timestamp. Supposedly we will have to first return the offset to the
consumer then the consumer can issue a fetch request. But this does mean
that we need to look up the index twice if we want to search to a
particular message using timestamp.

For 2, that is a good point. It looks we have to persist the max timestamp
of each segment. I am thinking of reserving the first time index entry in
each time index file. When broker shuts down or rolls out a new segment, we
persist the max timestamp in the segment by writing a time index entry to
the first entry. During timestamp search we will ignore the first time
index entry. So the log retention will always be able to know for sure if a
log segment is supposed to be deleted or not by looking at the first entry
of time index.

Thanks,

Jiangjie (Becket) Qin




On Tue, Mar 1, 2016 at 4:30 PM, Jun Rao  wrote:

> Hi, Jiangjie,
>
> I was thinking perhaps just reusing index.interval.bytes is enough. Not
> sure if there is much value in adding an additional time.index.interval.ms
> .
>
> For 1, the timestamp index has entries of timestamp -> file position. So,
> there is actually no offset in the index, right?
>
> For 2, what you said makes sense for time-based retention. Does that apply
> if the retention is trigged by size? The difference here is that we can't
> assume all segments with messages of timestamp smaller than the latest
> timestamp will be deleted after the message with the latest timestamp is
> deleted.
>
> Thanks,
>
> Jun
>
> On Tue, Mar 1, 2016 at 1:00 PM, Becket Qin  wrote:
>
> > Hi Jun,
> >
> > Rolling out a new segment when the time index is full sounds good. So
> both
> > time index and offset index will be sharing the configuration of max
> index
> > size.
> > If we do that, do you think we still want to reuse index.interval.bytes?
> If
> > we don't, the risk is that in some corner cases, we might end up with
> many
> > small segments. (e.g. small time.index.interval.ms with small max index
> > size). But this is probably more of a misconfiguration.
> >
> > 2. If the broker is still running when all the segments except the active
> > segment is deleted, we will have an in memory latest timestamp. So that
> is
> > not a problem.
> >
> > In another case, if a broker boots up and sees only one segment with an
> > empty time index file, we can scan the active segment and rebuild the
> time
> > index.  i.e. we do not need to care about the previous largest timestamp
> > but simply start over. (We need to scan the active segment because it is
> > possible that the last message appended to the log has a timestamp not
> > expired, but the broker died before inserting the time index entry for
> > it.). If all the messages in the active segment has expired, we should
> roll
> > out a new segment and reset the latest timetamp to -1.
> > The principal here is that we will try to build the time indices for the
> > existing segments that have not expired. If the message with previously
> > latest timestamp has already been deleted, there is no need to remember
> > that any more.
> >
> > That said, I believe this corner case is really because user is not
> > configuring the acceptable time difference threshold appropriately.
> >
> > Thanks,
> >
> > Jiangjie (Becket) Qin
> >
> >
> > On Tue, Mar 1, 2016 at 11:55 AM, Jun Rao  wrote:
> >
> > > Jiangjie,
> > >
> > > Currently, we roll a new log segment if the index is full. We can
> > probably
> > > just do the same on the time index. This will bound the index size.
> > >
> > > 1. Sounds good.
> > >
> > > 2. I was wondering an edge case where the largest timestamp is in the
> > > oldest segment and the time index is empty is in all newer segments. At
> > > some point, we delete the oldest segment since it has expired. Then, we
> > > delete all but the active segment. Now, what should the largest
> timestamp
> > > be? Should it be the previous largest timestamp that we have seen or
> > should
> > > we dig out the largest timestamp in the active segment?
> > >
> > > Thanks,
> > >
> > > Jun
> > >
> > >
> > > On Mon, Feb 29, 2016 at 7:29 PM, Becket Qin 
> > wrote:
> > >
> > > > Hi Jun,
> > > >
> > > > I think index.interval.bytes is used to 

Build failed in Jenkins: kafka-trunk-jdk8 #404

[Apache Jenkins Server]

See 

Changes:

[wangguoz] HOTFIX: Use the correct serde classes

--
[...truncated 5045 lines...]

org.apache.kafka.streams.processor.internals.ProcessorStateManagerTest > 
testNoTopic PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > 
testStorePartitions PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > testUpdateKTable 
PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > 
testUpdateNonPersistentStore PASSED

org.apache.kafka.streams.processor.internals.StandbyTaskTest > testUpdate PASSED

org.apache.kafka.streams.processor.internals.StreamThreadTest > testMaybeCommit 
PASSED

org.apache.kafka.streams.processor.internals.StreamThreadTest > 
testPartitionAssignmentChange PASSED

org.apache.kafka.streams.processor.internals.StreamThreadTest > testMaybeClean 
PASSED

org.apache.kafka.streams.processor.internals.QuickUnionTest > testUnite PASSED

org.apache.kafka.streams.processor.internals.QuickUnionTest > testUniteMany 
PASSED

org.apache.kafka.streams.processor.internals.MinTimestampTrackerTest > 
testTracking PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingMultiplexingTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingStatefulTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testDrivingSimpleTopology PASSED

org.apache.kafka.streams.processor.internals.ProcessorTopologyTest > 
testTopologyMetadata PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithStandbyReplicas PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithNewTasks PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignWithStates PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testAssignBasic PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testOnAssignment PASSED

org.apache.kafka.streams.processor.internals.StreamPartitionAssignorTest > 
testSubscription PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > testProcessOrder 
PASSED

org.apache.kafka.streams.processor.internals.StreamTaskTest > testPauseResume 
PASSED

org.apache.kafka.streams.processor.internals.RecordQueueTest > testTimeTracking 
PASSED

org.apache.kafka.streams.processor.internals.assignment.AssginmentInfoTest > 
testEncodeDecode PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testStickiness PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testAssignWithStandby PASSED

org.apache.kafka.streams.processor.internals.assignment.TaskAssignorTest > 
testAssignWithoutStandby PASSED

org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfoTest > 
testEncodeDecode PASSED

org.apache.kafka.streams.processor.internals.PartitionGroupTest > 
testTimeTracking PASSED

org.apache.kafka.streams.processor.internals.PunctuationQueueTest > 
testPunctuationInterval PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > testSourceTopics PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkWithSameName PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkWithSelfParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddProcessorWithSelfParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithSink PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > testTopicGroups PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > testBuild PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithSource PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSourceWithSameName PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddProcessorWithSameName PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSourceWithSameTopic PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testTopicGroupsByStateStore PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithDuplicates PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkWithWrongParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkConnectedWithMultipleParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddProcessorWithWrongParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > testAddStateStore 
PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddSinkConnectedWithParent PASSED

org.apache.kafka.streams.processor.TopologyBuilderTest > 
testAddStateStoreWithNonExistingProcessor PASSED

Build failed in Jenkins: kafka-trunk-jdk8 #405

[Apache Jenkins Server]

See 

Changes:

[me] MINOR: Move streams-examples source files under src folder

--
[...truncated 5713 lines...]

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testStartStop 
PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > 
testReloadOnStart PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testMissingTopic 
PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testGetSet PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testSetFailure 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testWriteFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testWriteNullValueFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testWriteNullKeyFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testNoOffsetsToFlush 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testFlushFailureReplacesOffsets PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testAlreadyFlushing 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelBeforeAwaitFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelAfterAwaitFlush PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testSaveRestore 
PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testGetSet PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > testStartStop PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > testPutTaskConfigs 
PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > testRestore PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > 
testPutTaskConfigsDoesNotResolveAllInconsistencies PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
putAndGetConnectorStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
putAndGetTaskStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
deleteTaskStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
deleteConnectorStatus PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddRemoveTask PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStopInvalidTask PASSED

org.apache.kafka.connect.runtime.WorkerTest > testCleanupTasksOnStop PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStartAndStopConnector PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddConnectorByAlias PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddConnectorByShortAlias 
PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStopInvalidConnector PASSED

org.apache.kafka.connect.runtime.WorkerTest > testReconfigureConnectorTasks 
PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > testPollRedelivery PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > 
testErrorInRebalancePartitionRevocation PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > 
testErrorInRebalancePartitionAssignment PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > stopBeforeStarting PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > standardStartup PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > cancelBeforeStopping PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testPollsInBackground 
FAILED
java.lang.AssertionError: 
  Expectation failure on verify:
Listener.onStartup(job-0): expected: 1, actual: 1
Listener.onShutdown(job-0): expected: 1, actual: 0
at org.easymock.internal.MocksControl.verify(MocksControl.java:225)
at 
org.powermock.api.easymock.internal.invocationcontrol.EasyMockMethodInvocationControl.verify(EasyMockMethodInvocationControl.java:132)
at org.powermock.api.easymock.PowerMock.verify(PowerMock.java:1466)
at org.powermock.api.easymock.PowerMock.verifyAll(PowerMock.java:1405)
at 
org.apache.kafka.connect.runtime.WorkerSourceTaskTest.testPollsInBackground(WorkerSourceTaskTest.java:151)

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testFailureInPoll PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testCommit FAILED
java.lang.AssertionError: 
  Expectation failure on verify:
Listener.onStartup(job-0): expected: 1, actual: 1
Listener.onShutdown(job-0): expected: 1, actual: 0
at org.easymock.internal.MocksControl.verify(MocksControl.java:225)
at 
org.powermock.api.easymock.internal.invocationcontrol.EasyMockMethodInvocationControl.verify(EasyMockMethodInvocationControl.java:132)
at org.powermock.api.easymock.PowerMock.verify(PowerMock.java:1466)
at 

Build failed in Jenkins: kafka-trunk-jdk7 #1078

[Apache Jenkins Server]

See 

Changes:

[me] MINOR: Move streams-examples source files under src folder

--
[...truncated 5610 lines...]

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > readTaskState 
PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > putTaskState 
PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorStateNonRetriableFailure PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorStateShouldOverride PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorStateRetriableFailure PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putSafeOverridesValueSetBySameWorker PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
readConnectorState PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putSafeConnectorIgnoresStaleStatus PASSED

org.apache.kafka.connect.storage.KafkaStatusBackingStoreTest > 
putConnectorState PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testStartStop 
PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > 
testReloadOnStart PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testMissingTopic 
PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testGetSet PASSED

org.apache.kafka.connect.storage.KafkaOffsetBackingStoreTest > testSetFailure 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testWriteFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testWriteNullValueFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testWriteNullKeyFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testNoOffsetsToFlush 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testFlushFailureReplacesOffsets PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > testAlreadyFlushing 
PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelBeforeAwaitFlush PASSED

org.apache.kafka.connect.storage.OffsetStorageWriterTest > 
testCancelAfterAwaitFlush PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testSaveRestore 
PASSED

org.apache.kafka.connect.storage.FileOffsetBackingStoreTest > testGetSet PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > testStartStop PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > 
testPutConnectorConfig PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > testPutTaskConfigs 
PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > testRestore PASSED

org.apache.kafka.connect.storage.KafkaConfigStorageTest > 
testPutTaskConfigsDoesNotResolveAllInconsistencies PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
putAndGetConnectorStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
putAndGetTaskStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
deleteTaskStatus PASSED

org.apache.kafka.connect.storage.MemoryStatusBackingStoreTest > 
deleteConnectorStatus PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStartAndStopConnector PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStopInvalidTask PASSED

org.apache.kafka.connect.runtime.WorkerTest > testCleanupTasksOnStop PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddConnectorByAlias PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddConnectorByShortAlias 
PASSED

org.apache.kafka.connect.runtime.WorkerTest > testStopInvalidConnector PASSED

org.apache.kafka.connect.runtime.WorkerTest > testReconfigureConnectorTasks 
PASSED

org.apache.kafka.connect.runtime.WorkerTest > testAddRemoveTask PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > 
testErrorInRebalancePartitionRevocation PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > 
testErrorInRebalancePartitionAssignment PASSED

org.apache.kafka.connect.runtime.WorkerSinkTaskTest > testPollRedelivery PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > stopBeforeStarting PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > standardStartup PASSED

org.apache.kafka.connect.runtime.WorkerTaskTest > cancelBeforeStopping PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testPollsInBackground 
PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testFailureInPoll PASSED

org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testCommit FAILED
java.lang.AssertionError: 
  Expectation failure on verify:
Listener.onStartup(job-0): expected: 1, actual: 1
Listener.onShutdown(job-0): expected: 1, actual: 0
at org.easymock.internal.MocksControl.verify(MocksControl.java:225)
at 

[GitHub] kafka pull request: Replaced the NPE with a nicer error and clean ...

[gwenshap]

GitHub user gwenshap opened a pull request:

https://github.com/apache/kafka/pull/993

Replaced the NPE with a nicer error and clean exit and added debug me…

…ssage to assist with figuring this out.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/gwenshap/kafka KAFKA-2944

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/993.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #993


commit 19c611fe899e8b2034357d744cfab2063aa42f8a
Author: Gwen Shapira 
Date:   2016-03-02T03:45:47Z

Replaced the NPE with a nicer error and clean exit and added debug message 
to assist with figuring this out.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Ismael Juma]

Also, with regards to the client flow, it says:

"If sasl.mechanism is not GSSAPI, send a packet with the mechanism name to
the server. Otherwise go to Step 3."

It sounds like it would be more regular and simpler for clients if they
always sent the mechanism, even if GSSAPI, right? The currently proposed
way has the benefit that newer clients would support older brokers _if_ the
sasl.mechanism was GSSAPI. Is this important? For Kafka, brokers have to be
upgraded before clients, generally. Brokers would still support older
clients that didn't send the mechanism either way.

Ismael


On Tue, Mar 1, 2016 at 3:11 AM, Ismael Juma  wrote:

> Hi Rajini,
>
> One question below.
>
> On Tue, Mar 1, 2016 at 2:47 AM, Rajini Sivaram <
> rajinisiva...@googlemail.com> wrote:
>
>>1. With GSSAPI, the first context establishment packet starts with the
>>byte 0x60 (tag for APPLICATION-0) followed by a variable-length encoded
>>size, followed by various tags and contents. And the packet also
>> contains a
>>checksum. This is completely different from the mechanism packet from
>> Kafka
>>clients which start with a two-byte version set to zero currently,
>> followed
>>by just a String mechanism.
>>
>
> Would it be better to assign an id to each mechanism and pass that instead
> of the String? That would be more space-efficient.
>
> Ismael
>

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Rajini Sivaram]

Jun,

Thank you for the review.


   1. With GSSAPI, the first context establishment packet starts with the
   byte 0x60 (tag for APPLICATION-0) followed by a variable-length encoded
   size, followed by various tags and contents. And the packet also contains a
   checksum. This is completely different from the mechanism packet from Kafka
   clients which start with a two-byte version set to zero currently, followed
   by just a String mechanism.
   2. Agreed, I have removed the version from the server response in the
   KIP. Thanks.


On Tue, Mar 1, 2016 at 2:33 AM, Jun Rao  wrote:

> Rajini,
>
> Thanks for the updates. Just a couple of minor comments.
>
> 1. With the default GSSAPI, what's the first packet that the client sends
> to the server? Is that completely different from the packet format that we
> will use for non-GSSAPI mechanisms?
>
> 2. In the server response, it doesn't seem that we need to include the
> version since the client knows the version of the request that it sends.
>
> Jun
>
> On Mon, Feb 29, 2016 at 10:14 AM, Rajini Sivaram <
> rajinisiva...@googlemail.com> wrote:
>
> > Harsha,
> >
> > Thank you for the review. I will wait another day to see if there is more
> > feedback and then start a voting thread.
> >
> > Rajini
> >
> > On Mon, Feb 29, 2016 at 2:51 PM, Harsha  wrote:
> >
> > > Rajini,
> > >   Thanks for the changes to the KIP. It looks good to me. I
> > >   think we can move to voting.
> > > Thanks,
> > > Harsha
> > >
> > > On Mon, Feb 29, 2016, at 12:43 AM, Rajini Sivaram wrote:
> > > > I have added some more detail to the KIP based on the discussion in
> the
> > > > last KIP meeting to simplify support for multiple mechanisms. Have
> also
> > > > changed the property names to reflect this.
> > > >
> > > > Also updated the PR in
> > https://issues.apache.org/jira/browse/KAFKA-3149
> > > > to
> > > > reflect the KIP.
> > > >
> > > > Any feedback is appreciated.
> > > >
> > > >
> > > > On Tue, Feb 23, 2016 at 9:36 PM, Rajini Sivaram <
> > > > rajinisiva...@googlemail.com> wrote:
> > > >
> > > > > I have updated the KIP based on the discussion in the KIP meeting
> > > today.
> > > > >
> > > > > Comments and feedback are welcome.
> > > > >
> > > > > On Wed, Feb 3, 2016 at 7:20 PM, Rajini Sivaram <
> > > > > rajinisiva...@googlemail.com> wrote:
> > > > >
> > > > >> Hi Harsha,
> > > > >>
> > > > >> Thank you for the review. Can you clarify - I think you are saying
> > > that
> > > > >> the client should send its mechanism over the wire to the server.
> Is
> > > that
> > > > >> correct? The exchange is slightly different in the KIP (the PR
> > > matches the
> > > > >> KIP) from the one you described to enable interoperability with
> > > 0.9.0.0.
> > > > >>
> > > > >>
> > > > >> On Wed, Feb 3, 2016 at 1:56 PM, Harsha  wrote:
> > > > >>
> > > > >>> Rajini,
> > > > >>>I looked at the PR you have. I think its better with
> > your
> > > > >>>earlier approach rather than extending the protocol.
> > > > >>> What I was thinking initially is, Broker has a config option of
> say
> > > > >>> sasl.mechanism = GSSAPI, PLAIN
> > > > >>> and the client can have similar config of sasl.mechanism=PLAIN.
> > > Client
> > > > >>> can send its sasl mechanism before the handshake starts and if
> the
> > > > >>> broker accepts that particular mechanism than it can go ahead
> with
> > > > >>> handshake otherwise return a error saying that the mechanism not
> > > > >>> allowed.
> > > > >>>
> > > > >>> Thanks,
> > > > >>> Harsha
> > > > >>>
> > > > >>> On Wed, Feb 3, 2016, at 04:58 AM, Rajini Sivaram wrote:
> > > > >>> > A slightly different approach for supporting different SASL
> > > mechanisms
> > > > >>> > within a broker is to allow the same "*security protocol*" to
> be
> > > used
> > > > >>> on
> > > > >>> > different ports with different configuration options. An
> > advantage
> > > of
> > > > >>> > this
> > > > >>> > approach is that it extends the configurability of not just
> SASL,
> > > but
> > > > >>> any
> > > > >>> > protocol. For instance, it would enable the use of SSL with
> > mutual
> > > > >>> client
> > > > >>> > authentication on one port or different certificate chains on
> > > another.
> > > > >>> > And
> > > > >>> > it avoids the need for SASL mechanism negotiation.
> > > > >>> >
> > > > >>> > Kafka would have the same "*security protocols" *defined as
> > today,
> > > but
> > > > >>> > with
> > > > >>> > (a single) configurable SASL mechanism. To have different
> > > > >>> configurations
> > > > >>> > of
> > > > >>> > a protocol within a broker, users can define new protocol names
> > > which
> > > > >>> are
> > > > >>> > configured versions of existing protocols, perhaps using just
> > > > >>> > configuration
> > > > >>> > entries and no additional code.
> > > > >>> >
> > > > >>> > For example:
> > > > >>> >
> > > > >>> > A single mechanism broker would be configured as:
> > > > 

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Ismael Juma]

Hi Rajini,

One question below.

On Tue, Mar 1, 2016 at 2:47 AM, Rajini Sivaram  wrote:

>1. With GSSAPI, the first context establishment packet starts with the
>byte 0x60 (tag for APPLICATION-0) followed by a variable-length encoded
>size, followed by various tags and contents. And the packet also
> contains a
>checksum. This is completely different from the mechanism packet from
> Kafka
>clients which start with a two-byte version set to zero currently,
> followed
>by just a String mechanism.
>

Would it be better to assign an id to each mechanism and pass that instead
of the String? That would be more space-efficient.

Ismael

[GitHub] kafka pull request: MINOR: use Vector instead of List

[xuwei-k]

GitHub user xuwei-k opened a pull request:

https://github.com/apache/kafka/pull/995

MINOR: use Vector instead of List

`Vector#:+` is more efficient rather than `List#++` in this case.

You can merge this pull request into a Git repository by running:

$ git pull https://github.com/xuwei-k/kafka KafkaServer-List-Vector

Alternatively you can review and apply these changes as the patch at:

https://github.com/apache/kafka/pull/995.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

This closes #995


commit 1e8802c6f23dce657624e1def16184768ef8d2ba
Author: xuwei-k <6b656e6...@gmail.com>
Date:   2016-03-02T06:35:34Z

MINOR: use Vector instead of List

`Vector#:+` is more efficient rather than `List#++` in this case.




---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Commented] (KAFKA-3290) WorkerSourceTask testCommit transient failure

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava commented on KAFKA-3290:
--

[~hachikuji] Did you reproduce this locally? This is happening a lot on 
Confluent's jenkins if we need a place we can reproduce it regularly: 
https://jenkins.confluent.io/job/kafka-trunk/

> WorkerSourceTask testCommit transient failure
> -
>
> Key: KAFKA-3290
> URL: https://issues.apache.org/jira/browse/KAFKA-3290
> Project: Kafka
>  Issue Type: Sub-task
>  Components: copycat
>Reporter: Jason Gustafson
>Assignee: Jason Gustafson
>
> From recent failed build:
> {code}
> org.apache.kafka.connect.runtime.WorkerSourceTaskTest > testCommit FAILED
> java.lang.AssertionError:
>   Expectation failure on verify:
> Listener.onStartup(job-0): expected: 1, actual: 1
> Listener.onShutdown(job-0): expected: 1, actual: 1
> at org.easymock.internal.MocksControl.verify(MocksControl.java:225)
> at 
> org.powermock.api.easymock.internal.invocationcontrol.EasyMockMethodInvocationControl.verify(EasyMockMethodInvocationControl.java:132)
> at org.powermock.api.easymock.PowerMock.verify(PowerMock.java:1466)
> at org.powermock.api.easymock.PowerMock.verifyAll(PowerMock.java:1405)
> at 
> org.apache.kafka.connect.runtime.WorkerSourceTaskTest.testCommit(WorkerSourceTaskTest.java:221)
> {code}



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

[jira] [Created] (KAFKA-3316) Add Connect REST API to list available connector classes

[Ewen Cheslack-Postava (JIRA)]

Ewen Cheslack-Postava created KAFKA-3316:


 Summary: Add Connect REST API to list available connector classes
 Key: KAFKA-3316
 URL: https://issues.apache.org/jira/browse/KAFKA-3316
 Project: Kafka
  Issue Type: Bug
  Components: copycat
Reporter: Ewen Cheslack-Postava
Assignee: Ewen Cheslack-Postava
 Fix For: 0.10.0.0


Each worker process's REST API should have an endpoint that can list available 
connector classes. This can use the same Reflections code as we used in 
KAFKA-2422 to find matching connector classes based on a short name. This is 
useful both for debugging and for any systems that want to work with different 
connect clusters and be able to tell which clusters support which connectors.

We may need a new top-level resource to support this. We have /connectors 
already, but that refers to instantiated connectors that have been named. In 
contrast, this resource would refer to the connector classes (uninstantiated). 
We might be able to use the same resource to, e.g., lookup config info in 
KAFKA-3315 (which occurs before connector instantiation).



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

[jira] [Created] (KAFKA-3315) Add Connect API to expose connector configuration info

[Ewen Cheslack-Postava (JIRA)]

Ewen Cheslack-Postava created KAFKA-3315:


 Summary: Add Connect API to expose connector configuration info
 Key: KAFKA-3315
 URL: https://issues.apache.org/jira/browse/KAFKA-3315
 Project: Kafka
  Issue Type: Bug
  Components: copycat
Reporter: Ewen Cheslack-Postava
Assignee: Liquan Pei
 Fix For: 0.10.0.0


Connectors should be able to provide information about how they can be 
configured. It will be nice to expose this programmatically as part of the 
standard interface for connectors. This can also include support for more than 
just a static set of config options. For example, a validation REST API could 
provide intermediate feedback based on a partial configuration and include 
recommendations/suggestions for fields based on the settings available so far.



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

[jira] [Commented] (KAFKA-3257) bootstrap-test-env.sh version check fails when grep has --colour option enabled.

[ASF GitHub Bot (JIRA)]

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

ASF GitHub Bot commented on KAFKA-3257:
---

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/969


> bootstrap-test-env.sh version check fails when grep has --colour option 
> enabled.
> 
>
> Key: KAFKA-3257
> URL: https://issues.apache.org/jira/browse/KAFKA-3257
> Project: Kafka
>  Issue Type: Bug
>  Components: system tests
>Affects Versions: 0.9.0.1
>Reporter: Jiangjie Qin
>Assignee: chen zhu
>  Labels: newbie++
> Fix For: 0.10.0.0
>
>
> When checking the versions, we use the following command:
> {code}
> vagrant --version | egrep -o "[0-9]+\.[0-9]+\.[0-9]+"
> {code}
> This does not work if user box has --colour option enabled. In my case it 
> complains:
> Found Vagrant version 1.8.1. Please upgrade to 1.6.4 or higher (see 
> http://www.vagrantup.com for details)
> We should change this line to:
> {code}
> vagrant --version | egrep --colour=never -o "[0-9]+\.[0-9]+\.[0-9]+"
> {code}



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

[GitHub] kafka pull request: KAFKA-3257: disable bootstrap-test-env.sh --co...

[asfgit]

Github user asfgit closed the pull request at:

https://github.com/apache/kafka/pull/969


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---

[jira] [Resolved] (KAFKA-3257) bootstrap-test-env.sh version check fails when grep has --colour option enabled.

[Ewen Cheslack-Postava (JIRA)]

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

Ewen Cheslack-Postava resolved KAFKA-3257.
--
Resolution: Fixed

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

> bootstrap-test-env.sh version check fails when grep has --colour option 
> enabled.
> 
>
> Key: KAFKA-3257
> URL: https://issues.apache.org/jira/browse/KAFKA-3257
> Project: Kafka
>  Issue Type: Bug
>  Components: system tests
>Affects Versions: 0.9.0.1
>Reporter: Jiangjie Qin
>Assignee: chen zhu
>  Labels: newbie++
> Fix For: 0.10.0.0
>
>
> When checking the versions, we use the following command:
> {code}
> vagrant --version | egrep -o "[0-9]+\.[0-9]+\.[0-9]+"
> {code}
> This does not work if user box has --colour option enabled. In my case it 
> complains:
> Found Vagrant version 1.8.1. Please upgrade to 1.6.4 or higher (see 
> http://www.vagrantup.com for details)
> We should change this line to:
> {code}
> vagrant --version | egrep --colour=never -o "[0-9]+\.[0-9]+\.[0-9]+"
> {code}



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

[jira] [Commented] (KAFKA-3236) Honor Producer Configuration "block.on.buffer.full"

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin commented on KAFKA-3236:
-

Hi Thomas, 

>From the description of the ticket, I think your confusion is from the 
>vagueness of the behavior when {{block.on.buffer.full}} is set to false. Is it 
>still a problem for you after we remove {{block.on.buffer.full}}?

I agree that having separate setting of blocking gives finer tuning 
granularity, but we found it might not necessary for user.

In your use case, if {{block.on.buffer.full = false}} and {{max.block.ms > 0}} 
is not a pure non-blocking mode because producer.send() can still block up to 
max.bock.ms, right? 

That is the exact rationale of having a single {{max.block.ms}} instead of 
decoupling time blocking on buffer full and metadata, because they provides the 
same guarantee from user's perspective: at most block for {{max.blocking.ms}}.

> Honor Producer Configuration "block.on.buffer.full"
> ---
>
> Key: KAFKA-3236
> URL: https://issues.apache.org/jira/browse/KAFKA-3236
> Project: Kafka
>  Issue Type: Improvement
>  Components: producer 
>Affects Versions: 0.9.0.0
>Reporter: Thomas Graves
>Assignee: Thomas Graves
>
> In Kafka-0.9, "max.block.ms" is used to control how long the following 
> methods will block.
> KafkaProducer.send() when
>* Buffer is full
>* Metadata is unavailable
> KafkaProducer.partitionsFor() when
>* Metadata is unavailable
> However when "block.on.buffer.full" is set to false, "max.block.ms" is in 
> effect whenever a buffer is requested/allocated from the Producer BufferPool. 
> Instead it should throw a BufferExhaustedException without waiting for 
> "max.block.ms"
> This is particulary useful if a producer application does not wish to block 
> at all on KafkaProducer.send() . We avoid waiting on KafkaProducer.send() 
> when metadata is unavailable by invoking send() only if the producer instance 
> has fetched the metadata for the topic in a different thread using the same 
> producer instance. However "max.block.ms" is still required to specify a 
> timeout for bootstrapping the metadata fetch.
> We should resolve this limitation by decoupling "max.block.ms" and 
> "block.on.buffer.full".
>* "max.block.ms" will be used exclusively for fetching metadata when
> "block.on.buffer.full" = false (in pure non-blocking mode )
>* "max.block.ms" will be applicable to both fetching metadata as well as 
> buffer allocation when "block.on.buffer.full = true



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

[jira] [Comment Edited] (KAFKA-3236) Honor Producer Configuration "block.on.buffer.full"

[Jiangjie Qin (JIRA)]

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

Jiangjie Qin edited comment on KAFKA-3236 at 3/1/16 6:01 PM:
-

Hi Thomas, 

>From the description of the ticket, I think your confusion is from the 
>vagueness of the behavior when {{block.on.buffer.full}} is set to false. Is it 
>still a problem for you after we remove {{block.on.buffer.full}}?

I agree that having separate settings of blocking gives finer tuning 
granularity, but we found it might not be necessary for user.

In your use case, if {{block.on.buffer.full = false}} and {{max.block.ms > 0}} 
is not a pure non-blocking mode because producer.send() can still block up to 
max.bock.ms, right? 

That is the exact rationale of having a single {{max.block.ms}} instead of 
decoupling time blocking on buffer full and metadata, because they provides the 
same guarantee from user's perspective: at most block for {{max.blocking.ms}}.


was (Author: becket_qin):
Hi Thomas, 

>From the description of the ticket, I think your confusion is from the 
>vagueness of the behavior when {{block.on.buffer.full}} is set to false. Is it 
>still a problem for you after we remove {{block.on.buffer.full}}?

I agree that having separate setting of blocking gives finer tuning 
granularity, but we found it might not necessary for user.

In your use case, if {{block.on.buffer.full = false}} and {{max.block.ms > 0}} 
is not a pure non-blocking mode because producer.send() can still block up to 
max.bock.ms, right? 

That is the exact rationale of having a single {{max.block.ms}} instead of 
decoupling time blocking on buffer full and metadata, because they provides the 
same guarantee from user's perspective: at most block for {{max.blocking.ms}}.

> Honor Producer Configuration "block.on.buffer.full"
> ---
>
> Key: KAFKA-3236
> URL: https://issues.apache.org/jira/browse/KAFKA-3236
> Project: Kafka
>  Issue Type: Improvement
>  Components: producer 
>Affects Versions: 0.9.0.0
>Reporter: Thomas Graves
>Assignee: Thomas Graves
>
> In Kafka-0.9, "max.block.ms" is used to control how long the following 
> methods will block.
> KafkaProducer.send() when
>* Buffer is full
>* Metadata is unavailable
> KafkaProducer.partitionsFor() when
>* Metadata is unavailable
> However when "block.on.buffer.full" is set to false, "max.block.ms" is in 
> effect whenever a buffer is requested/allocated from the Producer BufferPool. 
> Instead it should throw a BufferExhaustedException without waiting for 
> "max.block.ms"
> This is particulary useful if a producer application does not wish to block 
> at all on KafkaProducer.send() . We avoid waiting on KafkaProducer.send() 
> when metadata is unavailable by invoking send() only if the producer instance 
> has fetched the metadata for the topic in a different thread using the same 
> producer instance. However "max.block.ms" is still required to specify a 
> timeout for bootstrapping the metadata fetch.
> We should resolve this limitation by decoupling "max.block.ms" and 
> "block.on.buffer.full".
>* "max.block.ms" will be used exclusively for fetching metadata when
> "block.on.buffer.full" = false (in pure non-blocking mode )
>* "max.block.ms" will be applicable to both fetching metadata as well as 
> buffer allocation when "block.on.buffer.full = true



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

[jira] [Assigned] (KAFKA-3312) Add a offsets methods to ZkUtils and replace relevant usages

[Vahid Hashemian (JIRA)]

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

Vahid Hashemian reassigned KAFKA-3312:
--

Assignee: Vahid Hashemian

> Add a offsets methods to ZkUtils and replace relevant usages
> 
>
> Key: KAFKA-3312
> URL: https://issues.apache.org/jira/browse/KAFKA-3312
> Project: Kafka
>  Issue Type: Improvement
>Affects Versions: 0.9.0.1
>Reporter: Grant Henke
>Assignee: Vahid Hashemian
>
> There are many places in the code that manually build a zookeeper path and 
> get or update offsets. Moving this logic to a common location in ZkUtils 
> would be nice. 
> Ex:
> {code}
> zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
> {code}
> {code}
>  zkUtils.readData(topicDirs.consumerOffsetDir + "/" + 
> topicAndPartition.partition)._1.toLong
> {code}
> {code}
> zkUtils.updatePersistentPath(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}",
>  partitionData.offset.toString)
> {code}



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

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Ismael Juma]

Hi Rajini,

Thanks for clarifying. Comments inline.

On Tue, Mar 1, 2016 at 2:21 PM, Rajini Sivaram  wrote:
>
> Since we want to support arbitrary custom mechanisms, it feels better to
> use mechanism names rather than Strings. IDs would require ensuring that
> client and server have the same mapping. Since the mechanism name Strings
> are not particularly long and they are useful for debugging, I feel it is
> worthwhile to retain Strings rather than IDs.
>

Fair enough.

I wasn't sure whether Kafka attempted to retain compatibility in both
> directions. I would have preferred to send the mechanism even for GSSAPI to
> keep the code consistent, but went for version compatibility instead. I was
> thinking mainly of replication. If you have a cluster that uses 0.9.0.x
> with SASL replication, it would be easier to upgrade if new clients worked
> with old brokers.
>

That makes sense. We could use inter.broker.protocol.version for the
replication case, but `SaslClientAuthenticator` is shared between clients
and the broker and the way you did is probably simpler. However, the other
side of the coin is how we document the protocol for clients that are
distributed separately from Kafka itself. It would be good to make it clear
in such documentation that one can pass the mechanism even for the GSSAPI
case from 0.10.0.0 onwards.

Ismael

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Rajini Sivaram]

Ismael,

Thank you, will make it clear in the docs that mechanism can be passed even
for GSSAPI from 0.10.0.0.

On Tue, Mar 1, 2016 at 2:59 PM, Ismael Juma  wrote:

> Hi Rajini,
>
> Thanks for clarifying. Comments inline.
>
> On Tue, Mar 1, 2016 at 2:21 PM, Rajini Sivaram <
> rajinisiva...@googlemail.com
> > wrote:
> >
> > Since we want to support arbitrary custom mechanisms, it feels better to
> > use mechanism names rather than Strings. IDs would require ensuring that
> > client and server have the same mapping. Since the mechanism name Strings
> > are not particularly long and they are useful for debugging, I feel it is
> > worthwhile to retain Strings rather than IDs.
> >
>
> Fair enough.
>
> I wasn't sure whether Kafka attempted to retain compatibility in both
> > directions. I would have preferred to send the mechanism even for GSSAPI
> to
> > keep the code consistent, but went for version compatibility instead. I
> was
> > thinking mainly of replication. If you have a cluster that uses 0.9.0.x
> > with SASL replication, it would be easier to upgrade if new clients
> worked
> > with old brokers.
> >
>
> That makes sense. We could use inter.broker.protocol.version for the
> replication case, but `SaslClientAuthenticator` is shared between clients
> and the broker and the way you did is probably simpler. However, the other
> side of the coin is how we document the protocol for clients that are
> distributed separately from Kafka itself. It would be good to make it clear
> in such documentation that one can pass the mechanism even for the GSSAPI
> case from 0.10.0.0 onwards.
>
> Ismael
>



-- 
Regards,

Rajini

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Rajini Sivaram]

Ismael,

Thank you for the review.

"*Would it be better to assign an id to each mechanism and pass that
instead **of the String? That would be more space-efficient.*"

Since we want to support arbitrary custom mechanisms, it feels better to
use mechanism names rather than Strings. IDs would require ensuring that
client and server have the same mapping. Since the mechanism name Strings
are not particularly long and they are useful for debugging, I feel it is
worthwhile to retain Strings rather than IDs.

"
*The currently proposed way has the benefit that newer clients would
support older brokers _if_ thesasl.mechanism was GSSAPI. Is this important?
For Kafka, brokers have to be upgraded before clients, generally. Brokers
would still support older **clients that didn't send the mechanism either
way.*"

I wasn't sure whether Kafka attempted to retain compatibility in both
directions. I would have preferred to send the mechanism even for GSSAPI to
keep the code consistent, but went for version compatibility instead. I was
thinking mainly of replication. If you have a cluster that uses 0.9.0.x
with SASL replication, it would be easier to upgrade if new clients worked
with old brokers.

Thoughts?


On Tue, Mar 1, 2016 at 11:44 AM, Ismael Juma  wrote:

> Also, with regards to the client flow, it says:
>
> "If sasl.mechanism is not GSSAPI, send a packet with the mechanism name to
> the server. Otherwise go to Step 3."
>
> It sounds like it would be more regular and simpler for clients if they
> always sent the mechanism, even if GSSAPI, right? The currently proposed
> way has the benefit that newer clients would support older brokers _if_ the
> sasl.mechanism was GSSAPI. Is this important? For Kafka, brokers have to be
> upgraded before clients, generally. Brokers would still support older
> clients that didn't send the mechanism either way.
>
> Ismael
>
>
> On Tue, Mar 1, 2016 at 3:11 AM, Ismael Juma  wrote:
>
> > Hi Rajini,
> >
> > One question below.
> >
> > On Tue, Mar 1, 2016 at 2:47 AM, Rajini Sivaram <
> > rajinisiva...@googlemail.com> wrote:
> >
> >>1. With GSSAPI, the first context establishment packet starts with
> the
> >>byte 0x60 (tag for APPLICATION-0) followed by a variable-length
> encoded
> >>size, followed by various tags and contents. And the packet also
> >> contains a
> >>checksum. This is completely different from the mechanism packet from
> >> Kafka
> >>clients which start with a two-byte version set to zero currently,
> >> followed
> >>by just a String mechanism.
> >>
> >
> > Would it be better to assign an id to each mechanism and pass that
> instead
> > of the String? That would be more space-efficient.
> >
> > Ismael
> >
>



-- 
Regards,

Rajini

[jira] [Commented] (KAFKA-3236) Honor Producer Configuration "block.on.buffer.full"

[Thomas Graves (JIRA)]

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

Thomas Graves commented on KAFKA-3236:
--

Thanks for the response.  I saw that the config were deprecated in the code but 
we were hoping to bring them back for this use case.

Currently (Kafka 0.9) if block.on.buffer.full = false it blocks up to 
max.block.ms if either the metadata is unavailable of the buffer is full.
The behavior we want (and is implemented in pr) is for it to throw 
BufferExhaustedException immediately instead of waiting the max.block.ms when 
the buffer is full.  We still use max.block.ms for the metadata unavailable.

Do you think this is a reasonable request or is there a reason not to decouple 
these?

> Honor Producer Configuration "block.on.buffer.full"
> ---
>
> Key: KAFKA-3236
> URL: https://issues.apache.org/jira/browse/KAFKA-3236
> Project: Kafka
>  Issue Type: Improvement
>  Components: producer 
>Affects Versions: 0.9.0.0
>Reporter: Thomas Graves
>Assignee: Thomas Graves
>
> In Kafka-0.9, "max.block.ms" is used to control how long the following 
> methods will block.
> KafkaProducer.send() when
>* Buffer is full
>* Metadata is unavailable
> KafkaProducer.partitionsFor() when
>* Metadata is unavailable
> However when "block.on.buffer.full" is set to false, "max.block.ms" is in 
> effect whenever a buffer is requested/allocated from the Producer BufferPool. 
> Instead it should throw a BufferExhaustedException without waiting for 
> "max.block.ms"
> This is particulary useful if a producer application does not wish to block 
> at all on KafkaProducer.send() . We avoid waiting on KafkaProducer.send() 
> when metadata is unavailable by invoking send() only if the producer instance 
> has fetched the metadata for the topic in a different thread using the same 
> producer instance. However "max.block.ms" is still required to specify a 
> timeout for bootstrapping the metadata fetch.
> We should resolve this limitation by decoupling "max.block.ms" and 
> "block.on.buffer.full".
>* "max.block.ms" will be used exclusively for fetching metadata when
> "block.on.buffer.full" = false (in pure non-blocking mode )
>* "max.block.ms" will be applicable to both fetching metadata as well as 
> buffer allocation when "block.on.buffer.full = true



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

[jira] [Created] (KAFKA-3311) Move co-partition checking to PartitionAssignor and auto-create internal topics

[Guozhang Wang (JIRA)]

Guozhang Wang created KAFKA-3311:


 Summary: Move co-partition checking to PartitionAssignor and 
auto-create internal topics
 Key: KAFKA-3311
 URL: https://issues.apache.org/jira/browse/KAFKA-3311
 Project: Kafka
  Issue Type: Sub-task
Reporter: Guozhang Wang
Assignee: Guozhang Wang


Right now the internal topics management (i.e. re-partition topics for now) is 
buggy such that it was not auto-created by the assignor. We need to fix this by:

1) moving co-paritition info into the assignor.
2) let assignor create the internal topics with the right number of partitions 
as: if co-partitioned, equal to the other partitions; otherwise if there are 
other source topics in the sub-topology, equal to the maximum of other 
partitions; otherwise equal to the number of tasks writing to this topic.



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

[jira] [Commented] (KAFKA-3240) Replication issues on FreeBSD

[Johannes Huning (JIRA)]

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

Johannes Huning commented on KAFKA-3240:


>From my last runs it appears that this issue only arises when using LZ4 
>compression.

> Replication issues on FreeBSD
> -
>
> Key: KAFKA-3240
> URL: https://issues.apache.org/jira/browse/KAFKA-3240
> Project: Kafka
>  Issue Type: Bug
>  Components: core
>Affects Versions: 0.9.0.0, 0.8.2.2, 0.9.0.1
> Environment: FreeBSD 10.2-RELEASE-p9
>Reporter: Jan Omar
>
> Hi,
> We are trying to replace our 3-broker cluster running on 0.6 with a new 
> cluster on 0.9.0.1 (but tried 0.8.2.2 and 0.9.0.0 as well).
> - 3 kafka nodes with one zookeeper instance on each machine
> - FreeBSD 10.2 p9
> - Nagle off (sysctl net.inet.tcp.delayed_ack=0)
> - all kafka machines write a ZFS ZIL to a dedicated SSD
> - 3 producers on 3 machines, writing to 1 topics, partitioning 3, replication 
> factor 3
> - acks all
> - 10 Gigabit Ethernet, all machines on one switch, ping 0.05 ms worst case.
> While using the ProducerPerformance or rdkafka_performance we are seeing very 
> strange Replication errors. Any hint on what's going on would be highly 
> appreciated. Any suggestion on how to debug this properly would help as well.
> This is what our broker config looks like:
> {code}
> broker.id=5
> auto.create.topics.enable=false
> delete.topic.enable=true
> listeners=PLAINTEXT://:9092
> port=9092
> host.name=kafka-five.acc
> advertised.host.name=10.5.3.18
> zookeeper.connect=zookeeper-four.acc:2181,zookeeper-five.acc:2181,zookeeper-six.acc:2181
> zookeeper.connection.timeout.ms=6000
> num.replica.fetchers=1
> replica.fetch.max.bytes=1
> replica.fetch.wait.max.ms=500
> replica.high.watermark.checkpoint.interval.ms=5000
> replica.socket.timeout.ms=30
> replica.socket.receive.buffer.bytes=65536
> replica.lag.time.max.ms=1000
> min.insync.replicas=2
> controller.socket.timeout.ms=3
> controller.message.queue.size=100
> log.dirs=/var/db/kafka
> num.partitions=8
> message.max.bytes=1
> auto.create.topics.enable=false
> log.index.interval.bytes=4096
> log.index.size.max.bytes=10485760
> log.retention.hours=168
> log.flush.interval.ms=1
> log.flush.interval.messages=2
> log.flush.scheduler.interval.ms=2000
> log.roll.hours=168
> log.retention.check.interval.ms=30
> log.segment.bytes=536870912
> zookeeper.connection.timeout.ms=100
> zookeeper.sync.time.ms=5000
> num.io.threads=8
> num.network.threads=4
> socket.request.max.bytes=104857600
> socket.receive.buffer.bytes=1048576
> socket.send.buffer.bytes=1048576
> queued.max.requests=10
> fetch.purgatory.purge.interval.requests=100
> producer.purgatory.purge.interval.requests=100
> replica.lag.max.messages=1000
> {code}
> These are the errors we're seeing:
> {code:borderStyle=solid}
> ERROR [Replica Manager on Broker 5]: Error processing fetch operation on 
> partition [test,0] offset 50727 (kafka.server.ReplicaManager)
> java.lang.IllegalStateException: Invalid message size: 0
>   at kafka.log.FileMessageSet.searchFor(FileMessageSet.scala:141)
>   at kafka.log.LogSegment.translateOffset(LogSegment.scala:105)
>   at kafka.log.LogSegment.read(LogSegment.scala:126)
>   at kafka.log.Log.read(Log.scala:506)
>   at 
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:536)
>   at 
> kafka.server.ReplicaManager$$anonfun$readFromLocalLog$1.apply(ReplicaManager.scala:507)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
>   at 
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
>   at 
> scala.collection.immutable.HashMap$HashMap1.foreach(HashMap.scala:221)
>   at 
> scala.collection.immutable.HashMap$HashTrieMap.foreach(HashMap.scala:428)
>   at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
>   at scala.collection.AbstractTraversable.map(Traversable.scala:104)
>   at 
> kafka.server.ReplicaManager.readFromLocalLog(ReplicaManager.scala:507)
>   at kafka.server.ReplicaManager.fetchMessages(ReplicaManager.scala:462)
>   at kafka.server.KafkaApis.handleFetchRequest(KafkaApis.scala:431)
>   at kafka.server.KafkaApis.handle(KafkaApis.scala:69)
>   at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:60)
>   at java.lang.Thread.run(Thread.java:745)0
> {code}
> and 
> {code}
> ERROR Found invalid messages during fetch for partition [test,0] offset 2732 
> error Message found with corrupt size (0) in shallow iterator 
> (kafka.server.ReplicaFetcherThread)
> {code}



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

[jira] [Created] (KAFKA-3312) Add a offsets methods to ZkUtils and replace relevant usages

[Grant Henke (JIRA)]

Grant Henke created KAFKA-3312:
--

 Summary: Add a offsets methods to ZkUtils and replace relevant 
usages
 Key: KAFKA-3312
 URL: https://issues.apache.org/jira/browse/KAFKA-3312
 Project: Kafka
  Issue Type: Improvement
Affects Versions: 0.9.0.1
Reporter: Grant Henke


There are many places in the code that manually build a zookeeper path and get 
or update offsets. Moving this logic to a common location in ZkUtils would be 
nice. 

Ex:
{code}
zkUtils.readDataMaybeNull(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}")._1
{code}

{code}
 zkUtils.readData(topicDirs.consumerOffsetDir + "/" + 
topicAndPartition.partition)._1.toLong
{code}

{code}
zkUtils.updatePersistentPath(s"${topicDirs.consumerOffsetDir}/${topicPartition.partition}",
 partitionData.offset.toString)
{code}



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

[jira] [Commented] (KAFKA-3197) Producer can send message out of order even when in flight request is set to 1.

[Ismael Juma (JIRA)]

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

Ismael Juma commented on KAFKA-3197:


[~jjkoshy], since the patch doesn't introduce any new configuration and it's 
relatively simple I think it's fine to go with this approach for now. If we 
find a better way in the future, we can consider it then.

> Producer can send message out of order even when in flight request is set to 
> 1.
> ---
>
> Key: KAFKA-3197
> URL: https://issues.apache.org/jira/browse/KAFKA-3197
> Project: Kafka
>  Issue Type: Bug
>  Components: clients, producer 
>Affects Versions: 0.9.0.0
>Reporter: Jiangjie Qin
>Assignee: Jiangjie Qin
> Fix For: 0.10.0.0
>
>
> The issue we saw is following:
> 1. Producer send message 0 to topic-partition-0 on broker A. The in-flight 
> request to broker A is 1.
> 2. The request is somehow lost
> 3. Producer refreshed its topic metadata and found leader of 
> topic-partition-0 migrated from broker A to broker B.
> 4. Because there is no in-flight request to broker B. All the subsequent 
> messages to topic-partition-0 in the record accumulator are sent to broker B.
> 5. Later on when the request in step (1) times out, message 0 will be retried 
> and sent to broker B. At this point, all the later messages has already been 
> sent, so we have re-order.



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

Re: [DISCUSS] KIP-43: Kafka SASL enhancements

[Jun Rao]

Rajini,

Thanks for the explanation. For 1, this implies that we have to be careful
with changing the 2-byte version in the future to avoid conflict. Could you
document this in the KIP and also in the implementation?

Jun

On Tue, Mar 1, 2016 at 2:47 AM, Rajini Sivaram  wrote:

> Jun,
>
> Thank you for the review.
>
>
>1. With GSSAPI, the first context establishment packet starts with the
>byte 0x60 (tag for APPLICATION-0) followed by a variable-length encoded
>size, followed by various tags and contents. And the packet also
> contains a
>checksum. This is completely different from the mechanism packet from
> Kafka
>clients which start with a two-byte version set to zero currently,
> followed
>by just a String mechanism.
>2. Agreed, I have removed the version from the server response in the
>KIP. Thanks.
>
>
> On Tue, Mar 1, 2016 at 2:33 AM, Jun Rao  wrote:
>
> > Rajini,
> >
> > Thanks for the updates. Just a couple of minor comments.
> >
> > 1. With the default GSSAPI, what's the first packet that the client sends
> > to the server? Is that completely different from the packet format that
> we
> > will use for non-GSSAPI mechanisms?
> >
> > 2. In the server response, it doesn't seem that we need to include the
> > version since the client knows the version of the request that it sends.
> >
> > Jun
> >
> > On Mon, Feb 29, 2016 at 10:14 AM, Rajini Sivaram <
> > rajinisiva...@googlemail.com> wrote:
> >
> > > Harsha,
> > >
> > > Thank you for the review. I will wait another day to see if there is
> more
> > > feedback and then start a voting thread.
> > >
> > > Rajini
> > >
> > > On Mon, Feb 29, 2016 at 2:51 PM, Harsha  wrote:
> > >
> > > > Rajini,
> > > >   Thanks for the changes to the KIP. It looks good to
> me. I
> > > >   think we can move to voting.
> > > > Thanks,
> > > > Harsha
> > > >
> > > > On Mon, Feb 29, 2016, at 12:43 AM, Rajini Sivaram wrote:
> > > > > I have added some more detail to the KIP based on the discussion in
> > the
> > > > > last KIP meeting to simplify support for multiple mechanisms. Have
> > also
> > > > > changed the property names to reflect this.
> > > > >
> > > > > Also updated the PR in
> > > https://issues.apache.org/jira/browse/KAFKA-3149
> > > > > to
> > > > > reflect the KIP.
> > > > >
> > > > > Any feedback is appreciated.
> > > > >
> > > > >
> > > > > On Tue, Feb 23, 2016 at 9:36 PM, Rajini Sivaram <
> > > > > rajinisiva...@googlemail.com> wrote:
> > > > >
> > > > > > I have updated the KIP based on the discussion in the KIP meeting
> > > > today.
> > > > > >
> > > > > > Comments and feedback are welcome.
> > > > > >
> > > > > > On Wed, Feb 3, 2016 at 7:20 PM, Rajini Sivaram <
> > > > > > rajinisiva...@googlemail.com> wrote:
> > > > > >
> > > > > >> Hi Harsha,
> > > > > >>
> > > > > >> Thank you for the review. Can you clarify - I think you are
> saying
> > > > that
> > > > > >> the client should send its mechanism over the wire to the
> server.
> > Is
> > > > that
> > > > > >> correct? The exchange is slightly different in the KIP (the PR
> > > > matches the
> > > > > >> KIP) from the one you described to enable interoperability with
> > > > 0.9.0.0.
> > > > > >>
> > > > > >>
> > > > > >> On Wed, Feb 3, 2016 at 1:56 PM, Harsha  wrote:
> > > > > >>
> > > > > >>> Rajini,
> > > > > >>>I looked at the PR you have. I think its better with
> > > your
> > > > > >>>earlier approach rather than extending the protocol.
> > > > > >>> What I was thinking initially is, Broker has a config option of
> > say
> > > > > >>> sasl.mechanism = GSSAPI, PLAIN
> > > > > >>> and the client can have similar config of sasl.mechanism=PLAIN.
> > > > Client
> > > > > >>> can send its sasl mechanism before the handshake starts and if
> > the
> > > > > >>> broker accepts that particular mechanism than it can go ahead
> > with
> > > > > >>> handshake otherwise return a error saying that the mechanism
> not
> > > > > >>> allowed.
> > > > > >>>
> > > > > >>> Thanks,
> > > > > >>> Harsha
> > > > > >>>
> > > > > >>> On Wed, Feb 3, 2016, at 04:58 AM, Rajini Sivaram wrote:
> > > > > >>> > A slightly different approach for supporting different SASL
> > > > mechanisms
> > > > > >>> > within a broker is to allow the same "*security protocol*" to
> > be
> > > > used
> > > > > >>> on
> > > > > >>> > different ports with different configuration options. An
> > > advantage
> > > > of
> > > > > >>> > this
> > > > > >>> > approach is that it extends the configurability of not just
> > SASL,
> > > > but
> > > > > >>> any
> > > > > >>> > protocol. For instance, it would enable the use of SSL with
> > > mutual
> > > > > >>> client
> > > > > >>> > authentication on one port or different certificate chains on
> > > > another.
> > > > > >>> > And
> > > > > >>> > it avoids the need for SASL mechanism negotiation.
> > > > > >>> >
> > > > > >>> > Kafka would