[jira] [Resolved] (KAFKA-10297) Don't use deprecated producer config `retries`

[Matthias J. Sax (Jira)]

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

Matthias J. Sax resolved KAFKA-10297.
-
Resolution: Invalid

We change the KIP to not deprecate `retries`. Hence, this ticket is invalid now.

Cf [https://github.com/apache/kafka/pull/9333] 

> Don't use deprecated producer config `retries`
> --
>
> Key: KAFKA-10297
> URL: https://issues.apache.org/jira/browse/KAFKA-10297
> Project: Kafka
>  Issue Type: Improvement
>  Components: KafkaConnect
>Affects Versions: 2.7.0
>Reporter: Matthias J. Sax
>Priority: Blocker
> Fix For: 2.7.0
>
>
> In 2.7.0 release, producer config `retries` gets deprecated via KIP-572.
> Connect is still using this config what needs to be fixed (cf 
> [https://github.com/apache/kafka/pull/8864/files#r439685920])
> {quote}Btw: @hachikuji raise a concern about this issue, too: 
> https://github.com/apache/kafka/pull/8864#pullrequestreview-443424531
> > I just had one question about the proposal. Using retries=0 in the producer 
> > allows the user to have "at-most-once" delivery. This allows the 
> > application to implement its own retry logic for example. Do we still have 
> > a way to do this once this configuration is gone?
> So maybe we need to do some follow up work in the `Producer` to make it work 
> for Connect. But I would defer this to the follow up work.
> My original though was, that setting `max.deliver.timeout.ms := request 
> .timeout.ms` might prevent internal retries. But we need to verify this. It 
> was also brought to my attention, that this might not work if the network 
> disconnects -- only `retries=0` would prevent to re-open the connection but a 
> low timeout would not prevent retries.
> In KIP-572, we proposed for Kafka Streams itself, to treat `task.timeout.ms = 
> 0` as "no retries" -- maybe we can do a similar thing for the producer?
> There is also `max.block.ms` that we should consider. Unfortunately, I am not 
> an expert on the `Producer`. But I would like to move forward with KIP-572 
> for now and are happy to help to resolve those questions.
> {quote}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Re: [VOTE] KIP-572: Improve timeouts and retires in Kafka Streams

[Matthias J. Sax]

Sorry for the late reply Jason, and thanks for calling it out.

After couple if direct discussion with Jason, Colin, Ismael, and John, I
agree that we should keep `retries` config for both producer and admin
client.

I updated the KIP accordingly and opened a PR to revert the deprecation:
https://github.com/apache/kafka/pull/9333


-Matthias

On 7/27/20 12:45 PM, Jason Gustafson wrote:
> Hi Matthias,
> 
> Sorry for jumping in so late here. I am trying to understand why it was
> necessary to deprecate `retries` in the producer. One of the use cases that
> I see in practice is setting `retries` to 0. This allows applications to
> control the retry semantics themselves. For example, I have seen this in
> flume. As far as I can tell, once `retries` is gone, we will not have a way
> to do the same thing. The best we can suggest to users is to enable
> idempotence so that any retries will not cause duplicates. My concern is
> that this is going to slow client upgrades with little benefit in return.
> 
> Thanks,
> Jason
> 
> 
> 
> On Mon, Jul 20, 2020 at 2:40 PM Matthias J. Sax  wrote:
> 
>> While working on the PR, we realized that the command line tool
>>
>>   bin/kafka-console-producer.sh
>>
>> has a flag `--message-send-max-retries` to set the producer's `retries`
>> config. We also need to deprecate this flag.
>>
>> I updated the KIP accordingly. Please let us know if there are any
>> concerns to this minor change to the KIP.
>>
>> Thanks.
>>
>>
>> -Matthias
>>
>> On 6/10/20 11:16 AM, Matthias J. Sax wrote:
>>> Thanks!
>>>
>>> +1 (binding) from myself.
>>>
>>>
>>> I am closing the vote as accepted with 3 binding and 3 non-binding votes.
>>>
>>> binding:
>>>  - John
>>>  - Guozhang
>>>  - Matthias
>>>
>>> non-binding:
>>>  - Sophie
>>>  - Boyang
>>>  - Bruno
>>>
>>>
>>>
>>> -Matthias
>>>
>>> On 6/4/20 5:26 PM, Matthias J. Sax wrote:
 Guozhang,

 what you propose makes sense, but this seems to get into implementation
 detail territory already?

 Thus, if there are nor further change requests to the KIP wiki page
 itself, I would like to proceed with the VOTE.


 -Matthias

 On 5/20/20 12:30 PM, Guozhang Wang wrote:
> Thanks Matthias,
>
> I agree with you on all the bullet points above. Regarding the
>> admin-client
> outer-loop retries inside partition assignor, I think we should treat
>> error
> codes differently from those two blocking calls:
>
> Describe-topic:
> * unknown-topic (3): add this topic to the to-be-created topic list.
> * leader-not-available (5): do not try to create, retry in the outer
>> loop.
> * request-timeout: break the current loop and retry in the outer loop.
> * others: fatal error.
>
> Create-topic:
> * topic-already-exists: retry in the outer loop to validate the
> num.partitions match expectation.
> * request-timeout: break the current loop and retry in the outer loop.
> * others: fatal error.
>
> And in the outer-loop, I think we can have a global timer for the whole
> "assign()" function, not only for the internal-topic-manager, and the
>> timer
> can be hard-coded with, e.g. half of the rebalance.timeout to get rid
>> of
> the `retries`; if we cannot complete the assignment before the timeout
>> runs
> out, we can return just the partial assignment (e.g. if there are two
> tasks, but we can only get the topic metadata for one of them, then
>> just do
> the assignment for that one only) while encoding in the error-code
>> field to
> request for another rebalance.
>
> Guozhang
>
>
>
> On Mon, May 18, 2020 at 7:26 PM Matthias J. Sax 
>> wrote:
>
>> No worries Guozhang, any feedback is always very welcome! My reply is
>> going to be a little longer... Sorry.
>>
>>
>>
>>> 1) There are some inconsistent statements in the proposal regarding
>> what
>> to
>>> deprecated:
>>
>> The proposal of the KIP is to deprecate `retries` for producer, admin,
>> and Streams. Maybe the confusion is about the dependency of those
>> settings within Streams and that we handle the deprecation somewhat
>> different for plain clients vs Streams:
>>
>> For plain producer/admin the default `retries` is set to MAX_VALUE.
>> The
>> config will be deprecated but still be respected.
>>
>> For Streams, the default `retries` is set to zero, however, this
>> default
>> retry does _not_ affect the embedded producer/admin clients -- both
>> clients stay on their own default of MAX_VALUES.
>>
>> Currently, this introduces the issue, that if a user wants to increase
>> Streams retries, she might by accident reduce the embedded client
>> retries, too. To avoid this issue, she would need to set
>>
>> retries=100
>> producer.retires=MAX_VALUE
>> admin.retries=MAX_VALUE
>>
>> This KIP will fix this issue only in the long term th

Build failed in Jenkins: Kafka » kafka-trunk-jdk11 #84

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-10505: Fix parsing of generation log string. (#9312)

[github] KAFKA-9450: Follow-up; Forbid process after closed (#9083)


--
[...truncated 3.35 MB...]
org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureInternalTopicNamesIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureInternalTopicNamesIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnWallClockTimeDeprecated[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnWallClockTimeDeprecated[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessRecordForTopic[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessRecordForTopic[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldForwardRecordsFromSubtopologyToSubtopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldForwardRecordsFromSubtopologyToSubtopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForSmallerValue[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForSmallerValue[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCreateStateDirectoryForStatefulTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCreateStateDirectoryForStatefulTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfWallClockTimeAdvances[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfWallClockTimeAdvances[Eos enabled = false] PASSED

org.apache.kafka.streams.test.TestRecordTest > testConsumerRecord STARTED

org.apache.kafka.streams.test.TestRecordTest > testConsumerRecord PASSED

org.apache.kafka.streams.test.TestRecordTest > testToString STARTED

org.apache.kafka.streams.test.TestRecordTest > testToString PASSED

org.apache.kafka.streams.test.TestRecordTest > testInvalidRecords STARTED

org.apache.kafka.streams.test.TestRecordTest > testInvalidRecords PASSED

org.apache.kafka.streams.test.TestRecordTest > testPartialConstructorEquals 
STARTED

org.apache.kafka.streams.test.TestRecordTest > testPartialConstructorEquals 
PASSED

org.apache.kafka.streams.test.TestRecordTest > testMultiFieldMatcher STARTED

org.apache.kafka.streams.test.TestRecordTest > testMultiFieldMatcher PASSED

org.apache.kafka.streams.test.TestRecordTest > testFields STARTED

org.apache.kafka.streams.test.TestRecordTest > testFields PASSED

org.apache.kafka.streams.test.TestRecordTest > testProducerRecord STARTED

org.apache.kafka.streams.test.TestRecordTest > testProducerRecord PASSED

org.apache.kafka.streams.test.TestRecordTest > testEqualsAndHashCode STARTED

org.apache.kafka.streams.test.TestRecordTest > testEqualsAndHashCode PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithInMemoryStore STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithInMemoryStore PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithLogging STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithLogging PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithCaching STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithCaching PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithPersistentStore STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithPersistentStore PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnIsOpen 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnIsOpen 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldDeleteAndReturnPlainValue STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldDeleteAndReturnPlainValue PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnName 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnName 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutWithUnknownTimestamp PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutAllWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest

[jira] [Created] (KAFKA-10519) Unit tests for VotedState

[Jason Gustafson (Jira)]

Jason Gustafson created KAFKA-10519:
---

 Summary: Unit tests for VotedState
 Key: KAFKA-10519
 URL: https://issues.apache.org/jira/browse/KAFKA-10519
 Project: Kafka
  Issue Type: Sub-task
Reporter: Jason Gustafson


We accidentally checked in an empty test class `VotedStateTest`. We should add 
missing unit tests.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Build failed in Jenkins: Kafka » kafka-trunk-jdk15 #83

[Apache Jenkins Server]

See 


Changes:

[github] MINOR: Remove unneeded FIXME (#9330)


--
[...truncated 6.69 MB...]

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Re: [DISCUSS] KIP-516: Topic Identifiers

[Justine Olshan]

Hi Jun,

Thanks for the comments. I apologize for some of the typos and confusion.
I’ve updated the KIP to fix some of the issues you mentioned.

20.2 I’ve changed the type to String.
20.1/3 I’ve updated the TopicZNode to fix formatting and reflect the latest
version before this change.

21. You are correct and I’ve removed this line. I’ve also added a line
mentioning an IBP bump is necessary for migration

22. I think the wording was unclear but your summary is what was intended
by this line. I’ve updated to make this point more clear. “Remove deleted
topics from replicas by sending StopReplicaRequest V3 before the IBP bump
using the old logic, and using V4 and the new logic with topic IDs after
the IBP bump.”

23. I’ve removed the unspecified type from the KIP and mention that IBP
will be used to handle this request. “IBP will be used to determine whether
this new form of the request will be used. For older requests, we will
ignore this field and default to previous behavior.”

24. I’ve fixed this typo.

25. I've added a topics at a higher level for LeaderAndIsrResponse v5,
StopReplicaResponse v4. I've also changed StopReplicaRequest v4 to have
topics at a higher level.

26. I’ve updated forgotten_topics_data--added the topic ID and removed the
topic name

27. I’ve decided on plain text, and I’ve added an example.

28. I’ve added this idea to future work.

Thanks again for taking a look,

Justine

On Wed, Sep 23, 2020 at 10:28 AM Jun Rao  wrote:

> Hi, Justine,
>
> Thanks for the response. Made another pass. A few more comments below.
>
> 20. znode schema:
> 20.1 It seems that {"name": "version", "type": "int", "id": "UUID", "doc":
> "version id"} should be {"name": "version", "type": "int"}, {"name": "id",
> "type": "UUID", "doc": "version id"}.
> 20.2 The znode format is JSON which doesn't have UUID type. So the type
> probably should be string?
> 20.3 Also, the existing format used seems outdated. It should have the
> following format.
> Json.encodeAsBytes(Map(
>   "version" -> 2,
>   "partitions" -> replicaAssignmentJson.asJava,
>   "adding_replicas" -> addingReplicasAssignmentJson.asJava,
>   "removing_replicas" -> removingReplicasAssignmentJson.asJava
> ).asJava)
>   }
>
> 21. Migration: The KIP says "The migration process can take place without
> an inter-broker protocol bump, as the format stored in
> /brokers/topics/[topic] will be compatible with older broker versions."
> However, since we are bumping up the version of inter-broker requests, it
> seems that we need to use IBP for migration.
>
> 22. The KIP says "Remove deleted topics from replicas by sending
> StopReplicaRequest V3 for any topics which do not contain a topic ID, and
> V4 for any topics which do contain a topic ID.". However, if we use IBP, it
> seems that the controller will either send StopReplicaRequest V3
> or StopReplicaRequest V4, but never mixed V3 and V4 for different topics.
> Basically, before the IBP bump, V3 will be used. After the IBP bump,
> topicId will be created and V4 will be used.
>
> 23. Given that we depend on IBP, do we still need "0 UNSPECIFIED"
> in LeaderAndIsr?
>
> 24. LeaderAndIsrResponse v5 : It still has the topic field.
>
> 25. LeaderAndIsrResponse v5, StopReplicaResponse v4: Could we use this
> opportunity to organize the response in 2 levels, first by topic, then by
> partition, as most other requests/responses?
>
> 26. FetchRequest v13 : Should forgotten_topics_data use topicId too?
>
> 27. "This file can either be plain text (key/value pairs) or JSON." Have we
> decided which one to use? Also, it would be helpful to provide an example.
>
> 28. Future improvement: Another future improvement opportunity is to use
> topicId in GroupMetadataManager.offsetCommitKey in the offset_commit topic.
> This may save some space.
>
> Thanks,
>
> Jun
>
> On Wed, Sep 23, 2020 at 8:50 AM Justine Olshan 
> wrote:
>
> > Hi Tom,
> >
> > Thanks for the comment. I think this is a really good idea and it has
> been
> > added to the KIP under the newly added tooling section.
> >
> > Thanks again,
> > Justine
> >
> > On Wed, Sep 23, 2020 at 3:17 AM Tom Bentley  wrote:
> >
> > > Hi Justine,
> > >
> > > I know you started the vote thread, but on re-reading the KIP I noticed
> > > that although the topic id is included in the MetadataResponse it's not
> > > surfaced in the output from `kafka-topics.sh --describe`. Maybe that
> was
> > > intentional because ids are intentionally not really something the user
> > > should care deeply about, but it would also make life harder for anyone
> > > debugging Kafka and this would likely get worse the more topic ids got
> > > rolled out across the protocols, clients etc. It seems likely that
> > > `kafka-topics.sh` will eventually need the ability to show the id of a
> > > topic and perhaps find a topic name given an id. Is there any reason
> not
> > to
> > > implement that in this KIP?
> > >
> > > Many thanks,
> > >
> > > Tom
> > >
> > > On Mon, Sep 21, 2020 a

Re: ApacheCon Bug Bash

[Sohil Shah]

Hi Tom,

AIPlatform is no related to Kafka. It is an end-to-end platform for
building AI applications. Data Modeling, Deployment, Evolving Data Models
using Datasets ingested in the built-in Data Lake etc.

Please feel free to include under the “Apache” repos etc. No issues at all.

Thanks
Sohil

On Wed, Sep 23, 2020 at 2:39 PM Tom DuBuisson  wrote:

> Sohil,
>
>
>
> Is bugsbunnyshah/AIPlatform related to Apache Kafka in some manner?  I'm
>
> not familiar with it.  Regardless, Muse is free to use on public
>
> repositories - all you need to do is install the muse app (
>
> https://github.com/apps/muse-dev).  Muse will then respond to pull
> requests
>
> with new bugs or you can trigger an analysis of a particular branch via
>
> console.muse.dev [1].
>
>
>
> Regarding the Bug bash, this is an ASF specific conference and bash.  If
>
> AIPlatform is hosted somewhere under the GitHub "apache" organization then
>
> we can work on its inclusion.  If not then let's email off-list about what
>
> you have in mind.
>
>
>
> Cheers,
>
> Tom
>
>
>
> [1] I've actually forked the repository, added a configuration to specify
>
> jdk11, and am trying this now.  https://github.com/TomMD/AIPlatform
>
>
>
>
>
> On Tue, Sep 22, 2020 at 4:22 PM Sohil Shah 
>
> wrote:
>
>
>
> > Hi Tom,
>
> >
>
> > I would love to participate. You can scan the AIPlatform here
>
> > https://github.com/bugsbunnyshah/AIPlatform
>
> >
>
> > Thanks
>
> > Sohil
>
> >
>
> >
>
> > On Tue, Sep 22, 2020 at 6:11 PM Tom DuBuisson  wrote:
>
> >
>
> > > Kafka Developers,
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > As part of our sponsorship of ApacheCon, our company MuseDev is doing a
>
> > Bug
>
> > >
>
> > > Bash for select Apache projects. We'll bring members of the ApacheCon
>
> > >
>
> > > community together to find and fix a range of security and performance
>
> > bugs
>
> > >
>
> > > during the conference, and gameify the experience with teams, a
>
> > >
>
> > > leaderboard, and prizes. The bash is open to everyone whether attending
>
> > the
>
> > >
>
> > > conference or not, and our whole dev team will also be participating to
>
> > >
>
> > > help fix as many bugs as we can.
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > We're seeding the bug list with results from Muse, our code analysis
>
> > >
>
> > > platform, which runs as a Github App and comments on possible bugs as
>
> > part
>
> > >
>
> > > of the pull request workflow.  Here's an example of what it looks like:
>
> > >
>
> > >
>
> > >
>
> > > https://github.com/curl/curl/pull/5971#discussion_r490252196
>
> > >
>
> > > 
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > We explored a number of Apache projects and are reaching out because
> our
>
> > >
>
> > > analysis through Muse found some interesting bugs that could be fixed
>
> > >
>
> > > during the Bash.
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > We're writing to see if you'd be interested in having your project
>
> > included
>
> > >
>
> > > in the Bash. Everything is set up on our end, and if you're interested,
>
> > we
>
> > >
>
> > > would need you to say yes on this listserv, and we’ll work with the
>
> > Apache
>
> > >
>
> > > Infrastructure team to grant Muse access to your Github mirror. We'll
>
> > then
>
> > >
>
> > > make sure it's all set-up and ready for the Bash. And of course,
> everyone
>
> > >
>
> > > on the project is most welcome to join the Bash and help us smash some
>
> > > bugs.
>
> > >
>
> > >
>
> > >
>
> > >
>
> > >
>
> > > -Tom
>
> > >
>
> > >
>
> >
>
>

Re: [DISCUSS] KIP-671: Shutdown Streams Application when appropriate exception is thrown

[Walker Carlson]

Bruno,

I think that we can't guarantee that the message will get
propagated perfectly in every case of, say network partitioning, though it
will work for many cases. So I would say it's best effort and I will
mention it in the kip.

As for when to use it I think we can discuss if this will be
sufficient when we come to it, as long as we document its capabilities.

I hope this answers your question,

Walker

On Tue, Sep 22, 2020 at 12:33 AM Bruno Cadonna  wrote:

> Walker,
>
> I am sorry, but I still have a comment on the KIP although you have
> already started voting.
>
> What happens when a consumer of the group skips the rebalancing that
> propagates the shutdown request? Do you give a guarantee that all Kafka
> Streams clients are shutdown or is it best effort? If it is best effort,
> I guess the proposed method might not be used in critical cases where
> stopping record consumption may prevent or limit damage. I am not saying
> that it must be a guarantee, but this question should be answered in the
> KIP, IMO.
>
> Best,
> Bruno
>
> On 22.09.20 01:14, Walker Carlson wrote:
> > The error code right now is the assignor error, 2 is coded for shutdown
> > but it could be expanded to encode the causes or for other errors that
> need
> > to be communicated. For example we can add error code 3 to close the
> thread
> > but leave the client in an error state if we choose to do so in the
> future.
> >
> > On Mon, Sep 21, 2020 at 3:43 PM Boyang Chen 
> > wrote:
> >
> >> Thanks for the KIP Walker.
> >>
> >> In the KIP we mentioned "In order to communicate the shutdown request
> from
> >> one client to the others we propose to update the SubcriptionInfoData to
> >> include a short field which will encode an error code.", is there a
> >> dedicated error code that we should define here, or it is case-by-case?
> >>
> >> On Mon, Sep 21, 2020 at 1:38 PM Walker Carlson 
> >> wrote:
> >>
> >>> I am changing the name to "Add method to Shutdown entire Streams
> >>> Application" since we are no longer using an Exception, it seems more
> >>> appropriate.
> >>>
> >>> Also it looks like the discussion is pretty much finished so I will be
> >>> calling it to a vote.
> >>>
> >>> thanks,
> >>> Walker
> >>>
> >>> On Sat, Sep 19, 2020 at 7:49 PM Guozhang Wang 
> >> wrote:
> >>>
>  Sounds good to me. I also feel that this call should be non-blocking
> >> but
> >>> I
>  guess I was confused from the discussion thread that the API is
> >> designed
> >>> in
>  a blocking fashion which contradicts with my perspective and hence I
> >>> asked
>  for clarification :)
> 
>  Guozhang
> 
> 
>  On Wed, Sep 16, 2020 at 12:32 PM Walker Carlson <
> wcarl...@confluent.io
> >>>
>  wrote:
> 
> > Hello Guozhang,
> >
> > As for the logging I plan on having three logs. First, the client log
>  that
> > it is requesting an application shutdown, second, the leader log
>  processId
> > of the invoker, third, then the StreamRebalanceListener it logs that
> >> it
>  is
> > closing because of an `stream.appShutdown`. Hopefully this will be
> >>> enough
> > to make the cause of the close clear.
> >
> > I see what you mean about the name being dependent on the behavior of
> >>> the
> > method so I will try to clarify.  This is how I currently envision
> >> the
>  call
> > working.
> >
> > It is not an option to directly initiate a shutdown through a
>  StreamThread
> > object from a KafkaStreams object because "KafkaConsumer is not safe
> >>> for
> > multi-threaded access". So how it works is that the method in
>  KafkaStreams
> > finds the first alive thread and sets a flag in the StreamThread. The
> > StreamThread will receive the flag in its runloop then set the error
> >>> code
> > and trigger a rebalance, afterwards it will stop processing. After
> >> the
> > KafkaStreams has set the flag it will return true and continue
> >> running.
>  If
> > there are no alive threads the shutdown will fail and return false.
> >
> > What do you think the blocking behavior should be? I think that the
> > StreamThread should definitely stop to prevent any of the corruption
> >> we
>  are
> > trying to avoid by shutting down, but I don't see any advantage of
> >> the
> > KafkaStreams call blocking.
> >
> > You are correct to be concerned about the uncaught exception handler.
> >>> If
> > there are no live StreamThreads the rebalance will not be started at
> >>> all
> > and this would be a problem. However the user should be aware of this
> > because of the return of false and react appropriately. This would
> >> also
>  be
> > fixed if we implemented our own handler so we can rebalance before
> >> the
> > StreamThread closes.
> >
> > With that in mind I believe that `initiateClosingAllClients` would be
> >>> an
> > appropriate name. WDYT?
> >
> > 

Build failed in Jenkins: Kafka » kafka-trunk-jdk11 #83

[Apache Jenkins Server]

See 


Changes:

[github] MINOR: Remove unneeded FIXME (#9330)


--
[...truncated 3.35 MB...]
org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessRecordForTopic[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldForwardRecordsFromSubtopologyToSubtopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldForwardRecordsFromSubtopologyToSubtopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForSmallerValue[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForSmallerValue[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCreateStateDirectoryForStatefulTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCreateStateDirectoryForStatefulTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfWallClockTimeAdvances[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfWallClockTimeAdvances[Eos enabled = false] PASSED

org.apache.kafka.streams.test.TestRecordTest > testConsumerRecord STARTED

org.apache.kafka.streams.test.TestRecordTest > testConsumerRecord PASSED

org.apache.kafka.streams.test.TestRecordTest > testToString STARTED

org.apache.kafka.streams.test.TestRecordTest > testToString PASSED

org.apache.kafka.streams.test.TestRecordTest > testInvalidRecords STARTED

org.apache.kafka.streams.test.TestRecordTest > testInvalidRecords PASSED

org.apache.kafka.streams.test.TestRecordTest > testPartialConstructorEquals 
STARTED

org.apache.kafka.streams.test.TestRecordTest > testPartialConstructorEquals 
PASSED

org.apache.kafka.streams.test.TestRecordTest > testMultiFieldMatcher STARTED

org.apache.kafka.streams.test.TestRecordTest > testMultiFieldMatcher PASSED

org.apache.kafka.streams.test.TestRecordTest > testFields STARTED

org.apache.kafka.streams.test.TestRecordTest > testFields PASSED

org.apache.kafka.streams.test.TestRecordTest > testProducerRecord STARTED

org.apache.kafka.streams.test.TestRecordTest > testProducerRecord PASSED

org.apache.kafka.streams.test.TestRecordTest > testEqualsAndHashCode STARTED

org.apache.kafka.streams.test.TestRecordTest > testEqualsAndHashCode PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithInMemoryStore STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithInMemoryStore PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithLogging STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithLogging PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithCaching STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithCaching PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithPersistentStore STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithPersistentStore PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnIsOpen 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnIsOpen 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldDeleteAndReturnPlainValue STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldDeleteAndReturnPlainValue PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnName 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnName 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutWithUnknownTimestamp PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutAllWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutAllWithUnknownTimestamp PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldReturnIsPersistent STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldReturnIsPersistent PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutIfAbsentWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutIfAbsentWithUnknownTimestamp PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldForwardClose 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldForwardClose 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > s

[jira] [Created] (KAFKA-10518) Consumer fetches could be inefficient when lags are unbalanced

[Dhruvil Shah (Jira)]

Dhruvil Shah created KAFKA-10518:


 Summary: Consumer fetches could be inefficient when lags are 
unbalanced
 Key: KAFKA-10518
 URL: https://issues.apache.org/jira/browse/KAFKA-10518
 Project: Kafka
  Issue Type: Bug
Reporter: Dhruvil Shah


Consumer fetches are inefficient when lags are imbalanced across partitions, 
due to head of the line blocking and the behavior of blocking for `max.wait.ms` 
until data is available.

When the consumer receives a fetch response, it prepares the next fetch request 
and sends it out. The caveat is that the subsequent fetch request would 
explicitly exclude partitions for which the consumer received data in the 
previous round. This is to allow the consumer application to drain the data for 
those partitions, until the consumer fetches the other partitions it is 
subscribed to.

This behavior does not play out too well if the consumer is consuming when the 
lag is unbalanced, because it would receive data for the partitions it is 
lagging on, and then it would send a fetch request for partitions that do not 
have any data (or have little data). The latter will end up blocking for 
fetch.max.wait.ms on the broker before an empty response is sent back. This 
slows down the consumer’s overall consumption throughput since 
fetch.max.wait.ms is 500ms by default.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

[jira] [Created] (KAFKA-10517) Inefficient consumer processing with fetch sessions

[Dhruvil Shah (Jira)]

Dhruvil Shah created KAFKA-10517:


 Summary: Inefficient consumer processing with fetch sessions
 Key: KAFKA-10517
 URL: https://issues.apache.org/jira/browse/KAFKA-10517
 Project: Kafka
  Issue Type: Bug
Reporter: Dhruvil Shah


With the introduction of fetch sessions, the consumer and the broker share a 
unified view of the partitions being consumed and their current state 
(fetch_offset, last_propagated_hwm, last_propagated_start_offset, etc.). The 
consumer is still expected to consume in a round robin manner, however, we have 
observed certain cases where this is not the case.

Because of how we perform memory management on the consumer and implement fetch 
pipelining, we exclude partitions from a FetchRequest when they have not been 
drained by the application. This is done by adding these partitions to the 
`toForget` list in the `FetchRequest`. When partitions are added to the 
`toForget` list, the broker removes these partitions from its session cache. 
This causes bit of a divergence between the broker's and the client's view of 
the metadata.

When forgotten partitions are added back to the Fetch after the application 
have drained them, the server will immediately add them back to the session 
cache and return a response for them, even if there is no corresponding data. 
This re-triggers the behavior on the consumer to put this partition on the 
`toForget` list incorrectly, even though no data for the partition may have 
been returned.

We have seen this behavior to cause an imbalance in lags across partitions as 
the consumer no longer obeys the round-robin sequence given that the partitions 
keep shuffling between the `toForget` and `toSend` lists.

At a high level, this is caused due to the out of sync session caches on the 
consumer and broker. This ends up in a state where the partition balance is 
being maintained by external factors (such as whether metadata was returned for 
a partition), rather than following the round-robin ordering.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Build failed in Jenkins: Kafka » kafka-trunk-jdk11 #82

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-10514: Advance mock time for state directory cleanup (#9323)


--
[...truncated 6.69 MB...]
org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureInternalTopicNamesIfWrittenInto[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCaptureInternalTopicNamesIfWrittenInto[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnWallClockTimeDeprecated[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateOnWallClockTimeDeprecated[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessRecordForTopic[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldProcessRecordForTopic[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldForwardRecordsFromSubtopologyToSubtopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldForwardRecordsFromSubtopologyToSubtopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForSmallerValue[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldNotUpdateStoreForSmallerValue[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCreateStateDirectoryForStatefulTopology[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldCreateStateDirectoryForStatefulTopology[Eos enabled = false] PASSED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfWallClockTimeAdvances[Eos enabled = false] STARTED

org.apache.kafka.streams.TopologyTestDriverTest > 
shouldPunctuateIfWallClockTimeAdvances[Eos enabled = false] PASSED

org.apache.kafka.streams.test.TestRecordTest > testConsumerRecord STARTED

org.apache.kafka.streams.test.TestRecordTest > testConsumerRecord PASSED

org.apache.kafka.streams.test.TestRecordTest > testToString STARTED

org.apache.kafka.streams.test.TestRecordTest > testToString PASSED

org.apache.kafka.streams.test.TestRecordTest > testInvalidRecords STARTED

org.apache.kafka.streams.test.TestRecordTest > testInvalidRecords PASSED

org.apache.kafka.streams.test.TestRecordTest > testPartialConstructorEquals 
STARTED

org.apache.kafka.streams.test.TestRecordTest > testPartialConstructorEquals 
PASSED

org.apache.kafka.streams.test.TestRecordTest > testMultiFieldMatcher STARTED

org.apache.kafka.streams.test.TestRecordTest > testMultiFieldMatcher PASSED

org.apache.kafka.streams.test.TestRecordTest > testFields STARTED

org.apache.kafka.streams.test.TestRecordTest > testFields PASSED

org.apache.kafka.streams.test.TestRecordTest > testProducerRecord STARTED

org.apache.kafka.streams.test.TestRecordTest > testProducerRecord PASSED

org.apache.kafka.streams.test.TestRecordTest > testEqualsAndHashCode STARTED

org.apache.kafka.streams.test.TestRecordTest > testEqualsAndHashCode PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithInMemoryStore STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithInMemoryStore PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithLogging STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithLogging PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithCaching STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldFailWithCaching PASSED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithPersistentStore STARTED

org.apache.kafka.streams.test.wordcount.WindowedWordCountProcessorTest > 
shouldWorkWithPersistentStore PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnIsOpen 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnIsOpen 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldDeleteAndReturnPlainValue STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldDeleteAndReturnPlainValue PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnName 
STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > shouldReturnName 
PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutWithUnknownTimestamp PASSED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutAllWithUnknownTimestamp STARTED

org.apache.kafka.streams.internals.KeyValueStoreFacadeTest > 
shouldPutAllWithUnknownTimestamp PASSED

org.apache.kafka

Jenkins build is back to normal : Kafka » kafka-trunk-jdk15 #82

[Apache Jenkins Server]

See 

Re: ApacheCon Bug Bash

[Tom DuBuisson]

Sohil,

Is bugsbunnyshah/AIPlatform related to Apache Kafka in some manner?  I'm
not familiar with it.  Regardless, Muse is free to use on public
repositories - all you need to do is install the muse app (
https://github.com/apps/muse-dev).  Muse will then respond to pull requests
with new bugs or you can trigger an analysis of a particular branch via
console.muse.dev [1].

Regarding the Bug bash, this is an ASF specific conference and bash.  If
AIPlatform is hosted somewhere under the GitHub "apache" organization then
we can work on its inclusion.  If not then let's email off-list about what
you have in mind.

Cheers,
Tom

[1] I've actually forked the repository, added a configuration to specify
jdk11, and am trying this now.  https://github.com/TomMD/AIPlatform


On Tue, Sep 22, 2020 at 4:22 PM Sohil Shah 
wrote:

> Hi Tom,
>
> I would love to participate. You can scan the AIPlatform here
> https://github.com/bugsbunnyshah/AIPlatform
>
> Thanks
> Sohil
>
>
> On Tue, Sep 22, 2020 at 6:11 PM Tom DuBuisson  wrote:
>
> > Kafka Developers,
> >
> >
> >
> >
> >
> >
> >
> > As part of our sponsorship of ApacheCon, our company MuseDev is doing a
> Bug
> >
> > Bash for select Apache projects. We'll bring members of the ApacheCon
> >
> > community together to find and fix a range of security and performance
> bugs
> >
> > during the conference, and gameify the experience with teams, a
> >
> > leaderboard, and prizes. The bash is open to everyone whether attending
> the
> >
> > conference or not, and our whole dev team will also be participating to
> >
> > help fix as many bugs as we can.
> >
> >
> >
> >
> >
> >
> >
> > We're seeding the bug list with results from Muse, our code analysis
> >
> > platform, which runs as a Github App and comments on possible bugs as
> part
> >
> > of the pull request workflow.  Here's an example of what it looks like:
> >
> >
> >
> > https://github.com/curl/curl/pull/5971#discussion_r490252196
> >
> > 
> >
> >
> >
> >
> >
> >
> >
> > We explored a number of Apache projects and are reaching out because our
> >
> > analysis through Muse found some interesting bugs that could be fixed
> >
> > during the Bash.
> >
> >
> >
> >
> >
> >
> >
> > We're writing to see if you'd be interested in having your project
> included
> >
> > in the Bash. Everything is set up on our end, and if you're interested,
> we
> >
> > would need you to say yes on this listserv, and we’ll work with the
> Apache
> >
> > Infrastructure team to grant Muse access to your Github mirror. We'll
> then
> >
> > make sure it's all set-up and ready for the Bash. And of course, everyone
> >
> > on the project is most welcome to join the Bash and help us smash some
> > bugs.
> >
> >
> >
> >
> >
> > -Tom
> >
> >
>

Re: [DISCUSS] KIP-516: Topic Identifiers

[Jun Rao]

Hi, Justine,

Thanks for the response. Made another pass. A few more comments below.

20. znode schema:
20.1 It seems that {"name": "version", "type": "int", "id": "UUID", "doc":
"version id"} should be {"name": "version", "type": "int"}, {"name": "id",
"type": "UUID", "doc": "version id"}.
20.2 The znode format is JSON which doesn't have UUID type. So the type
probably should be string?
20.3 Also, the existing format used seems outdated. It should have the
following format.
Json.encodeAsBytes(Map(
  "version" -> 2,
  "partitions" -> replicaAssignmentJson.asJava,
  "adding_replicas" -> addingReplicasAssignmentJson.asJava,
  "removing_replicas" -> removingReplicasAssignmentJson.asJava
).asJava)
  }

21. Migration: The KIP says "The migration process can take place without
an inter-broker protocol bump, as the format stored in
/brokers/topics/[topic] will be compatible with older broker versions."
However, since we are bumping up the version of inter-broker requests, it
seems that we need to use IBP for migration.

22. The KIP says "Remove deleted topics from replicas by sending
StopReplicaRequest V3 for any topics which do not contain a topic ID, and
V4 for any topics which do contain a topic ID.". However, if we use IBP, it
seems that the controller will either send StopReplicaRequest V3
or StopReplicaRequest V4, but never mixed V3 and V4 for different topics.
Basically, before the IBP bump, V3 will be used. After the IBP bump,
topicId will be created and V4 will be used.

23. Given that we depend on IBP, do we still need "0 UNSPECIFIED"
in LeaderAndIsr?

24. LeaderAndIsrResponse v5 : It still has the topic field.

25. LeaderAndIsrResponse v5, StopReplicaResponse v4: Could we use this
opportunity to organize the response in 2 levels, first by topic, then by
partition, as most other requests/responses?

26. FetchRequest v13 : Should forgotten_topics_data use topicId too?

27. "This file can either be plain text (key/value pairs) or JSON." Have we
decided which one to use? Also, it would be helpful to provide an example.

28. Future improvement: Another future improvement opportunity is to use
topicId in GroupMetadataManager.offsetCommitKey in the offset_commit topic.
This may save some space.

Thanks,

Jun

On Wed, Sep 23, 2020 at 8:50 AM Justine Olshan  wrote:

> Hi Tom,
>
> Thanks for the comment. I think this is a really good idea and it has been
> added to the KIP under the newly added tooling section.
>
> Thanks again,
> Justine
>
> On Wed, Sep 23, 2020 at 3:17 AM Tom Bentley  wrote:
>
> > Hi Justine,
> >
> > I know you started the vote thread, but on re-reading the KIP I noticed
> > that although the topic id is included in the MetadataResponse it's not
> > surfaced in the output from `kafka-topics.sh --describe`. Maybe that was
> > intentional because ids are intentionally not really something the user
> > should care deeply about, but it would also make life harder for anyone
> > debugging Kafka and this would likely get worse the more topic ids got
> > rolled out across the protocols, clients etc. It seems likely that
> > `kafka-topics.sh` will eventually need the ability to show the id of a
> > topic and perhaps find a topic name given an id. Is there any reason not
> to
> > implement that in this KIP?
> >
> > Many thanks,
> >
> > Tom
> >
> > On Mon, Sep 21, 2020 at 9:54 PM Justine Olshan 
> > wrote:
> >
> > > Hi all,
> > >
> > > After thinking about it, I've decided to remove the topic name from the
> > > Fetch Request and Response after all. Since there are so many of these
> > > requests per second, it is worth removing the extra information. I've
> > > updated the KIP to reflect this change.
> > >
> > > Please let me know if there is anything else we should discuss before
> > > voting.
> > >
> > > Thank you,
> > > Justine
> > >
> > > On Fri, Sep 18, 2020 at 9:46 AM Justine Olshan 
> > > wrote:
> > >
> > > > Hi Jun,
> > > >
> > > > I see what you are saying. For now we can remove the extra
> information.
> > > > I'll leave the option to add more fields to the file in the future.
> The
> > > KIP
> > > > has been updated to reflect this change.
> > > >
> > > > Thanks,
> > > > Justine
> > > >
> > > > On Fri, Sep 18, 2020 at 8:46 AM Jun Rao  wrote:
> > > >
> > > >> Hi, Justine,
> > > >>
> > > >> Thanks for the reply.
> > > >>
> > > >> 13. If the log directory is the source of truth, it means that the
> > > >> redundant info in the metadata file will be ignored. Then the
> question
> > > is
> > > >> why do we need to put the redundant info in the metadata file now?
> > > >>
> > > >> Thanks,
> > > >>
> > > >> Jun
> > > >>
> > > >> On Thu, Sep 17, 2020 at 5:07 PM Justine Olshan <
> jols...@confluent.io>
> > > >> wrote:
> > > >>
> > > >> > Hi Jun,
> > > >> > Thanks for the quick response!
> > > >> >
> > > >> > 12. I've decided to bump up the versions on the requests and
> updated
> > > the
> > > >> > KIP. I think it's good we thoroughly discussed the options here,
> so
> > we
> 

Kafka consumer Exception handling and offset management

[Preethkumar Thirupathi]

Hi Team,

Currently I'm implementing Spring boot @kafkalistener. I need to know that
what all are the exception that may raise, while try to consume the record
from kafka topic. How to commit the current offset after successfully
processed the record.In between that I have handle the exception.
Kindly advise the scenarios, what type of exception we can implement snd it
would be better to have an architectural diagram.

Thanks,
PREETHKUMAR T

Re: Question Regarding Offset Behavior When Calling Poll()

[Matthias J. Sax]

I guess it depends where the exception comes from? Can you clarify?

-Matthias

On 9/23/20 12:53 AM, Zhen Zhang wrote:
> Hi there,
> 
> I am new to Kafka and I would like to get some clarifications for a newbie
> question,
> 
> Let's say if I have set up my consumer's "enable.auto.commit" to false, and
> then poll the records one at a time. So when calling poll(), starting from
> offset 0, if any exception is thrown, should I expect to get the record at
> offset 0 or offset 1 when I call poll() again? The reason I'm asking for
> this is bc in the Kafka Doc, it says that,
> "The position of the consumer gives the offset of the next record that will
> be given out. It will be one larger than the highest offset the consumer
> has seen in that partition. It automatically advances every time the
> consumer receives messages in a call to poll(Duration)."
> 
> But in my described situation above, an exception is thrown, I'm not sure
> if this is counted as a successful poll (meaning that the next poll() will
> give the next record) or a failed one (meaning that the next poll() will
> give the same record again).
> 
> I would really appreciate it for your help.
> 
> Thanks,
> Zhen Zhang
> Software Engineer
> [image: Twilio] 
> MOBILE (949) 771-6073
> EMAIL zzh...@twilio.com
> 

Jenkins build is back to normal : Kafka » kafka-trunk-jdk8 #80

[Apache Jenkins Server]

See 

Re: [DISCUSS] KIP-516: Topic Identifiers

[Justine Olshan]

Hi Tom,

Thanks for the comment. I think this is a really good idea and it has been
added to the KIP under the newly added tooling section.

Thanks again,
Justine

On Wed, Sep 23, 2020 at 3:17 AM Tom Bentley  wrote:

> Hi Justine,
>
> I know you started the vote thread, but on re-reading the KIP I noticed
> that although the topic id is included in the MetadataResponse it's not
> surfaced in the output from `kafka-topics.sh --describe`. Maybe that was
> intentional because ids are intentionally not really something the user
> should care deeply about, but it would also make life harder for anyone
> debugging Kafka and this would likely get worse the more topic ids got
> rolled out across the protocols, clients etc. It seems likely that
> `kafka-topics.sh` will eventually need the ability to show the id of a
> topic and perhaps find a topic name given an id. Is there any reason not to
> implement that in this KIP?
>
> Many thanks,
>
> Tom
>
> On Mon, Sep 21, 2020 at 9:54 PM Justine Olshan 
> wrote:
>
> > Hi all,
> >
> > After thinking about it, I've decided to remove the topic name from the
> > Fetch Request and Response after all. Since there are so many of these
> > requests per second, it is worth removing the extra information. I've
> > updated the KIP to reflect this change.
> >
> > Please let me know if there is anything else we should discuss before
> > voting.
> >
> > Thank you,
> > Justine
> >
> > On Fri, Sep 18, 2020 at 9:46 AM Justine Olshan 
> > wrote:
> >
> > > Hi Jun,
> > >
> > > I see what you are saying. For now we can remove the extra information.
> > > I'll leave the option to add more fields to the file in the future. The
> > KIP
> > > has been updated to reflect this change.
> > >
> > > Thanks,
> > > Justine
> > >
> > > On Fri, Sep 18, 2020 at 8:46 AM Jun Rao  wrote:
> > >
> > >> Hi, Justine,
> > >>
> > >> Thanks for the reply.
> > >>
> > >> 13. If the log directory is the source of truth, it means that the
> > >> redundant info in the metadata file will be ignored. Then the question
> > is
> > >> why do we need to put the redundant info in the metadata file now?
> > >>
> > >> Thanks,
> > >>
> > >> Jun
> > >>
> > >> On Thu, Sep 17, 2020 at 5:07 PM Justine Olshan 
> > >> wrote:
> > >>
> > >> > Hi Jun,
> > >> > Thanks for the quick response!
> > >> >
> > >> > 12. I've decided to bump up the versions on the requests and updated
> > the
> > >> > KIP. I think it's good we thoroughly discussed the options here, so
> we
> > >> know
> > >> > we made a good choice. :)
> > >> >
> > >> > 13. This is an interesting situation. I think if this does occur we
> > >> should
> > >> > give a warning. I agree that it's hard to know the source of truth
> for
> > >> sure
> > >> > since the directory or the file could be manually modified. I guess
> > the
> > >> > directory could be used as the source of truth. To be honest, I'm
> not
> > >> > really sure what happens in kafka when the log directory is renamed
> > >> > manually in such a way. I'm also wondering if the situation is
> > >> recoverable
> > >> > in this scenario.
> > >> >
> > >> > Thanks,
> > >> > Justine
> > >> >
> > >> > On Thu, Sep 17, 2020 at 4:28 PM Jun Rao  wrote:
> > >> >
> > >> > > Hi, Justine,
> > >> > >
> > >> > > Thanks for the reply.
> > >> > >
> > >> > > 12. I don't have a strong preference either. However, if we need
> IBP
> > >> > > anyway, maybe it's easier to just bump up the version for all
> inter
> > >> > broker
> > >> > > requests and add the topic id field as a regular field. A regular
> > >> field
> > >> > is
> > >> > > a bit more concise in wire transfer than a flexible field.
> > >> > >
> > >> > > 13. The confusion that I was referring to is between the topic
> name
> > >> and
> > >> > > partition number between the log dir and the metadata file. For
> > >> example,
> > >> > if
> > >> > > the log dir is topicA-1 and the metadata file in it has topicB and
> > >> > > partition 0 (say due to a bug or manual modification), which one
> do
> > we
> > >> > use
> > >> > > as the source of truth?
> > >> > >
> > >> > > Jun
> > >> > >
> > >> > > On Thu, Sep 17, 2020 at 3:43 PM Justine Olshan <
> > jols...@confluent.io>
> > >> > > wrote:
> > >> > >
> > >> > > > Hi Jun,
> > >> > > > Thanks for the comments.
> > >> > > >
> > >> > > > 12. I bumped the LeaderAndIsrRequest because I removed the topic
> > >> name
> > >> > > field
> > >> > > > in the response. It may be possible to avoid bumping the version
> > >> > without
> > >> > > > that change, but I may be missing something.
> > >> > > > I believe StopReplica is actually on version 3 now, but because
> > >> > version 2
> > >> > > > is flexible, I kept that listed as version 2 on the KIP page.
> > >> However,
> > >> > > you
> > >> > > > may be right in that we may need to bump the version on
> > StopReplica
> > >> to
> > >> > > deal
> > >> > > > with deletion differently as mentioned above. I don't know if I
> > >> have a
> > >> > > big
> > >> > > > preference over used tagged fields or

Build failed in Jenkins: Kafka » kafka-trunk-jdk15 #81

[Apache Jenkins Server]

See 


Changes:

[github] KAFKA-10482: Fix flaky testDynamicListenerConnectionCreationRateQuota 
(#9301)


--
[...truncated 3.34 MB...]

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

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

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

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

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

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

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

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

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

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

Re: [DISCUSS] KIP-653: Upgrade log4j to log4j2

[Dongjin Lee]

Hi Tom,

Thanks for the detailed analysis. Recently, I was also thinking about API
compatibility. I initially thought that the difference between the root
logger name would break the compatibility (as the KIP states), it seems
like I found a workaround:

1. When the user requests arrive, regard the logger name 'root' as an empty
string. (i.e., translate the request into the log4j2 equivalent.)
2. When generating the response, change the logger name '' into 'root'.
(i.e., translate the response into the log4j equivalent.)
3. Remove (or make reverse) the workaround above when we make log4j2
default.

In short, it seems like we can handle the API incompatibility introduced by
the root logger name change by adding a facade.

How do you think?

Thanks,
Dongjin

On Wed, Sep 23, 2020 at 7:36 PM Tom Bentley  wrote:

> Hi Dongjin,
>
> I'd like to see this feature, but if I understand correctly, the KIP in its
> current form breaks a couple of Kafka APIs. For Kafka Connect it says "From
> log4j2, the name of the root logger becomes empty string from 'root'. It
> impacts Kafka connect's dynamic logging control feature. (And should be
> documented.)". This seems to imply that any tooling that a user might have
> written about logging in Kafka Connect will break because the client and
> server don't have a shared understanding of how to identify the root
> logger. The same would be true for the DescribeConfigs, AlterConfigs and
> IncrementalAlterConfigs protocols using the BROKER_LOGGER resource type, I
> think.
>
> Maybe that's OK if the behaviour was changing in a new major release of
> Kafka (e.g. 3.0), but I don't think it's allowed in Kafka 2.7 given the
> project's compatibility requirements & semantic versioning.
>
> If these API compatibility issues are easily fixed I think it would be
> great to have this in 2.7, but if not it might be easier to target this for
> Kafka 3.0. That would also allow you to change the logging config format as
> suggested by Ismael.
>
> Many thanks,
>
> Tom
>
> On Tue, Sep 22, 2020 at 5:15 PM Dongjin Lee  wrote:
>
> > Hi devs,
> >
> > I updated the KIP with the migration plan I discussed with Ismael.
> >
> > I think 2.7.0 is the perfect time for starting migration into log4j2. If
> we
> > miss this opportunity, the migration would be much harder. So please
> have a
> > look at this proposal.
> >
> > I also opened a voting thread for this.
> >
> > Thanks,
> > Dongjin
> >
> > On Thu, Sep 17, 2020 at 2:29 AM Dongjin Lee  wrote:
> >
> > > Hi Ismael,
> > >
> > > > Have we considered switching to the log4j2 logging config format by
> > > default and providing a mechanism to use the old format?
> > >
> > > As of present, the proposal leaves the default config format switching
> to
> > > sometime in the future. However, I think it is not a difficult task and
> > is
> > > up to the community's decision. The draft implementation already
> includes
> > > log4j2 counterparts for all existing 1.x format (template) configs.
> > > Although it still uses traditional log4j format as a default for
> backward
> > > compatibility, the users who prefer the log4j2 configs can use it by
> > > setting `export
> > >
> KAFKA_LOG4J_OPTS="-Dlog4j.configurationFile={log4j2-config-file-path}"`.
> > > Whenever we change the default logging format, we must don't forget to
> > > switch this functionality to the reverse, i.e., making log4j 1.x format
> > > available as an opt-in.
> > >
> > > I am so concerned about the community's opinion when would be adequate
> to
> > > make the log4j2 config as default.
> > >
> > > Thanks,
> > > Dongjin
> > >
> > > +1. As a note, I have an almost-completed implementation of log4j2
> > > equivalent for the log4j-appender. I think it would be great if this
> > > feature can be provided with changing the default logging config
> format.
> > >
> > > On Wed, Sep 16, 2020 at 11:49 PM Ismael Juma 
> wrote:
> > >
> > >> Thanks for the KIP, Dongjin. Have we considered switching to the
> log4j2
> > >> logging config format by default and providing a mechanism to use the
> > old
> > >> format? It is likely that we will release 3.0 as the release after
> 2.7,
> > so
> > >> it would provide a good opportunity to move on from the legacy config
> > >> format. The other option is to stick with the old format for 3.0 and
> > >> migrate to the new format in 4.0.
> > >>
> > >> Ismael
> > >>
> > >> On Wed, Aug 5, 2020 at 7:45 AM Dongjin Lee 
> wrote:
> > >>
> > >> > Hi, Kafka dev,
> > >> >
> > >> > I hope to initiate the discussion of KIP-653, upgrading log4j to
> > log4j2.
> > >> >
> > >> >
> > >> >
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-653%3A+Upgrade+log4j+to+log4j2
> > >> >
> > >> > All kinds of feedbacks are greatly appreciated!
> > >> >
> > >> > Best,
> > >> > Dongjin
> > >> >
> > >> > --
> > >> > *Dongjin Lee*
> > >> >
> > >> > *A hitchhiker in the mathematical world.*
> > >> >
> > >> >
> > >> >
> > >> >
> > >> > *github:  github.com/dongjin

[jira] [Created] (KAFKA-10516) Implement Topic Command changes

[David Jacot (Jira)]

David Jacot created KAFKA-10516:
---

 Summary: Implement Topic Command changes
 Key: KAFKA-10516
 URL: https://issues.apache.org/jira/browse/KAFKA-10516
 Project: Kafka
  Issue Type: Sub-task
Reporter: David Jacot
Assignee: David Jacot






--
This message was sent by Atlassian Jira
(v8.3.4#803005)

[GitHub] [kafka-site] ankit-kumar-25 edited a comment on pull request #220: KAFKA-8360: Docs do not mention RequestQueueSize JMX metric

[GitBox]

ankit-kumar-25 edited a comment on pull request #220:
URL: https://github.com/apache/kafka-site/pull/220#issuecomment-696188603


   Hey @viktorsomogyi, 
   
   Thank you for the pointers, I have created a PR against the ops.html 
available in the Kafka project: https://github.com/apache/kafka/pull/9325



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [kafka-site] ankit-kumar-25 closed pull request #220: KAFKA-8360: Docs do not mention RequestQueueSize JMX metric

[GitBox]

ankit-kumar-25 closed pull request #220:
URL: https://github.com/apache/kafka-site/pull/220


   



This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [DISCUSS] KIP-653: Upgrade log4j to log4j2

[Tom Bentley]

Hi Dongjin,

I'd like to see this feature, but if I understand correctly, the KIP in its
current form breaks a couple of Kafka APIs. For Kafka Connect it says "From
log4j2, the name of the root logger becomes empty string from 'root'. It
impacts Kafka connect's dynamic logging control feature. (And should be
documented.)". This seems to imply that any tooling that a user might have
written about logging in Kafka Connect will break because the client and
server don't have a shared understanding of how to identify the root
logger. The same would be true for the DescribeConfigs, AlterConfigs and
IncrementalAlterConfigs protocols using the BROKER_LOGGER resource type, I
think.

Maybe that's OK if the behaviour was changing in a new major release of
Kafka (e.g. 3.0), but I don't think it's allowed in Kafka 2.7 given the
project's compatibility requirements & semantic versioning.

If these API compatibility issues are easily fixed I think it would be
great to have this in 2.7, but if not it might be easier to target this for
Kafka 3.0. That would also allow you to change the logging config format as
suggested by Ismael.

Many thanks,

Tom

On Tue, Sep 22, 2020 at 5:15 PM Dongjin Lee  wrote:

> Hi devs,
>
> I updated the KIP with the migration plan I discussed with Ismael.
>
> I think 2.7.0 is the perfect time for starting migration into log4j2. If we
> miss this opportunity, the migration would be much harder. So please have a
> look at this proposal.
>
> I also opened a voting thread for this.
>
> Thanks,
> Dongjin
>
> On Thu, Sep 17, 2020 at 2:29 AM Dongjin Lee  wrote:
>
> > Hi Ismael,
> >
> > > Have we considered switching to the log4j2 logging config format by
> > default and providing a mechanism to use the old format?
> >
> > As of present, the proposal leaves the default config format switching to
> > sometime in the future. However, I think it is not a difficult task and
> is
> > up to the community's decision. The draft implementation already includes
> > log4j2 counterparts for all existing 1.x format (template) configs.
> > Although it still uses traditional log4j format as a default for backward
> > compatibility, the users who prefer the log4j2 configs can use it by
> > setting `export
> > KAFKA_LOG4J_OPTS="-Dlog4j.configurationFile={log4j2-config-file-path}"`.
> > Whenever we change the default logging format, we must don't forget to
> > switch this functionality to the reverse, i.e., making log4j 1.x format
> > available as an opt-in.
> >
> > I am so concerned about the community's opinion when would be adequate to
> > make the log4j2 config as default.
> >
> > Thanks,
> > Dongjin
> >
> > +1. As a note, I have an almost-completed implementation of log4j2
> > equivalent for the log4j-appender. I think it would be great if this
> > feature can be provided with changing the default logging config format.
> >
> > On Wed, Sep 16, 2020 at 11:49 PM Ismael Juma  wrote:
> >
> >> Thanks for the KIP, Dongjin. Have we considered switching to the log4j2
> >> logging config format by default and providing a mechanism to use the
> old
> >> format? It is likely that we will release 3.0 as the release after 2.7,
> so
> >> it would provide a good opportunity to move on from the legacy config
> >> format. The other option is to stick with the old format for 3.0 and
> >> migrate to the new format in 4.0.
> >>
> >> Ismael
> >>
> >> On Wed, Aug 5, 2020 at 7:45 AM Dongjin Lee  wrote:
> >>
> >> > Hi, Kafka dev,
> >> >
> >> > I hope to initiate the discussion of KIP-653, upgrading log4j to
> log4j2.
> >> >
> >> >
> >> >
> >>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-653%3A+Upgrade+log4j+to+log4j2
> >> >
> >> > All kinds of feedbacks are greatly appreciated!
> >> >
> >> > Best,
> >> > Dongjin
> >> >
> >> > --
> >> > *Dongjin Lee*
> >> >
> >> > *A hitchhiker in the mathematical world.*
> >> >
> >> >
> >> >
> >> >
> >> > *github:  github.com/dongjinleekr
> >> > keybase:
> >> https://keybase.io/dongjinleekr
> >> > linkedin:
> >> kr.linkedin.com/in/dongjinleekr
> >> > speakerdeck:
> >> > speakerdeck.com/dongjin
> >> > *
> >> >
> >>
> >
> >
> > --
> > *Dongjin Lee*
> >
> > *A hitchhiker in the mathematical world.*
> >
> >
> >
> >
> > *github:  github.com/dongjinleekr
> > keybase:
> https://keybase.io/dongjinleekr
> > linkedin:
> kr.linkedin.com/in/dongjinleekr
> > speakerdeck:
> speakerdeck.com/dongjin
> > *
> >
>
>
> --
> *Dongjin Lee*
>
> *A hitchhiker in the mathematical world.*
>
>
>
>
> *github:  github.com/dongjinleekr
> keybase: https://keybase.io/dongjinleekr
> linkedin: kr.linkedin.com/in/dongji

Re: [DISCUSS] KIP-516: Topic Identifiers

[Tom Bentley]

Hi Justine,

I know you started the vote thread, but on re-reading the KIP I noticed
that although the topic id is included in the MetadataResponse it's not
surfaced in the output from `kafka-topics.sh --describe`. Maybe that was
intentional because ids are intentionally not really something the user
should care deeply about, but it would also make life harder for anyone
debugging Kafka and this would likely get worse the more topic ids got
rolled out across the protocols, clients etc. It seems likely that
`kafka-topics.sh` will eventually need the ability to show the id of a
topic and perhaps find a topic name given an id. Is there any reason not to
implement that in this KIP?

Many thanks,

Tom

On Mon, Sep 21, 2020 at 9:54 PM Justine Olshan  wrote:

> Hi all,
>
> After thinking about it, I've decided to remove the topic name from the
> Fetch Request and Response after all. Since there are so many of these
> requests per second, it is worth removing the extra information. I've
> updated the KIP to reflect this change.
>
> Please let me know if there is anything else we should discuss before
> voting.
>
> Thank you,
> Justine
>
> On Fri, Sep 18, 2020 at 9:46 AM Justine Olshan 
> wrote:
>
> > Hi Jun,
> >
> > I see what you are saying. For now we can remove the extra information.
> > I'll leave the option to add more fields to the file in the future. The
> KIP
> > has been updated to reflect this change.
> >
> > Thanks,
> > Justine
> >
> > On Fri, Sep 18, 2020 at 8:46 AM Jun Rao  wrote:
> >
> >> Hi, Justine,
> >>
> >> Thanks for the reply.
> >>
> >> 13. If the log directory is the source of truth, it means that the
> >> redundant info in the metadata file will be ignored. Then the question
> is
> >> why do we need to put the redundant info in the metadata file now?
> >>
> >> Thanks,
> >>
> >> Jun
> >>
> >> On Thu, Sep 17, 2020 at 5:07 PM Justine Olshan 
> >> wrote:
> >>
> >> > Hi Jun,
> >> > Thanks for the quick response!
> >> >
> >> > 12. I've decided to bump up the versions on the requests and updated
> the
> >> > KIP. I think it's good we thoroughly discussed the options here, so we
> >> know
> >> > we made a good choice. :)
> >> >
> >> > 13. This is an interesting situation. I think if this does occur we
> >> should
> >> > give a warning. I agree that it's hard to know the source of truth for
> >> sure
> >> > since the directory or the file could be manually modified. I guess
> the
> >> > directory could be used as the source of truth. To be honest, I'm not
> >> > really sure what happens in kafka when the log directory is renamed
> >> > manually in such a way. I'm also wondering if the situation is
> >> recoverable
> >> > in this scenario.
> >> >
> >> > Thanks,
> >> > Justine
> >> >
> >> > On Thu, Sep 17, 2020 at 4:28 PM Jun Rao  wrote:
> >> >
> >> > > Hi, Justine,
> >> > >
> >> > > Thanks for the reply.
> >> > >
> >> > > 12. I don't have a strong preference either. However, if we need IBP
> >> > > anyway, maybe it's easier to just bump up the version for all inter
> >> > broker
> >> > > requests and add the topic id field as a regular field. A regular
> >> field
> >> > is
> >> > > a bit more concise in wire transfer than a flexible field.
> >> > >
> >> > > 13. The confusion that I was referring to is between the topic name
> >> and
> >> > > partition number between the log dir and the metadata file. For
> >> example,
> >> > if
> >> > > the log dir is topicA-1 and the metadata file in it has topicB and
> >> > > partition 0 (say due to a bug or manual modification), which one do
> we
> >> > use
> >> > > as the source of truth?
> >> > >
> >> > > Jun
> >> > >
> >> > > On Thu, Sep 17, 2020 at 3:43 PM Justine Olshan <
> jols...@confluent.io>
> >> > > wrote:
> >> > >
> >> > > > Hi Jun,
> >> > > > Thanks for the comments.
> >> > > >
> >> > > > 12. I bumped the LeaderAndIsrRequest because I removed the topic
> >> name
> >> > > field
> >> > > > in the response. It may be possible to avoid bumping the version
> >> > without
> >> > > > that change, but I may be missing something.
> >> > > > I believe StopReplica is actually on version 3 now, but because
> >> > version 2
> >> > > > is flexible, I kept that listed as version 2 on the KIP page.
> >> However,
> >> > > you
> >> > > > may be right in that we may need to bump the version on
> StopReplica
> >> to
> >> > > deal
> >> > > > with deletion differently as mentioned above. I don't know if I
> >> have a
> >> > > big
> >> > > > preference over used tagged fields or not.
> >> > > >
> >> > > > 13. I was thinking that in the case where the file and the request
> >> > topic
> >> > > > ids don't match, it means that the broker's topic/the one in the
> >> file
> >> > has
> >> > > > been deleted. In that case, we would need to delete the old topic
> >> and
> >> > > start
> >> > > > receiving the new version. If the topic name were to change, but
> the
> >> > ids
> >> > > > still matched, the file would also need to update. Am I missing a
> >> case
> >> > > > where the f

[jira] [Created] (KAFKA-10515) NPE: Foreign key join serde may not be initialized with default serde if application is distributed

[Thorsten Hake (Jira)]

Thorsten Hake created KAFKA-10515:
-

 Summary: NPE: Foreign key join serde may not be initialized with 
default serde if application is distributed
 Key: KAFKA-10515
 URL: https://issues.apache.org/jira/browse/KAFKA-10515
 Project: Kafka
  Issue Type: Bug
  Components: streams
Affects Versions: 2.5.1, 2.6.0
Reporter: Thorsten Hake


The fix of KAFKA-9517 fixed the initialization of the foreign key joins serdes 
for KStream applications that do not run distributed over multiple instances.

However, if an application runs distributed over multiple instances, the 
foreign key join serdes may still not be initialized leading to the following 
NPE:
{noformat}
Encountered the following error during 
processing:java.lang.NullPointerException: null
at 
org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapperSerde$SubscriptionWrapperSerializer.serialize(SubscriptionWrapperSerde.java:85)
at 
org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionWrapperSerde$SubscriptionWrapperSerializer.serialize(SubscriptionWrapperSerde.java:52)
at 
org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:59)
at 
org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:50)
at 
org.apache.kafka.streams.state.internals.ValueAndTimestampSerializer.serialize(ValueAndTimestampSerializer.java:27)
at 
org.apache.kafka.streams.state.StateSerdes.rawValue(StateSerdes.java:192)
at 
org.apache.kafka.streams.state.internals.MeteredKeyValueStore.lambda$put$3(MeteredKeyValueStore.java:144)
at 
org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
at 
org.apache.kafka.streams.state.internals.MeteredKeyValueStore.put(MeteredKeyValueStore.java:144)
at 
org.apache.kafka.streams.processor.internals.ProcessorContextImpl$KeyValueStoreReadWriteDecorator.put(ProcessorContextImpl.java:487)
at 
org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionStoreReceiveProcessorSupplier$1.process(SubscriptionStoreReceiveProcessorSupplier.java:102)
at 
org.apache.kafka.streams.kstream.internals.foreignkeyjoin.SubscriptionStoreReceiveProcessorSupplier$1.process(SubscriptionStoreReceiveProcessorSupplier.java:55)
at 
org.apache.kafka.streams.processor.internals.ProcessorNode.lambda$process$2(ProcessorNode.java:142)
at 
org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
at 
org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:142)
at 
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
at 
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)
at 
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:133)
at 
org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:104)
at 
org.apache.kafka.streams.processor.internals.StreamTask.lambda$process$3(StreamTask.java:383)
at 
org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806)
at 
org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:383)
at 
org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:475)
at 
org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:550)
at 
org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:802)
at 
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697)
at 
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670){noformat}

This happens because the processors for foreign key joins will be distributed 
across multiple tasks. The serde will only be initialized with the default 
serde during the initialization of the task containing the sink node 
("subscription-registration-sink"). So if the task containing the 
SubscriptionStoreReceiveProcessor ("subscription-receive") is not assigned to 
the same instance as the task containing the sink node, a NPE will be thrown 
because the Serde of the state store used within the 
SubscriptionStoreReceiveProcessor is not initialized.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Question Regarding Offset Behavior When Calling Poll()

[Zhen Zhang]

Hi there,

I am new to Kafka and I would like to get some clarifications for a newbie
question,

Let's say if I have set up my consumer's "enable.auto.commit" to false, and
then poll the records one at a time. So when calling poll(), starting from
offset 0, if any exception is thrown, should I expect to get the record at
offset 0 or offset 1 when I call poll() again? The reason I'm asking for
this is bc in the Kafka Doc, it says that,
"The position of the consumer gives the offset of the next record that will
be given out. It will be one larger than the highest offset the consumer
has seen in that partition. It automatically advances every time the
consumer receives messages in a call to poll(Duration)."

But in my described situation above, an exception is thrown, I'm not sure
if this is counted as a successful poll (meaning that the next poll() will
give the next record) or a failed one (meaning that the next poll() will
give the same record again).

I would really appreciate it for your help.

Thanks,
Zhen Zhang
Software Engineer
[image: Twilio] 
MOBILE (949) 771-6073
EMAIL zzh...@twilio.com