Build failed in Jenkins: kafka-trunk-jdk11 #52

[Apache Jenkins Server]

See 


Changes:

[mjsax] MINOR: buffer should ignore caching (#5819)

--
[...truncated 2.34 MB...]

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualForCompareValueTimestampWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualForCompareValueTimestampWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReverseForCompareValueTimestampWithProducerRecord
 STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReverseForCompareValueTimestampWithProducerRecord
 PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateConsumerRecordsFromKeyValuePairs STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateConsumerRecordsFromKeyValuePairs PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullTopicNameWithNullKeyAndDefaultTimestamp 
STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullTopicNameWithNullKeyAndDefaultTimestamp 
PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithDefaultTimestamp 
STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithDefaultTimestamp 
PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullTopicNameWithNullKey STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullTopicNameWithNullKey PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateNullKeyConsumerRecordWithOtherTopicNameAndTimestampWithTimetamp 
STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateNullKeyConsumerRecordWithOtherTopicNameAndTimestampWithTimetamp 
PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateConsumerRecordWithTimestamp STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateConsumerRecordWithTimestamp PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullHeaders STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullHeaders PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullTopicNameWithDefaultTimestamp STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldNotAllowToCreateTopicWithNullTopicNameWithDefaultTimestamp PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldRequireCustomTopicNameIfNotDefaultFactoryTopicName STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldRequireCustomTopicNameIfNotDefaultFactoryTopicName PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithNullKey STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithNullKey PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateConsumerRecord STARTED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateConsumerRecord PASSED

org.apache.kafka.streams.test.ConsumerRecordFactoryTest > 
shouldCreateNullKeyConsumerRecord STARTED

org.apache.kafka.stream

[jira] [Created] (KAFKA-7537) Only include live brokers in the UpdateMetadataRequest sent to existing brokers if there is no change in the partition states

[Zhanxiang (Patrick) Huang (JIRA)]

Zhanxiang (Patrick) Huang created KAFKA-7537:


 Summary: Only include live brokers in the UpdateMetadataRequest 
sent to existing brokers if there is no change in the partition states
 Key: KAFKA-7537
 URL: https://issues.apache.org/jira/browse/KAFKA-7537
 Project: Kafka
  Issue Type: Improvement
  Components: controller
Reporter: Zhanxiang (Patrick) Huang
Assignee: Zhanxiang (Patrick) Huang


Currently if when brokers join/leave the cluster without any partition states 
changes, controller will send out UpdateMetadataRequests containing the states 
of all partitions to all brokers. But for existing brokers in the cluster, the 
metadata diff between controller and the broker should only be the 
"live_brokers" info. Only the brokers with empty metadata cache need the full 
UpdateMetadataRequest. Sending the full UpdateMetadataRequest to all brokers 
can place nonnegligible memory pressure on the controller side.

Let's say in total we have N brokers, M partitions in the cluster and we want 
to add 1 brand new broker in the cluster. With RF=2, the memory footprint per 
partition in the UpdateMetadataRequest is ~200 Bytes. In the current controller 
implementation, if each of the N RequestSendThreads serializes and sends out 
the UpdateMetadataRequest at roughly the same time (which is very likely the 
case), we will end up using *(N+1)*M*200B*. In a large kafka cluster, we can 
have:
{noformat}
N=99
M=100k

Memory usage to send out UpdateMetadataRequest to all brokers:
100 * 100K * 200B = 2G

However, we only need to send out full UpdateMetadataRequest to the newly added 
broker. We only need to include live broker ids (4B * 100 brokers) in the 
UpdateMetadataRequest sent to the existing 99 brokers. So the amount of data 
that is actully needed will be:
1 * 100K * 200B + 99 * (100 * 4B) = ~21M


We will can potentially reduce 2G / 21M = ~95x memory footprint as well as the 
data tranferred in the network.{noformat}
 

This issue kind of hurts the scalability of a kafka cluster. KIP-380 and 
KAFKA-7186 also help to further reduce the controller memory footprint.

 

In terms of implementation, we can keep some in-memory state in the controller 
side to differentiate existing brokers and uninitialized brokers (e.g. brand 
new brokers) so that if there is no change in partition states, we only send 
out live brokers info to existing brokers.

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[VOTE] 2.1.0 RC0

[Dong Lin]

Hello Kafka users, developers and client-developers,

This is the first candidate for feature release of Apache Kafka 2.1.0.

This is a major version release of Apache Kafka. It includes 28 new KIPs and

critical bug fixes. Please see the Kafka 2.1.0 release plan for more
details:

*https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=91554044*


Here are a few notable highlights:

- Java 11 support
- Support for Zstandard, which achieves compression comparable to gzip with
higher compression and especially decompression speeds(KIP-110)
- Avoid expiring committed offsets for active consumer group (KIP-211)
- Provide Intuitive User Timeouts in The Producer (KIP-91)
- Kafka's replication protocol now supports improved fencing of zombies.
Previously, under certain rare conditions, if a broker became partitioned
from Zookeeper but not the rest of the cluster, then the logs of replicated
partitions could diverge and cause data loss in the worst case (KIP-320)
- Streams API improvements (KIP-319, KIP-321, KIP-330, KIP-353, KIP-356)
- Admin script and admin client API improvements to simplify admin
operation (KIP-231, KIP-308, KIP-322, KIP-324, KIP-338, KIP-340)
- DNS handling improvements (KIP-235, KIP-302)

Release notes for the 2.1.0 release:
http://home.apache.org/~lindong/kafka-2.1.0-rc0/RELEASE_NOTES.html

*** Please download, test and vote ***

* Kafka's KEYS file containing PGP keys we use to sign the release:
http://kafka.apache.org/KEYS

* Release artifacts to be voted upon (source and binary):
http://home.apache.org/~lindong/kafka-2.1.0-rc0/

* Maven artifacts to be voted upon:
https://repository.apache.org/content/groups/staging/

* Javadoc:
http://home.apache.org/~lindong/kafka-2.1.0-rc0/javadoc/

* Tag to be voted upon (off 2.1 branch) is the 2.1.0-rc0 tag:
https://github.com/apache/kafka/tree/2.1.0-rc0

* Documentation:
*http://kafka.apache.org/21/documentation.html*


* Protocol:
http://kafka.apache.org/21/protocol.html

* Successful Jenkins builds for the 2.1 branch:
Unit/integration tests: *https://builds.apache.org/job/kafka-2.1-jdk8/38/
*

Please test and verify the release artifacts and submit a vote for this RC,
or report any issues so we can fix them and get a new RC out ASAP. Although
this release vote requires PMC votes to pass, testing, votes, and bug
reports are valuable and appreciated from everyone.

Cheers,
Dong

classes with Test in their name

[abeceda4]

Hi,
 
there, in the code, is lot of classes with "Test" in their name. For example 
KafkaProducerTest, KafkaAdminClientTest or ClusterConnectionStateTest etc. 
Could you tell me please, if these classe are used somewhere in Kafka by 
regular way or
if these classes are used only for testing purposes?
 
Thank you very much
 
Mrkvica

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

[Apache Jenkins Server]

See 


Changes:

[mjsax] MINOR: buffer should ignore caching (#5819)

--
[...truncated 2.85 MB...]
org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualForCompareKeyValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualForCompareKeyValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualForCompareValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualForCompareValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullExpectedRecordForCompareKeyValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullExpectedRecordForCompareKeyValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareValueTimestampWithProducerRecord STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareValueTimestampWithProducerRecord PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullReversForCompareKeyValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullExpectedRecordForCompareKeyValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullExpectedRecordForCompareKeyValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareValueWithProducerRecord STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentForCompareValueWithProducerRecord PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueIsEqualForCompareValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueIsEqualForCompareValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordWithExpectedRecordForCompareKeyValueTimestamp 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordWithExpectedRecordForCompareKeyValueTimestamp 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentForCompareKeyValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentForCompareKeyValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordWithExpectedRecordForCompareKeyValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldNotAllowNullProducerRecordWithExpectedRecordForCompareKeyValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullForCompareKeyValueWithProducerRecord STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDifferentWithNullForCompareKeyValueWithProducerRecord PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueAndTimestampIsEqualForCompareKeyValueTimestampWithProducerRecord
 STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfKeyAndValueAndTimestampIsEqualForCompareKeyValueTimestampWithProducerRecord
 PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueIsEqualWithNullForCompareValue STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueIsEqualWithNullForCompareValue PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualWithNullForCompareValueTimestamp STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldPassIfValueAndTimestampIsEqualWithNullForCompareValueTimestamp PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareValueTimestampWithProducerRecord 
STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareValueTimestampWithProducerRecord 
PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareValueWithProducerRecord STARTED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfValueIsDifferentWithNullForCompareValueWithProducerRecord PASSED

org.apache.kafka.streams.test.OutputVerifierTest > 
shouldFailIfKeyIsDiffere

Re: classes with Test in their name

[Suman B N]

Those are for unit and integration tests only. Looks like you are very new
to Java/Scala and Kafka.

On Wed, Oct 24, 2018 at 2:38 PM  wrote:

> Hi,
>
> there, in the code, is lot of classes with "Test" in their name. For
> example KafkaProducerTest, KafkaAdminClientTest or
> ClusterConnectionStateTest etc. Could you tell me please, if these classe
> are used somewhere in Kafka by regular way or
> if these classes are used only for testing purposes?
>
> Thank you very much
>
> Mrkvica
>


-- 
*Suman*
*OlaCabs*

Re: classes with Test in their name

[abeceda4]

You are right, I am beginner in Kafka. Therefore I very appreciate your help. 
Thank you very much.
__
> Od: "Suman B N" 
> Komu: dev@kafka.apache.org
> Dátum: 24.10.2018 11:22
> Predmet: Re: classes with Test in their name
>
Those are for unit and integration tests only. Looks like you are very new
 to Java/Scala and Kafka.
 
 On Wed, Oct 24, 2018 at 2:38 PM  wrote:
 
 > Hi,
 >
 > there, in the code, is lot of classes with "Test" in their name. For
 > example KafkaProducerTest, KafkaAdminClientTest or
 > ClusterConnectionStateTest etc. Could you tell me please, if these classe
 > are used somewhere in Kafka by regular way or
 > if these classes are used only for testing purposes?
 >
 > Thank you very much
 >
 > Mrkvica
 >
 
 
 -- 
 *Suman*
 *OlaCabs*

Re: [DISCUSS] KIP-386: Make Min metrics' default value consistent with Max metrics

[Stanislav Kozlovski]

Hi Kevin,

Thanks for providing context.

I've edited the KIP to use `NaN` for all three metric types.
Here is the updated link: KIP-386: Standardize on Min/Avg/Max metrics'
default value


On Tue, Oct 23, 2018 at 7:30 PM Kevin Lu  wrote:

> Hi Stanislav,
>
> Thanks for the KIP!
>
> Standardizing this would be extremely helpful. We have been publishing
> client-side metrics to a time-series database using metrics reporter since
> 0.8, and we had to do explicit checks like "!Double.isNaN(metricValue)",
> "metricValue != Double.NEGATIVE_INFINITY", and such to skip these values.
>
> I like John's suggestion for using NaN for the same reason that -INF
> and +INF are still technically valid values as they are just the extreme
> bounds.
>
> Regards,
> Kevin
>
> On Tue, Oct 23, 2018 at 3:46 AM Stanislav Kozlovski <
> stanis...@confluent.io>
> wrote:
>
> > Hey John,
> >
> > I think NaN would be the better option semantically. If we were to use
> > that, maybe it makes sense to change `Avg()`'s default value from 0.0 to
> > NaN as well.
> >
> > I am a bit more concerned with backwards compatibility if we were to use
> > NaN since we would change three types of metrics. There were only three
> > metrics that use `Min` but `Avg` and `Max` are used everywhere.
> > I guess it boils down to how likely it is that such a change can break
> > users' tools and if we're okay with it. My assumption is that it won't -
> > most tools should be mature enough to handle these values.
> >
> > Thanks for the suggestion! I will wait out a bit more for other people to
> > share their thoughts and update the KIP with `NaN`
> >
> >
> > On Tue, Oct 23, 2018 at 6:12 AM John Roesler  wrote:
> >
> > > Hi Stanislav,
> > > Thanks for this KIP. I coincidentally just noticed these strange
> initial
> > > values while doing some performance measurements.
> > >
> > > Since the metric type is a double, could we consider NaN instead? It
> > seems
> > > like +Inf is somewhat arbitrary for Max, so it might as well be an
> > > arbitrary value that actually means "this is not a number".
> > >
> > > Consider that +- Infinity is technically "in range" for max and min,
> > while
> > > NaN is not. NaN is "in range" for an average or a rate, but in those
> > cases
> > > it would mean that the result is over 0 samples or 0 time,
> respectively.
> > I
> > > think this only happens when nothing has been recorded, so it would
> still
> > > be sound for the situation you're attempting to address.
> > >
> > > Just to throw it out there, `null` is technically also (maybe moreso) a
> > > sound choice, but I'd be concerned about causing a bunch of null
> > > dereference errors.
> > >
> > > Thanks again,
> > > -John
> > >
> > > On Mon, Oct 22, 2018 at 2:27 PM Stanislav Kozlovski <
> > > stanis...@confluent.io>
> > > wrote:
> > >
> > > > Hi Suman,
> > > >
> > > > Thanks for taking a look at this. Yeah, it's very minor but it
> changes
> > > the
> > > > public API (even if this is a very slight change) and as far as I
> know
> > > this
> > > > warrants some discussion
> > > >
> > > > On Mon, Oct 22, 2018 at 9:28 PM Suman B N 
> > wrote:
> > > >
> > > > > Looks good to me. Maintains uniformity. +1 from me.
> > > > > But its more of a bug rather than improvement proposal. Let's see
> > what
> > > > > contributors got to say.
> > > > >
> > > > > On Mon, Oct 22, 2018 at 7:23 PM Stanislav Kozlovski <
> > > > > stanis...@confluent.io>
> > > > > wrote:
> > > > >
> > > > > > Hey everybody,
> > > > > >
> > > > > > I've opened up a very short KIP to make the Max and Min metrics'
> > > > default
> > > > > > values consistent with each other.
> > > > > >
> > > > > > KIP:
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-386%3A+Make+Min+metrics%27+default+value+consistent+with+Max+metrics
> > > > > > JIRA: https://issues.apache.org/jira/browse/KAFKA-7528
> > > > > >
> > > > > > This is hopefully a very straightforward change. Please provide
> > > > feedback.
> > > > > > Thanks
> > > > > >
> > > > > > --
> > > > > > Best,
> > > > > > Stanislav
> > > > > >
> > > > >
> > > > >
> > > > > --
> > > > > *Suman*
> > > > > *OlaCabs*
> > > > >
> > > >
> > > >
> > > > --
> > > > Best,
> > > > Stanislav
> > > >
> > >
> >
> >
> > --
> > Best,
> > Stanislav
> >
>


-- 
Best,
Stanislav

Re: [VOTE] 2.1.0 RC0

[Andras Beni]

+1 (non-binding)

Verified signatures and checksums of release artifacts
Performed quickstart steps on rc artifacts (both scala 2.11 and 2.12) and
one built from tag 2.1.0-rc0

Andras

On Wed, Oct 24, 2018 at 10:17 AM Dong Lin  wrote:

> Hello Kafka users, developers and client-developers,
>
> This is the first candidate for feature release of Apache Kafka 2.1.0.
>
> This is a major version release of Apache Kafka. It includes 28 new KIPs
> and
>
> critical bug fixes. Please see the Kafka 2.1.0 release plan for more
> details:
>
> *
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=91554044*
>  >
>
> Here are a few notable highlights:
>
> - Java 11 support
> - Support for Zstandard, which achieves compression comparable to gzip with
> higher compression and especially decompression speeds(KIP-110)
> - Avoid expiring committed offsets for active consumer group (KIP-211)
> - Provide Intuitive User Timeouts in The Producer (KIP-91)
> - Kafka's replication protocol now supports improved fencing of zombies.
> Previously, under certain rare conditions, if a broker became partitioned
> from Zookeeper but not the rest of the cluster, then the logs of replicated
> partitions could diverge and cause data loss in the worst case (KIP-320)
> - Streams API improvements (KIP-319, KIP-321, KIP-330, KIP-353, KIP-356)
> - Admin script and admin client API improvements to simplify admin
> operation (KIP-231, KIP-308, KIP-322, KIP-324, KIP-338, KIP-340)
> - DNS handling improvements (KIP-235, KIP-302)
>
> Release notes for the 2.1.0 release:
> http://home.apache.org/~lindong/kafka-2.1.0-rc0/RELEASE_NOTES.html
>
> *** Please download, test and vote ***
>
> * Kafka's KEYS file containing PGP keys we use to sign the release:
> http://kafka.apache.org/KEYS
>
> * Release artifacts to be voted upon (source and binary):
> http://home.apache.org/~lindong/kafka-2.1.0-rc0/
>
> * Maven artifacts to be voted upon:
> https://repository.apache.org/content/groups/staging/
>
> * Javadoc:
> http://home.apache.org/~lindong/kafka-2.1.0-rc0/javadoc/
>
> * Tag to be voted upon (off 2.1 branch) is the 2.1.0-rc0 tag:
> https://github.com/apache/kafka/tree/2.1.0-rc0
>
> * Documentation:
> *http://kafka.apache.org/21/documentation.html*
> 
>
> * Protocol:
> http://kafka.apache.org/21/protocol.html
>
> * Successful Jenkins builds for the 2.1 branch:
> Unit/integration tests: *https://builds.apache.org/job/kafka-2.1-jdk8/38/
> *
>
> Please test and verify the release artifacts and submit a vote for this RC,
> or report any issues so we can fix them and get a new RC out ASAP. Although
> this release vote requires PMC votes to pass, testing, votes, and bug
> reports are valuable and appreciated from everyone.
>
> Cheers,
> Dong
>

[jira] [Created] (KAFKA-7538) Improve locking model used to update ISRs and HW

[Rajini Sivaram (JIRA)]

Rajini Sivaram created KAFKA-7538:
-

 Summary: Improve locking model used to update ISRs and HW
 Key: KAFKA-7538
 URL: https://issues.apache.org/jira/browse/KAFKA-7538
 Project: Kafka
  Issue Type: Improvement
  Components: core
Affects Versions: 2.1.0
Reporter: Rajini Sivaram
Assignee: Rajini Sivaram
 Fix For: 2.2.0


We currently use a ReadWriteLock in Partition to update ISRs and high water 
mark for the partition. This can result in severe lock contention if there are 
multiple producers writing a large amount of data into a single partition.

The current locking model is:
 # read lock while appending to log on every Produce request on the request 
handler thread
 # write lock on leader change, updating ISRs etc. on request handler or 
scheduler thread
 # write lock on every replica fetch request to check if ISRs need to be 
updated and to update HW and ISR on the request handler thread

2) is infrequent, but 1) and 3) may be frequent and can result in lock 
contention. If there are lots of produce requests to a partition from multiple 
processes, on the leader broker we may see:
 # one slow log append locks up one request thread for that produce while 
holding onto the read lock
 # (replicationFactor-1) request threads can be blocked waiting for write lock 
to process replica fetch request
 # potentially several other request threads processing Produce may be queued 
up to acquire read lock because of the waiting writers.

In a thread dump with this issue, we noticed several request threads blocked 
waiting for write, possibly to due to replication fetch retries.

 

Possible fixes:
 # Process `Partition#maybeExpandIsr` on a single scheduler thread similar to 
`Partition#maybeShrinkIsr` so that only a single thread is blocked on the write 
lock. But this will delay updating ISRs and HW.
 # Change locking in `Partition#maybeExpandIsr` so that only read lock is 
acquired to check if ISR needs updating and write lock is acquired only to 
update ISRs. Also use a different lock for updating HW (perhaps just the 
Partition object lock) so that typical replica fetch requests complete without 
acquiring Partition write lock on the request handler thread.

I will submit a PR for 2) , but other suggestions to fix this are welcome.

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

Re: [DISCUSS] KIP-377: TopicCommand to use AdminClient

[Viktor Somogyi-Vass]

Hi All,

Colin, thanks for the heads-up. I'll rethink this metadata protocol thing
as in a global sense there might be other options as you mentioned and
start separate a discussion.

I'll start a vote soon as the KIP itself is relatively simple.

Viktor

On Tue, Oct 23, 2018 at 3:33 AM Kevin Lu  wrote:

> Hi Viktor,
>
> +1 to this KIP.
>
> I would very much like to see AdminClient in TopicCommand. This would also
> allow us to efficiently implement new features like the "--under-min-isr"
> option I proposed in KIP-351
> <
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-351%3A+Add+--under-min-isr+option+to+describe+topics+command
> >
> .
>
> Thanks.
>
> Regards,
> Kevin
>
> On Sat, Oct 20, 2018 at 10:52 PM Colin McCabe  wrote:
>
> > Hi Viktor,
> >
> > Sounds good.  If you want to propose a way of improving the metadata
> > protocol so that "[deleted]" could be supported, you could probably
> create
> > that KIP in parallel.
> >
> > The last KIP in that area that I can remember is KIP-142, which didn't
> get
> > adopted (yet?)
> >
> >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-142%3A+Add+ListTopicsRequest+to+efficiently+list+all+the+topics+in+a+cluster
> >
> > There have been other discussions though.  In general there are a lot of
> > features that would be nice to have in the metadata protocol (pagniation,
> > regexes, skip stuff we don't need).
> >
> > best,
> > Colin
> >
> >
> > On Tue, Oct 16, 2018, at 10:11, Viktor Somogyi-Vass wrote:
> > > Hi Colin,
> > >
> > > Thanks, it makes sense and simplifies this KIP tremendously. I'll move
> > this
> > > section to the rejected alternatives with a note that KIP-142 will have
> > > this feature.
> > > On a similar note: is there a KIP for describe topics protocol or have
> > you
> > > been thinking about it? I guess there it's the same problem, we often
> > don't
> > > want to forward the entire metadata.
> > >
> > > Viktor
> > >
> > > On Fri, Oct 12, 2018 at 12:03 PM Colin McCabe 
> > wrote:
> > >
> > > > Hi Viktor,
> > > >
> > > > Thanks for bumping this thread.
> > > >
> > > > I think we should just focus on transitioning the TopicCommand to use
> > > > AdminClient, and talk about protocol changes in a separate KIP.
> > Protocol
> > > > changes often involve a lot of discussion.  This does mean that we
> > couldn't
> > > > implement the "list topics under deletion" feature when using
> > AdminClient
> > > > at the moment.  We could add a note to the tool output indicating
> this.
> > > >
> > > > We should move the protocol discussion to a separate thread.
> Probably
> > > > also look at KIP-142 as well.
> > > >
> > > > best,
> > > > Colin
> > > >
> > > >
> > > > On Tue, Oct 9, 2018, at 07:45, Viktor Somogyi-Vass wrote:
> > > > > Hi All,
> > > > >
> > > > > Would like to bump this as the conversation sank a little bit, but
> > more
> > > > > importantly I'd like to validate my plans/ideas on extending the
> > Metadata
> > > > > protocol. I was thinking about two other alternatives, namely:
> > > > > 1. Create a ListTopicUnderDeletion protocol. This however would be
> > > > > unnecessary: it'd have one very narrow functionality which we can't
> > > > extend.
> > > > > I'd make sense to have a list topics or describe topics protocol
> > where we
> > > > > can list/describe topics under deletion but for normal
> > listing/describing
> > > > > we already use the metadata, so it would be a duplication of
> > > > functionality.
> > > > > 2. DeleteTopicsResponse could return the topics under deletion if
> the
> > > > > request's argument list is empty which might make sense at the
> first
> > > > look,
> > > > > but actually we'd mix the query functionality with the delete
> > > > functionality
> > > > > which is counterintuitive.
> > > > >
> > > > > Even though most clients won't need these "limbo" topics (which are
> > under
> > > > > deletion) in the foreseeable future, it can be considered as part
> of
> > the
> > > > > cluster state or metadata and to me it makes sense. Also it doesn't
> > have
> > > > a
> > > > > big overhead in the response size as typically users don't delete
> > topics
> > > > > too often as far as I experienced.
> > > > >
> > > > > I'd be happy to receive some ideas/feedback on this.
> > > > >
> > > > > Cheers,
> > > > > Viktor
> > > > >
> > > > >
> > > > > On Fri, Sep 28, 2018 at 4:51 PM Viktor Somogyi-Vass <
> > > > viktorsomo...@gmail.com>
> > > > > wrote:
> > > > >
> > > > > > Hi All,
> > > > > >
> > > > > > I made an update to the KIP. Just in short:
> > > > > > Currently KafkaAdminClient.describeTopics() and
> > > > > > KafkaAdminClient.listTopics() uses the Metadata protocol to
> acquire
> > > > topic
> > > > > > information. The returned response however won't contain the
> topics
> > > > that
> > > > > > are under deletion but couldn't complete yet (for instance
> because
> > of
> > > > some
> > > > > > replicas offline), therefore it is not possible to implement the
> > > > current
> > > > > > command's 

[VOTE] KIP-377: TopicCommand to use AdminClient

[Viktor Somogyi-Vass]

Hi All,

I'd like to start a vote on KIP-377:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-377%3A+TopicCommand+to+use+AdminClient.

Summary:
The KIP basically proposes to add --bootstrap-server and
--command-config option to TopicsCommand and implement topic
administration with AdminClient in a backwards compatible way (so
wouldn't drop or change the --zookeeper option usage).

I'd appreciate any votes or feedback.

Viktor

Re: [VOTE] KIP-377: TopicCommand to use AdminClient

[Mickael Maison]

+1 (non-binding)
Thanks for the KIP!
On Wed, Oct 24, 2018 at 1:28 PM Viktor Somogyi-Vass
 wrote:
>
> Hi All,
>
> I'd like to start a vote on KIP-377:
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-377%3A+TopicCommand+to+use+AdminClient.
>
> Summary:
> The KIP basically proposes to add --bootstrap-server and
> --command-config option to TopicsCommand and implement topic
> administration with AdminClient in a backwards compatible way (so
> wouldn't drop or change the --zookeeper option usage).
>
> I'd appreciate any votes or feedback.
>
> Viktor

RE: KAFKA-3932 - Consumer fails to consume in a round robin fashion

[ChienHsing Wu]

I don't see any comments/concerns. I would like to implement and commit to this 
ticket. Could anyone let me know how to request for the permission to assign 
that ticket to me?

Thanks, CH

From: ChienHsing Wu
Sent: Monday, October 22, 2018 1:40 PM
To: 'dev@kafka.apache.org' 
Subject: KAFKA-3932 - Consumer fails to consume in a round robin fashion


Hi,



I encountered the issue documented in the jira 
KAFKA-3932.
 Upon studying the source code and the 
PIP,
 I think the issues is the statement in PIP: "As before, we'd keep track of 
which partition we left off at so that the next iteration would begin there." I 
think it should NOT use the last partition in the next iteration; it should 
pick the next one instead.

If this behavior is agreeable, the simplest solution to impose the order to 
pick the next one is to use the order the consumer.internals.Fetcher receives 
the partition messages, as determined by completedFetches queue in that class. 
To avoid parsing the partition messages repeatedly. we can save those parsed 
fetches to a list and maintain the next partition to get messages there.

Does it sound like a good approach? If this is not the right place to discuss 
the design please let me know where to engage. If this is agreeable I can 
contribute the implementation.



Thanks, CH

Re: [DISCUSS] KIP-374: Add '--help' option to all available Kafka CLI commands

[Srinivas Reddy]

Thank you Colin for the response. Any feedback from others?



--
Srinivas Reddy

http://mrsrinivas.com/


(Sent via gmail web)


On Tue, 2 Oct 2018 at 02:18, Colin McCabe  wrote:

> Hi Srinivas,
>
> Thanks for the KIP.
>
> +1 for adding a --help option to all commands that don't already have it.
>
> If everyone agrees, then maybe you can call a vote on this KIP.  Once the
> KIP is accepted, then we can accept the corresponding PR.  I know it seems
> a bit heavyweight, but we prefer to be conservative when changing public
> command line interfaces.
>
> best,
> Colin
>
> On Sun, Sep 30, 2018, at 21:02, Srinivas Reddy wrote:
> > Hi Dongjin,
> >
> > I have identified most of the code changes for this task. I can raise a
> > GitHub PR if community think this is a good to have feature in Kafka.
> >
> > Should I start working on it?
> >
> > -
> > Srinivas
> >
> > - Typed on tiny keys. pls ignore typos.{mobile app}
> >
> > On Thu 27 Sep, 2018, 14:40 Dongjin Lee,  wrote:
> >
> > > Hi Srinivas,
> > >
> > > Thanks for your proposal. It sounds reasonable to me. However, the code
> > > freeze of 2.1.0 is imminent so it seems like all the committers are on
> > > working frenzy on it. How about to postpone reboot this discussion
> after
> > > the release of 2.1.0? I will kick off the thread as soon as the
> release is
> > > completed.
> > >
> > > Thanks,
> > > Dongjin
> > >
> > > On Thu, Sep 27, 2018 at 1:31 PM Srinivas Reddy <
> srinivas96all...@gmail.com
> > > >
> > > wrote:
> > >
> > > > Any thoughts about this KIP ?
> > > >
> > > > -
> > > > Srinivas
> > > >
> > > > - Typed on tiny keys. pls ignore typos.{mobile app}
> > > >
> > > > On Fri 21 Sep, 2018, 16:07 Srinivas Reddy, <
> srinivas96all...@gmail.com>
> > > > wrote:
> > > >
> > > > > Hi All,
> > > > >
> > > > > I'd like to start a discussion on KIP-374 which is to add "--help"
> > > option
> > > > > for all the possible CLI commands in Kafka. right now, few commands
> > > > support
> > > > > it and few does not. We brings up this KIP to provide consistent
> user
> > > > > experience.
> > > > >
> > > > > More details added in this KIP page(including code changes):
> > > > >
> > > > >
> > > >
> > >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-374%3A+Add+%27--help%27+option+to+all+available+Kafka+CLI+commands
> > > > >
> > > > > I look forward to feedback and comments.
> > > > >
> > > > >
> > > > >
> > > > > --
> > > > > Srinivas Reddy
> > > > >
> > > > > http://mrsrinivas.com/
> > > > >
> > > > >
> > > > > (Sent via gmail web)
> > > > >
> > > >
> > >
> > >
> > > --
> > > *Dongjin Lee*
> > >
> > > *A hitchhiker in the mathematical world.*
> > >
> > > *github:  github.com/dongjinleekr
> > > linkedin:
> kr.linkedin.com/in/dongjinleekr
> > > slideshare:
> > > www.slideshare.net/dongjinleekr
> > > *
> > >
>

Re: [VOTE] KIP-380: Detect outdated control requests and bounced brokers using broker generation

[Jun Rao]

Hi, Patrick,

Could you update the KIP with the changes to ControlledShutdownRequest
based on the discussion thread?

Thanks,

Jun


On Sun, Oct 21, 2018 at 2:25 PM, Mickael Maison 
wrote:

> +1( non-binding)
> Thanks for the KIP!
>
> On Sun, Oct 21, 2018, 03:31 Harsha Chintalapani  wrote:
>
> > +1(binding). LGTM.
> > -Harsha
> > On Oct 20, 2018, 4:49 PM -0700, Dong Lin , wrote:
> > > Thanks much for the KIP Patrick. Looks pretty good.
> > >
> > > +1 (binding)
> > >
> > > On Fri, Oct 19, 2018 at 10:17 AM Patrick Huang 
> > wrote:
> > >
> > > > Hi All,
> > > >
> > > > I would like to call for a vote on KIP-380:
> > > >
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 380%3A+Detect+outdated+control+requests+and+bounced+brokers+using+broker+
> generation
> > > >
> > > > Here is the discussion thread:
> > > >
> > > >
> > https://lists.apache.org/thread.html/2497114df64993342eaf9c78c0f14b
> f8c1795bc3305f13b03dd39afd@%3Cdev.kafka.apache.org%3E
> > > > KIP-380
> > > > <
> > https://lists.apache.org/thread.html/2497114df64993342eaf9c78c0f14b
> f8c1795bc3305f13b03dd39afd@%3Cdev.kafka.apache.org%3EKIP-380
> > >:
> > > > Detect outdated control requests and bounced ...<
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 380%3A+Detect+outdated+control+requests+and+bounced+brokers+using+broker+
> generation
> > > > >
> > > > Note: Normalizing the schema is a good-to-have optimization because
> the
> > > > memory footprint for the control requests hinders the controller from
> > > > scaling up if we have many topics with large partition counts.
> > > > cwiki.apache.org
> > > >
> > > >
> > > >
> > > > Thanks,
> > > > Zhanxiang (Patrick) Huang
> > > >
> >
>

Re: [VOTE] 2.1.0 RC0

[Ted Yu]

+1

InternalTopicIntegrationTest failed during test suite run but passed with
rerun.

On Wed, Oct 24, 2018 at 3:48 AM Andras Beni 
wrote:

> +1 (non-binding)
>
> Verified signatures and checksums of release artifacts
> Performed quickstart steps on rc artifacts (both scala 2.11 and 2.12) and
> one built from tag 2.1.0-rc0
>
> Andras
>
> On Wed, Oct 24, 2018 at 10:17 AM Dong Lin  wrote:
>
> > Hello Kafka users, developers and client-developers,
> >
> > This is the first candidate for feature release of Apache Kafka 2.1.0.
> >
> > This is a major version release of Apache Kafka. It includes 28 new KIPs
> > and
> >
> > critical bug fixes. Please see the Kafka 2.1.0 release plan for more
> > details:
> >
> > *
> >
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=91554044*
> > <
> https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=91554044
> > >
> >
> > Here are a few notable highlights:
> >
> > - Java 11 support
> > - Support for Zstandard, which achieves compression comparable to gzip
> with
> > higher compression and especially decompression speeds(KIP-110)
> > - Avoid expiring committed offsets for active consumer group (KIP-211)
> > - Provide Intuitive User Timeouts in The Producer (KIP-91)
> > - Kafka's replication protocol now supports improved fencing of zombies.
> > Previously, under certain rare conditions, if a broker became partitioned
> > from Zookeeper but not the rest of the cluster, then the logs of
> replicated
> > partitions could diverge and cause data loss in the worst case (KIP-320)
> > - Streams API improvements (KIP-319, KIP-321, KIP-330, KIP-353, KIP-356)
> > - Admin script and admin client API improvements to simplify admin
> > operation (KIP-231, KIP-308, KIP-322, KIP-324, KIP-338, KIP-340)
> > - DNS handling improvements (KIP-235, KIP-302)
> >
> > Release notes for the 2.1.0 release:
> > http://home.apache.org/~lindong/kafka-2.1.0-rc0/RELEASE_NOTES.html
> >
> > *** Please download, test and vote ***
> >
> > * Kafka's KEYS file containing PGP keys we use to sign the release:
> > http://kafka.apache.org/KEYS
> >
> > * Release artifacts to be voted upon (source and binary):
> > http://home.apache.org/~lindong/kafka-2.1.0-rc0/
> >
> > * Maven artifacts to be voted upon:
> > https://repository.apache.org/content/groups/staging/
> >
> > * Javadoc:
> > http://home.apache.org/~lindong/kafka-2.1.0-rc0/javadoc/
> >
> > * Tag to be voted upon (off 2.1 branch) is the 2.1.0-rc0 tag:
> > https://github.com/apache/kafka/tree/2.1.0-rc0
> >
> > * Documentation:
> > *http://kafka.apache.org/21/documentation.html*
> > 
> >
> > * Protocol:
> > http://kafka.apache.org/21/protocol.html
> >
> > * Successful Jenkins builds for the 2.1 branch:
> > Unit/integration tests: *
> https://builds.apache.org/job/kafka-2.1-jdk8/38/
> > *
> >
> > Please test and verify the release artifacts and submit a vote for this
> RC,
> > or report any issues so we can fix them and get a new RC out ASAP.
> Although
> > this release vote requires PMC votes to pass, testing, votes, and bug
> > reports are valuable and appreciated from everyone.
> >
> > Cheers,
> > Dong
> >
>

Requesting wiki permissions

[Zahari Dichev]

Hi there Can I get permissions for
WIKI id: zaharidichev

Zahari

Re: Throwing away prefetched records optimisation.

[Colin McCabe]

In general, the official discussion thread for a KIP starts after the KIP is 
written and posted.  So you would typically start a new email thread with a 
[DISCUSS] string in the title somewhere.  You can certainly link back to this 
email thread if you want, though, since it has some useful context for 
everything.

best,
Colin


On Tue, Oct 23, 2018, at 23:21, Zahari Dichev wrote:
> Colin, I agree
> 
> I will take a closer looks at the Fetcher itself and see whether that is
> feasible and update my KIP accordingly. I guess we can label this one, the
> official discussion thread for it or should I start another one ?
> 
> Zahari
> 
> On Wed, Oct 24, 2018 at 6:43 AM Colin McCabe  wrote:
> 
> > On Tue, Oct 23, 2018, at 12:38, Zahari Dichev wrote:
> > > Hi there Matthias, I looked through the code of Kafka Streams. Quite
> > > impressive work ! If I have to put the logic of buffering within the
> > > context of what we are doing in Akka though, I might end up with the
> > > following situation.
> > >
> > > 1. Poll is called with two partition being active *TP1, TP2*
> > > 2. We get some data for both, both of them also prefetch some data.
> > > 3. So now we have some data that we have obtained and some data that sits
> > > with the buffer of the fetcher, waiting to be obtained.
> > > 4. We put the data that we have obtained from the poll into the
> > respective
> > > buffers of the partitions.
> > > 5. Since both of our buffers are "full", we call pause on both *TP1* and
> > > *TP2*.
> > > 6. A little time has passed and the client of *TP1* has processed all its
> > > records from the buffer, while the one of *TP2* has processed none
> > > 7. Buffer of *TP1* gets empty, we call resume on *TP1*
> > > 8. We call poll again with *TP1* resumed and *TP2* paused.
> > > 9. We get some records for TP1 and we throw away all the records that
> > were
> > > prefetched for *TP2* in step 2
> > >
> > > This can go on and on and due to the dynamic nature of the speed of
> > > processing records and the theoretically unlimited number of topic
> > > partitions, I find it possible that this scenario can happen more than
> > once
> > > over the lifetime of a client. And instead of trying to calculate the
> > > probability of this happening and attempt to minimise it, I would prefer
> > to
> > > have one of two options:
> > >
> > > 1. Having control to allow me to enable the returning of already
> > prefetched
> > > data, and simply store it in a buffer of my own until I have enough
> > > capacity to deal with it
> > >
> > > OR
> > >
> > > 2. Keep the data in the fetcher and not throw it away but use it on the
> > > next poll (not sure how viable that is as I have not looked at the
> > details
> > > of it all).
> >
> > I haven't thought about it that hard, but it sounds like the second option
> > might be better.  I have a hard time thinking of a case where we actually
> > want to throw away data for paused partitions.  If you're still subscribed
> > to it, presumably you'll eventually unpause it and use the cache, right?
> > It makes sense for unsubscribe to clear those records, but not pause, as
> > far as I can see.
> >
> > best,
> > Colin
> >
> >
> > >
> > > The first option is what I suggested initially and the second option is
> > the
> > > one that will allow us to skip the introduction of a configuration
> > > parameter as Colin suggested. These are the things I can suggest at the
> > > moment. As mentioned, I am willing to carry out the work. There is also
> > an
> > > official discussion thread, but I guess we have deviated from that, so I
> > > can just put that current on in JIRA instead if that is OK ?
> > >
> > > Matthias, regarding how the fetcher works. From what I have looked at,
> > > whenever the consumer polls and returns some data, we immediately issue
> > > another fetch request that delivered us records that are returned on the
> > > next poll. All these fetched records, that have not made it to the caller
> > > of poll but have been fetched are thrown away in case at the time of the
> > > nest poll() the partition is in paused state. This is what is causing the
> > > inefficiency.
> > >
> > > Any more comments are welcome.
> > >
> > > On Mon, Oct 22, 2018 at 6:00 AM Ismael Juma  wrote:
> > >
> > > > Hi,
> > > >
> > > > I think a KIP to discuss a concrete proposal makes sense. One
> > suggestion is
> > > > to explore the possibility of fixing the issue without a new config.
> > Would
> > > > that break existing users? Generally, we should strive for avoiding
> > configs
> > > > if at all possible.
> > > >
> > > > Ismael
> > > >
> > > > On 16 Oct 2018 12:30 am, "Zahari Dichev" 
> > wrote:
> > > >
> > > > Hi there Kafka developers,
> > > >
> > > > I am currently trying to find a solution to an issue that has been
> > > > manifesting itself in the Akka streams implementation of the Kafka
> > > > connector. When it comes to consuming messages, the implementation
> > relies
> > > > heavily on the fact that we 

Re: Requesting wiki permissions

[Matthias J. Sax]

Done.

On 10/24/18 11:49 AM, Zahari Dichev wrote:
> Hi there Can I get permissions for
> WIKI id: zaharidichev
> 
> Zahari
> 



signature.asc
Description: OpenPGP digital signature

Jenkins build is back to normal : kafka-trunk-jdk11 #53

[Apache Jenkins Server]

See 

Re: KAFKA-3932 - Consumer fails to consume in a round robin fashion

[Matthias J. Sax]

CH,

Thanks for contributing to Kafka. Do you have a Jira account already? If
yes, what is your account id? If not, you need to create one first and
share your id so we can grant permission to self-assign tickets.

I was just looking into the ticket itself, and it's marked as 0.10.0.0.
You say you encountered this issues. Do you use 0.10.0.x version? AFAIK,
the consumer was updated in later versions, and the behavior should be
different. Before you start working on the ticket, we should verify that
it is not already fixed. For this case, we would just resolve the ticket
with corresponding fixed version.

Note, that the behavior (at least from my point of view) is not a bug,
but addressing it would be an improvement. Thus, if you work on it, the
patch would be released with 2.2.0 version, but _not_ with a potential
0.10.0.2 release.

Does this make sense?


-Matthias

On 10/24/18 6:27 AM, ChienHsing Wu wrote:
> I don't see any comments/concerns. I would like to implement and commit to 
> this ticket. Could anyone let me know how to request for the permission to 
> assign that ticket to me?
> 
> Thanks, CH
> 
> From: ChienHsing Wu
> Sent: Monday, October 22, 2018 1:40 PM
> To: 'dev@kafka.apache.org' 
> Subject: KAFKA-3932 - Consumer fails to consume in a round robin fashion
> 
> 
> Hi,
> 
> 
> 
> I encountered the issue documented in the jira 
> KAFKA-3932.
>  Upon studying the source code and the 
> PIP,
>  I think the issues is the statement in PIP: "As before, we'd keep track of 
> which partition we left off at so that the next iteration would begin there." 
> I think it should NOT use the last partition in the next iteration; it should 
> pick the next one instead.
> 
> If this behavior is agreeable, the simplest solution to impose the order to 
> pick the next one is to use the order the consumer.internals.Fetcher receives 
> the partition messages, as determined by completedFetches queue in that 
> class. To avoid parsing the partition messages repeatedly. we can save those 
> parsed fetches to a list and maintain the next partition to get messages 
> there.
> 
> Does it sound like a good approach? If this is not the right place to discuss 
> the design please let me know where to engage. If this is agreeable I can 
> contribute the implementation.
> 
> 
> 
> Thanks, CH
> 
> 



signature.asc
Description: OpenPGP digital signature

[jira] [Created] (KAFKA-7539) ConsumerBounceTest.testClose transient failure

[John Roesler (JIRA)]

John Roesler created KAFKA-7539:
---

 Summary: ConsumerBounceTest.testClose transient failure
 Key: KAFKA-7539
 URL: https://issues.apache.org/jira/browse/KAFKA-7539
 Project: Kafka
  Issue Type: Sub-task
Reporter: John Roesler
Assignee: Fangmin Lv
 Fix For: 0.9.0.0


{code}
kafka.api.ConsumerBounceTest > testSeekAndCommitWithBrokerFailures FAILED
java.lang.AssertionError: expected:<1000> but was:<976>
at org.junit.Assert.fail(Assert.java:92)
at org.junit.Assert.failNotEquals(Assert.java:689)
at org.junit.Assert.assertEquals(Assert.java:127)
at org.junit.Assert.assertEquals(Assert.java:514)
at org.junit.Assert.assertEquals(Assert.java:498)
at 
kafka.api.ConsumerBounceTest.seekAndCommitWithBrokerFailures(ConsumerBounceTest.scala:117)
at 
kafka.api.ConsumerBounceTest.testSeekAndCommitWithBrokerFailures(ConsumerBounceTest.scala:98)

kafka.api.ConsumerBounceTest > testSeekAndCommitWithBrokerFailures FAILED
java.lang.AssertionError: expected:<1000> but was:<913>
at org.junit.Assert.fail(Assert.java:92)
at org.junit.Assert.failNotEquals(Assert.java:689)
at org.junit.Assert.assertEquals(Assert.java:127)
at org.junit.Assert.assertEquals(Assert.java:514)
at org.junit.Assert.assertEquals(Assert.java:498)
at 
kafka.api.ConsumerBounceTest.seekAndCommitWithBrokerFailures(ConsumerBounceTest.scala:117)
at 
kafka.api.ConsumerBounceTest.testSeekAndCommitWithBrokerFailures(ConsumerBounceTest.scala:98)
{code}



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Resolved] (KAFKA-7539) ConsumerBounceTest.testClose transient failure

[John Roesler (JIRA)]

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

John Roesler resolved KAFKA-7539.
-
Resolution: Invalid
  Reviewer:   (was: Jason Gustafson)

Cloning to create this issue was a mistake. I'm going to create a fresh one.

> ConsumerBounceTest.testClose transient failure
> --
>
> Key: KAFKA-7539
> URL: https://issues.apache.org/jira/browse/KAFKA-7539
> Project: Kafka
>  Issue Type: Sub-task
>Reporter: John Roesler
>Priority: Major
>  Labels: flaky-test
>
> {code:java}
> java.lang.ArrayIndexOutOfBoundsException: -1
>   at 
> kafka.integration.KafkaServerTestHarness.killBroker(KafkaServerTestHarness.scala:146)
>   at 
> kafka.api.ConsumerBounceTest.checkCloseWithCoordinatorFailure(ConsumerBounceTest.scala:238)
>   at kafka.api.ConsumerBounceTest.testClose(ConsumerBounceTest.scala:211)
>   at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>   at 
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:498)
>   at 
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
>   at 
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
>   at 
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
>   at 
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
>   at 
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
>   at 
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
>   at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
>   at 
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
>   at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
>   at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
>   at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
>   at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
>   at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
>   at 
> org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
>   at 
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
>   at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
>   at 
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
>   at 
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
>   at 
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
>   at 
> org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
>   at 
> org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
>   at sun.reflect.GeneratedMethodAccessor12.invoke(Unknown Source)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:498)
>   at 
> org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
>   at 
> org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
>   at 
> org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
>   at 
> org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
>   at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
>   at 
> org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:117)
>   at sun.reflect.GeneratedMethodAccessor11.invoke(Unknown Source)
>   at 
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>   at java.lang.reflect.Method.invoke(Method.java:498)
>   at 
> org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
>   at 
> org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
>   at 
> org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:155)
>   at 
> org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConne

Re: [VOTE] KIP-369 Alternative Partitioner to Support "Always Round-Robin" Selection

[M. Manna]

Hey IJ,

Thanks for your interest in the KIP.

My point was simply that the round-robin should happen even if the key is
not null. As for the importance of key in our case, we treat the key as
metadata. Each key is composed of certain info which are parsed by our
consumer thread. We will then determine whether it's an actionable message
(e.g. process it), or a loopback(ignore it). You could argue, "Why not
append this metadata with the record and parse it there?". But that means
the following:

1) I'm always passing null key to achieve this - I would like to pass
Null/Not-Null/Other key i.e. flexibility
2) Suppose the message size is 99 KB and and max message bytes allowed is
100K. Now prefixing metadata with message results into the actual message
being 101K. This will fail at producer level and cause a retry/log this in
our DB for future pickup.

To avoid all these, we are simply proposing this new partitioner class. but
all Kafka new releases will still have DefaultPartitioner as default,
unless they change the prop file to use our new class.

Regards,

On Sun, 21 Oct 2018 at 04:05, Ismael Juma  wrote:

> Thanks for the KIP. Can you please elaborate on the need for the key in
> this case? The KIP simply states that the key is needed for metadata, but
> doesn't give any more details.
>
> Ismael
>
> On Tue, Sep 4, 2018 at 3:39 AM M. Manna  wrote:
>
> > Hello,
> >
> > I have made necessary changes as per the original discussion thread, and
> > would like to put it for votes.
> >
> > Thank you very much for your suggestion and guidance so far.
> >
> > Regards,
> >
>

[jira] [Created] (KAFKA-7540) Transient failure: kafka.api.ConsumerBounceTest.testClose

[John Roesler (JIRA)]

John Roesler created KAFKA-7540:
---

 Summary: Transient failure: kafka.api.ConsumerBounceTest.testClose
 Key: KAFKA-7540
 URL: https://issues.apache.org/jira/browse/KAFKA-7540
 Project: Kafka
  Issue Type: Bug
Reporter: John Roesler


Observed on Java 8: 
[https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/17314/testReport/junit/kafka.api/ConsumerBounceTest/testClose/]

 

Stacktrace:
{noformat}
java.lang.ArrayIndexOutOfBoundsException: -1
at 
kafka.integration.KafkaServerTestHarness.killBroker(KafkaServerTestHarness.scala:146)
at 
kafka.api.ConsumerBounceTest.checkCloseWithCoordinatorFailure(ConsumerBounceTest.scala:238)
at kafka.api.ConsumerBounceTest.testClose(ConsumerBounceTest.scala:211)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at 
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at 
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at 
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at 
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
at 
org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
at 
org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
at sun.reflect.GeneratedMethodAccessor12.invoke(Unknown Source)
at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
at 
org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
at 
org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:117)
at sun.reflect.GeneratedMethodAccessor11.invoke(Unknown Source)
at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:155)
at 
org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnection$DispatchWrapper.dispatch(MessageHubBackedObjectConnection.java:137)
at 
org.gradle.internal.remote.internal.hub.MessageHub$Handler.run(MessageHub.java:404)
at 
org.gradle.internal.concurrent.ExecutorPolicy$CatchAndRecordFailures.onExecute(ExecutorPoli

[jira] [Created] (KAFKA-7541) Transient Failure: kafka.server.DynamicBrokerReconfigurationTest.testUncleanLeaderElectionEnable

[John Roesler (JIRA)]

John Roesler created KAFKA-7541:
---

 Summary: Transient Failure: 
kafka.server.DynamicBrokerReconfigurationTest.testUncleanLeaderElectionEnable
 Key: KAFKA-7541
 URL: https://issues.apache.org/jira/browse/KAFKA-7541
 Project: Kafka
  Issue Type: Bug
Reporter: John Roesler


Observed on Java 11: 
[https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/264/testReport/junit/kafka.server/DynamicBrokerReconfigurationTest/testUncleanLeaderElectionEnable/]

 

Stacktrace:
{noformat}
java.lang.AssertionError: Unclean leader not elected
at org.junit.Assert.fail(Assert.java:88)
at org.junit.Assert.assertTrue(Assert.java:41)
at 
kafka.server.DynamicBrokerReconfigurationTest.testUncleanLeaderElectionEnable(DynamicBrokerReconfigurationTest.scala:487)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at 
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at 
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at 
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
at 
org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
at 
org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
at 
org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
at 
org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:117)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.remote.internal.hub.MessageHubBackedObjectConnec

[jira] [Created] (KAFKA-7543) Kafka Connect JDBC Sink failing to establish connection to Topic, while the connection is working fine with standalone consumer

[Kashyap Ivaturi (JIRA)]

Kashyap Ivaturi created KAFKA-7543:
--

 Summary: Kafka Connect JDBC Sink failing to establish connection 
to Topic, while the connection is working fine with standalone consumer
 Key: KAFKA-7543
 URL: https://issues.apache.org/jira/browse/KAFKA-7543
 Project: Kafka
  Issue Type: Task
  Components: KafkaConnect
Reporter: Kashyap Ivaturi


Hi,

I'am trying to build Kafka Connect JDBC Sink Connector to have my DB updated 
with the data I get in Kafka Topic. I had implemented JDBC Source Connectors 
before which worked very well but in this case when I try to run the Sink 
Connector its internally failing to connect to the Topic and disconnecting from 
the Kafka broker and this is happening in loop. When I have enabled TRACE I got 
below details in the log. Any idea why the consumer is unable to connect to the 
Topic?. Actually when I have used a standalone consumer from my another 
application it worked pretty well in connecting to the Topic and reading 
messages from it. Please let me know if you have any suggestions.

 

[2018-10-24 23:03:24,134] INFO WorkerSinkTask\{id=hrmsAckEvents-0} Sink task 
finished initialization and start 
(org.apache.kafka.connect.runtime.WorkerSinkTask:268)

[2018-10-24 23:03:24,135] TRACE WorkerSinkTask\{id=hrmsAckEvents-0} Polling 
consumer with timeout 4875 ms 
(org.apache.kafka.connect.runtime.WorkerSinkTask:282)

[2018-10-24 23:03:24,136] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,136] DEBUG [Consumer clientId=consumer-1, groupId=hrmsack] 
Sending GroupCoordinator request to broker messaging-rtp3.cisco.com:9093 (id: 
-1 rack: null) 
(org.apache.kafka.clients.consumer.internals.AbstractCoordinator:183)

[2018-10-24 23:03:24,281] DEBUG [Consumer clientId=consumer-1, groupId=hrmsack] 
Initiating connection to node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:183)

[2018-10-24 23:03:24,293] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,295] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,346] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,365] DEBUG Added sensor with name node--1.bytes-sent 
(org.apache.kafka.common.metrics.Metrics:404)

[2018-10-24 23:03:24,367] DEBUG Added sensor with name node--1.bytes-received 
(org.apache.kafka.common.metrics.Metrics:404)

[2018-10-24 23:03:24,374] DEBUG Added sensor with name node--1.latency 
(org.apache.kafka.common.metrics.Metrics:404)

[2018-10-24 23:03:24,376] DEBUG [Consumer clientId=consumer-1, groupId=hrmsack] 
Created socket with SO_RCVBUF = 65536, SO_SNDBUF = 131072, SO_TIMEOUT = 0 to 
node -1 (org.apache.kafka.common.network.Selector:195)

[2018-10-24 23:03:24,377] DEBUG [Consumer clientId=consumer-1, groupId=hrmsack] 
Completed connection to node -1. Fetching API versions. 
(org.apache.kafka.clients.NetworkClient:183)

[2018-10-24 23:03:24,377] DEBUG [Consumer clientId=consumer-1, groupId=hrmsack] 
Initiating API versions fetch from node -1. 
(org.apache.kafka.clients.NetworkClient:183)

[2018-10-24 23:03:24,378] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
No version information found when sending API_VERSIONS with correlation id 1 to 
node -1. Assuming version 1. (org.apache.kafka.clients.NetworkClient:135)

[2018-10-24 23:03:24,380] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Sending API_VERSIONS {} with correlation id 1 to node -1 
(org.apache.kafka.clients.NetworkClient:135)

[2018-10-24 23:03:24,385] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,389] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,724] TRACE [Consumer clientId=consumer-1, groupId=hrmsack] 
Found least loaded node messaging-rtp3.cisco.com:9093 (id: -1 rack: null) 
(org.apache.kafka.clients.NetworkClient:123)

[2018-10-24 23:03:24,725] DEBUG [Consumer clientId=consumer-1, groupId=hrmsack] 
Connection with messaging-rtp3.cisco.com/64.101.96.6 disconnected 
(org.apache.kafka.common.network.Selector:189)

java.io.EOFException

at 
org.apache.kafka.common.network.NetworkReceive.readFromReada

[jira] [Created] (KAFKA-7542) Transient Failure: kafka.server.GssapiAuthenticationTest.testServerAuthenticationFailure

[John Roesler (JIRA)]

John Roesler created KAFKA-7542:
---

 Summary: Transient Failure: 
kafka.server.GssapiAuthenticationTest.testServerAuthenticationFailure
 Key: KAFKA-7542
 URL: https://issues.apache.org/jira/browse/KAFKA-7542
 Project: Kafka
  Issue Type: Bug
Reporter: John Roesler


Observed on java 11: 
[https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/264/testReport/junit/kafka.server/GssapiAuthenticationTest/testServerAuthenticationFailure/]

at 
[https://github.com/apache/kafka/pull/5795/commits/5bdcd0e023c6f406d585155399f6541bb6a9f9c2]

 

Stacktrace:
{noformat}
org.scalatest.junit.JUnitTestFailedError: 
at 
org.scalatest.junit.AssertionsForJUnit$class.newAssertionFailedException(AssertionsForJUnit.scala:100)
at 
org.scalatest.junit.JUnitSuite.newAssertionFailedException(JUnitSuite.scala:71)
at org.scalatest.Assertions$class.fail(Assertions.scala:1075)
at org.scalatest.junit.JUnitSuite.fail(JUnitSuite.scala:71)
at 
kafka.server.GssapiAuthenticationTest.testServerAuthenticationFailure(GssapiAuthenticationTest.scala:128)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at 
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at 
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at 
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at 
org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
at 
org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
at 
org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
at 
org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
at 
org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:117)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.i

[jira] [Created] (KAFKA-7544) Transient Failure: org.apache.kafka.streams.integration.EosIntegrationTest.shouldNotViolateEosIfOneTaskFails

[John Roesler (JIRA)]

John Roesler created KAFKA-7544:
---

 Summary: Transient Failure: 
org.apache.kafka.streams.integration.EosIntegrationTest.shouldNotViolateEosIfOneTaskFails
 Key: KAFKA-7544
 URL: https://issues.apache.org/jira/browse/KAFKA-7544
 Project: Kafka
  Issue Type: Bug
  Components: streams
Reporter: John Roesler


Observed on Java 11: 
[https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/264/testReport/junit/org.apache.kafka.streams.integration/EosIntegrationTest/shouldNotViolateEosIfOneTaskFails/]

at 
[https://github.com/apache/kafka/pull/5795/commits/5bdcd0e023c6f406d585155399f6541bb6a9f9c2]

 

stacktrace:
{noformat}
java.lang.AssertionError: 
Expected: <[KeyValue(0, 0), KeyValue(0, 1), KeyValue(0, 2), KeyValue(0, 3), 
KeyValue(0, 4), KeyValue(0, 5), KeyValue(0, 6), KeyValue(0, 7), KeyValue(0, 8), 
KeyValue(0, 9)]>
 but: was <[KeyValue(0, 0), KeyValue(0, 1), KeyValue(0, 2), KeyValue(0, 3), 
KeyValue(0, 4), KeyValue(0, 5), KeyValue(0, 6), KeyValue(0, 7), KeyValue(0, 8), 
KeyValue(0, 9), KeyValue(0, 10), KeyValue(0, 11), KeyValue(0, 12), KeyValue(0, 
13), KeyValue(0, 14)]>
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:8)
at 
org.apache.kafka.streams.integration.EosIntegrationTest.checkResultPerKey(EosIntegrationTest.java:218)
at 
org.apache.kafka.streams.integration.EosIntegrationTest.shouldNotViolateEosIfOneTaskFails(EosIntegrationTest.java:360)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at 
org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at 
org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at 
org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at 
org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at 
org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
at org.junit.rules.RunRules.evaluate(RunRules.java:20)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
at 
org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
at 
org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
at 
org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
at 
org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
at 
org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
at 
org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:93)
at com.sun.proxy.$Proxy2.processTestClass(Unknown Source)
at 
org.gradle.api.internal.tasks.testing.worker.TestWorker.processTestClass(TestWorker.java:117)
at 
java.base/jdk.internal.reflect.NativeMethodAccessorImp

Re: [VOTE] KIP-380: Detect outdated control requests and bounced brokers using broker generation

[Patrick Huang]

Hi Jun,

Sure. I already updated the KIP. Thanks!

Best,
Zhanxiang (Patrick) Huang


From: Jun Rao 
Sent: Wednesday, October 24, 2018 14:17
To: dev
Subject: Re: [VOTE] KIP-380: Detect outdated control requests and bounced 
brokers using broker generation

Hi, Patrick,

Could you update the KIP with the changes to ControlledShutdownRequest
based on the discussion thread?

Thanks,

Jun


On Sun, Oct 21, 2018 at 2:25 PM, Mickael Maison 
wrote:

> +1( non-binding)
> Thanks for the KIP!
>
> On Sun, Oct 21, 2018, 03:31 Harsha Chintalapani  wrote:
>
> > +1(binding). LGTM.
> > -Harsha
> > On Oct 20, 2018, 4:49 PM -0700, Dong Lin , wrote:
> > > Thanks much for the KIP Patrick. Looks pretty good.
> > >
> > > +1 (binding)
> > >
> > > On Fri, Oct 19, 2018 at 10:17 AM Patrick Huang 
> > wrote:
> > >
> > > > Hi All,
> > > >
> > > > I would like to call for a vote on KIP-380:
> > > >
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 380%3A+Detect+outdated+control+requests+and+bounced+brokers+using+broker+
> generation
> > > >
> > > > Here is the discussion thread:
> > > >
> > > >
> > https://lists.apache.org/thread.html/2497114df64993342eaf9c78c0f14b
> f8c1795bc3305f13b03dd39afd@%3Cdev.kafka.apache.org%3E
> > > > KIP-380
> > > > <
> > https://lists.apache.org/thread.html/2497114df64993342eaf9c78c0f14b
> f8c1795bc3305f13b03dd39afd@%3Cdev.kafka.apache.org%3EKIP-380
> > >:
> > > > Detect outdated control requests and bounced ...<
> > > >
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-
> 380%3A+Detect+outdated+control+requests+and+bounced+brokers+using+broker+
> generation
> > > > >
> > > > Note: Normalizing the schema is a good-to-have optimization because
> the
> > > > memory footprint for the control requests hinders the controller from
> > > > scaling up if we have many topics with large partition counts.
> > > > cwiki.apache.org
> > > >
> > > >
> > > >
> > > > Thanks,
> > > > Zhanxiang (Patrick) Huang
> > > >
> >
>

Jenkins build is back to normal : kafka-trunk-jdk8 #3158

[Apache Jenkins Server]

See 

Re: [VOTE] KIP-377: TopicCommand to use AdminClient

[Colin McCabe]

Thanks, Viktor.  +1 (binding).

One note: can we add a deprecation warning when --zookeeper is used, to 
indicate that this option will be phased out in the future?

best,
Colin

On Wed, Oct 24, 2018, at 05:47, Mickael Maison wrote:
> +1 (non-binding)
> Thanks for the KIP!
> On Wed, Oct 24, 2018 at 1:28 PM Viktor Somogyi-Vass
>  wrote:
> >
> > Hi All,
> >
> > I'd like to start a vote on KIP-377:
> > https://cwiki.apache.org/confluence/display/KAFKA/KIP-377%3A+TopicCommand+to+use+AdminClient.
> >
> > Summary:
> > The KIP basically proposes to add --bootstrap-server and
> > --command-config option to TopicsCommand and implement topic
> > administration with AdminClient in a backwards compatible way (so
> > wouldn't drop or change the --zookeeper option usage).
> >
> > I'd appreciate any votes or feedback.
> >
> > Viktor

[jira] [Created] (KAFKA-7545) Use auth_to_local rules from krb5.conf

[Pradeep Bansal (JIRA)]

Pradeep Bansal created KAFKA-7545:
-

 Summary: Use auth_to_local rules from krb5.conf
 Key: KAFKA-7545
 URL: https://issues.apache.org/jira/browse/KAFKA-7545
 Project: Kafka
  Issue Type: Improvement
  Components: security
Reporter: Pradeep Bansal


Currently I have to replicate all auth_to_local rules from my krb5.conf and 
pass it to sasl.kerberos.principal.to.local.rules to make them work. This is 
causing maintenance issue.

 

It will be very helpful/useful if kafka can read auth_to_local rules from 
krb5.conf directly.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7546) Java implementation for Authorizer

[Pradeep Bansal (JIRA)]

Pradeep Bansal created KAFKA-7546:
-

 Summary: Java implementation for Authorizer
 Key: KAFKA-7546
 URL: https://issues.apache.org/jira/browse/KAFKA-7546
 Project: Kafka
  Issue Type: Improvement
  Components: security
Reporter: Pradeep Bansal


I am using kafka with authentication and authorization. I wanted to plugin my 
own implementation of Authorizer which doesn't use zookeeper instead has 
permission mapping in SQL database. Is it possible to write Authorizer code in 
Java?



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7547) Avoid relogin in kafka if connection is already established.

[Pradeep Bansal (JIRA)]

Pradeep Bansal created KAFKA-7547:
-

 Summary: Avoid relogin in kafka if connection is already 
established.
 Key: KAFKA-7547
 URL: https://issues.apache.org/jira/browse/KAFKA-7547
 Project: Kafka
  Issue Type: Improvement
  Components: security
Reporter: Pradeep Bansal


I am new to kafka and may be there are ways already there for my requirement. I 
didn't find a way so far and hence I though I will post it here.

Currently, I observed that kafka periodically tries to renew kerberos token 
using kinit -R command. I found that I can set 
sasl.kerberos.min.time.before.relogin and change default from 1 minute to 1 day 
max. But in my case I am not clear on why renew is even required.

 

If it is not really required is there a way to turn it off?



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)

[jira] [Created] (KAFKA-7548) KafkaConsumer should not throw away already fetched data for paused partitions.

[Mayuresh Gharat (JIRA)]

Mayuresh Gharat created KAFKA-7548:
--

 Summary: KafkaConsumer should not throw away already fetched data 
for paused partitions.
 Key: KAFKA-7548
 URL: https://issues.apache.org/jira/browse/KAFKA-7548
 Project: Kafka
  Issue Type: Improvement
  Components: clients
Reporter: Mayuresh Gharat
Assignee: Mayuresh Gharat


In KafkaConsumer, when we do a poll, we fetch data asynchronously from kafka 
brokers that is buffered in completedFetch queue. Now if we pause a few 
partitions, it seems that in next call to poll we remove the completedFetches 
for those paused partitions. Normally, if an application is calling pause on 
topicPartitions, it is likely to return to those topicPartitions in near future 
and when it does, with the current design we would have to re-fetch that data.

At Linkedin, we made a hotfix to see if NOT throwing away the prefetched data 
would improve the performance for stream applications like Samza. We ran a 
benchmark were we compared what is the throughput w.r.t to different values of 
maxPollRecords.

We had a consumer subscribed to 10 partitions of a high volume topic and paused 
different number of partitions for every poll call. Here are the results :

*Before fix (records consumed)*
|maxPollRecords->
Number of Partitions
Paused
\|
V|10|5|1|
|0|8605320
(60.022276059 sec)|8337690
(60.026690095 sec)|6424753
(60.67003 sec)|
|2|101910
(60.006989628 sec)|49350
(60.022598668 sec)|10495
(60.020077555 sec)|
|4|48420
(60.022096537 sec)|24850
(60.007451162 sec)|5004
(60.009773507 sec) |
|6|30420
(60.018380086 sec)|15385
(60.011912135 sec)|3152
(60.013573487 sec)|
|8|23390
(60.043122495 sec)|11390
(60.013297496 sec)|2237
(60.038921333 sec)|
|9|20230 (60.026183204 sec)|10355
(60.015584914 sec)|2087
(60.00319069 sec)|

 

*After fix (records consumed)*
|Number of Partitions
Paused / maxPollRecords|10|5|1|
|0|8662740 (60.011527576 sec)|8203445
(60.022204036 sec)|5846512
(60.0168916 sec)|
|2|8257390
(60.011121061 sec)|7776150
(60.01620875 sec)|5269557
(60.022581248 sec)|
|4|7938510
(60.011829002 sec)|7510140
(60.017571391 sec)|5213496
(60.000230139 sec)|
|6|7100970
(60.007220465 sec)|6382845
(60.038580526 sec)|4519645
(60.48034 sec)|
|8|6799956 (60.001850171 sec)|6482421
(60.001997219 sec)|4383300 (60.4836 sec)|
|9|7045177 (60.035366096 sec)|6465839 
(60.41961 sec)|4884693
(60.42054 sec)|



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)