[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[
https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215
]
Kirk True edited comment on KAFKA-16792 at 6/3/24 9:09 PM:
---
The following don't work for reasons other than the timeout:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFSETS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions: this is expecting {{poll()}} to
call the {{LIST_OFFSETS}} RPC. The new consumer doesn't do this, but instead
sends {{FETCH_OFFSETS}}
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
was (Author: kirktrue):
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFSETS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions: this is expecting {{poll()}} to
call the {{LIST_OFFSETS}} RPC. The new consumer doesn't do this, but instead
sends {{FETCH_OFFSETS}}
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
> Enable consumer unit tests that fail to fetch offsets only for new consumer
> with poll(0)
>
>
> Key: KAFKA-16792
> URL: https://issues.apache.org/jira/browse/KAFKA-16792
> Project: Kafka
> Issue Type: Sub-task
> Components: clients, consumer
>Reporter: Lianet Magrans
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> Enable the following unit tests for the new async consumer in
> KafkaConsumerTest:
> - testFetchStableOffsetThrowInPoll
> - testCurrentLag
> - testListOffsetShouldUpdateSubscriptions
> - testResetToCommittedOffset
> - testResetUsingAutoResetPolicy
> - testPollReturnsRecords
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[
https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215
]
Kirk True edited comment on KAFKA-16792 at 6/3/24 9:03 PM:
---
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFSETS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions: this is expecting {{poll()}} to
call the {{LIST_OFFSETS}} RPC. The new consumer doesn't do this, but instead
sends {{FETCH_OFFSETS}}
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
was (Author: kirktrue):
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFSETS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
> Enable consumer unit tests that fail to fetch offsets only for new consumer
> with poll(0)
>
>
> Key: KAFKA-16792
> URL: https://issues.apache.org/jira/browse/KAFKA-16792
> Project: Kafka
> Issue Type: Sub-task
> Components: clients, consumer
>Reporter: Lianet Magrans
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> Enable the following unit tests for the new async consumer in
> KafkaConsumerTest:
> - testFetchStableOffsetThrowInPoll
> - testCurrentLag
> - testListOffsetShouldUpdateSubscriptions
> - testResetToCommittedOffset
> - testResetUsingAutoResetPolicy
> - testPollReturnsRecords
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[
https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215
]
Kirk True edited comment on KAFKA-16792 at 5/30/24 9:41 PM:
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFSETS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
was (Author: kirktrue):
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFESTS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
> Enable consumer unit tests that fail to fetch offsets only for new consumer
> with poll(0)
>
>
> Key: KAFKA-16792
> URL: https://issues.apache.org/jira/browse/KAFKA-16792
> Project: Kafka
> Issue Type: Sub-task
> Components: clients, consumer
>Reporter: Lianet Magrans
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> Enable the following unit tests for the new async consumer in
> KafkaConsumerTest:
> - testFetchStableOffsetThrowInPoll
> - testCurrentLag
> - testListOffsetShouldUpdateSubscriptions
> - testResetToCommittedOffset
> - testResetUsingAutoResetPolicy
> - testPollReturnsRecords
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[
https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215
]
Kirk True edited comment on KAFKA-16792 at 5/30/24 9:40 PM:
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag: this is expecting {{poll()}} to call the {{LIST_OFFSETS}}
RPC. The new consumer doesn't do this, but instead sends {{FETCH_OFFESTS}}
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
was (Author: kirktrue):
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
> Enable consumer unit tests that fail to fetch offsets only for new consumer
> with poll(0)
>
>
> Key: KAFKA-16792
> URL: https://issues.apache.org/jira/browse/KAFKA-16792
> Project: Kafka
> Issue Type: Sub-task
> Components: clients, consumer
>Reporter: Lianet Magrans
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> Enable the following unit tests for the new async consumer in
> KafkaConsumerTest:
> - testFetchStableOffsetThrowInPoll
> - testCurrentLag
> - testListOffsetShouldUpdateSubscriptions
> - testResetToCommittedOffset
> - testResetUsingAutoResetPolicy
> - testPollReturnsRecords
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[
https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215
]
Kirk True edited comment on KAFKA-16792 at 5/30/24 9:34 PM:
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords: uses the {{CLASSIC}} rebalance RPCs
- testResetToCommittedOffset
was (Author: kirktrue):
These tests work with KAFKA-16200:
- testResetUsingAutoResetPolicy
- testFetchStableOffsetThrowInCommitted (wasn't in the above list)
The following still don't work:
- testCurrentLag
- testFetchStableOffsetThrowInPoll
- testListOffsetShouldUpdateSubscriptions
- testPollReturnsRecords
- testResetToCommittedOffset
> Enable consumer unit tests that fail to fetch offsets only for new consumer
> with poll(0)
>
>
> Key: KAFKA-16792
> URL: https://issues.apache.org/jira/browse/KAFKA-16792
> Project: Kafka
> Issue Type: Sub-task
> Components: clients, consumer
>Reporter: Lianet Magrans
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> Enable the following unit tests for the new async consumer in
> KafkaConsumerTest:
> - testFetchStableOffsetThrowInPoll
> - testCurrentLag
> - testListOffsetShouldUpdateSubscriptions
> - testResetToCommittedOffset
> - testResetUsingAutoResetPolicy
> - testPollReturnsRecords
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[ https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215 ] Kirk True edited comment on KAFKA-16792 at 5/29/24 10:04 PM: - These tests work with KAFKA-16200: - testResetUsingAutoResetPolicy - testFetchStableOffsetThrowInCommitted (wasn't in the above list) The following still don't work: - testCurrentLag - testFetchStableOffsetThrowInPoll - testListOffsetShouldUpdateSubscriptions - testPollReturnsRecords - testResetToCommittedOffset was (Author: kirktrue): These tests work with KAFKA-16200: - testResetUsingAutoResetPolicy - testFetchStableOffsetThrowInCommitted (wasn't in the above list) - testFetchStableOffsetThrowInPosition (wasn't in the above list) The following still don't work: - testCurrentLag - testFetchStableOffsetThrowInPoll - testListOffsetShouldUpdateSubscriptions - testPollReturnsRecords - testResetToCommittedOffset > Enable consumer unit tests that fail to fetch offsets only for new consumer > with poll(0) > > > Key: KAFKA-16792 > URL: https://issues.apache.org/jira/browse/KAFKA-16792 > Project: Kafka > Issue Type: Sub-task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Kirk True >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > > Enable the following unit tests for the new async consumer in > KafkaConsumerTest: > - testFetchStableOffsetThrowInPoll > - testCurrentLag > - testListOffsetShouldUpdateSubscriptions > - testResetToCommittedOffset > - testResetUsingAutoResetPolicy > - testPollReturnsRecords > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Comment Edited] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[ https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215 ] Kirk True edited comment on KAFKA-16792 at 5/29/24 10:03 PM: - These tests work with KAFKA-16200: - testResetUsingAutoResetPolicy - testFetchStableOffsetThrowInCommitted (wasn't in the above list) - testFetchStableOffsetThrowInPosition (wasn't in the above list) The following still don't work: - testCurrentLag - testFetchStableOffsetThrowInPoll - testListOffsetShouldUpdateSubscriptions - testPollReturnsRecords - testResetToCommittedOffset was (Author: kirktrue): These tests work with KAFKA-16200: - testResetUsingAutoResetPolicy - testFetchStableOffsetThrowInCommitted (wasn't in the above list) - testFetchStableOffsetThrowInPosition (wasn't in the above list) The following still don't work: - testCurrentLag - testListOffsetShouldUpdateSubscriptions - testPollReturnsRecords - testResetToCommittedOffset > Enable consumer unit tests that fail to fetch offsets only for new consumer > with poll(0) > > > Key: KAFKA-16792 > URL: https://issues.apache.org/jira/browse/KAFKA-16792 > Project: Kafka > Issue Type: Sub-task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Kirk True >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > > Enable the following unit tests for the new async consumer in > KafkaConsumerTest: > - testFetchStableOffsetThrowInPoll > - testCurrentLag > - testListOffsetShouldUpdateSubscriptions > - testResetToCommittedOffset > - testResetUsingAutoResetPolicy > - testPollReturnsRecords > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (KAFKA-15305) The background thread should try to process the remaining task until the shutdown timer is expired
[ https://issues.apache.org/jira/browse/KAFKA-15305?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True reassigned KAFKA-15305: - Assignee: (was: Kirk True) > The background thread should try to process the remaining task until the > shutdown timer is expired > -- > > Key: KAFKA-15305 > URL: https://issues.apache.org/jira/browse/KAFKA-15305 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: Philip Nee >Priority: Major > Labels: consumer-threading-refactor, timeout > Fix For: 3.8.0 > > > While working on https://issues.apache.org/jira/browse/KAFKA-15304 > close() API supplies a timeout parameter so that the consumer can have a > grace period to process things before shutting down. The background thread > currently doesn't do that, when close() is initiated, it will immediately > close all of its dependencies. > > This might not be desirable because there could be remaining tasks to be > processed before closing. Maybe the correct things to do is to first stop > accepting API request, second, let the runOnce() continue to run before the > shutdown timer expires, then we can force closing all of its dependencies. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16835) Add Support for consumer to read in commit order.
[ https://issues.apache.org/jira/browse/KAFKA-16835?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16835: -- Component/s: clients > Add Support for consumer to read in commit order. > - > > Key: KAFKA-16835 > URL: https://issues.apache.org/jira/browse/KAFKA-16835 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer, offset manager >Reporter: Manjunath >Priority: Critical > > Currently consumer supports offset order to receive messages.There are some > cases where commit order is very important.For example assume case where > PostgreSQL-14 randomly streams multiple in-progress large transactions to > some intermediate client which starts transactional producer instances for > multiple in-progress transactions,using this producer instances client pushes > data to kafka. Now consumer should strictly read messages based on commit > order. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (KAFKA-16777) New consumer should throw NoOffsetForPartitionException on continuous poll zero if no reset strategy
[
https://issues.apache.org/jira/browse/KAFKA-16777?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True reassigned KAFKA-16777:
-
Assignee: Kirk True
> New consumer should throw NoOffsetForPartitionException on continuous poll
> zero if no reset strategy
>
>
> Key: KAFKA-16777
> URL: https://issues.apache.org/jira/browse/KAFKA-16777
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Reporter: Lianet Magrans
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> If the consumer does not define an offset reset strategy, a call to poll
> should fail with NoOffsetForPartitionException. That works as expected on the
> new consumer when polling with a timeout > 0 (existing integration test
> [here|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/core/src/test/scala/integration/kafka/api/PlaintextConsumerFetchTest.scala#L36]),
> but fails when polling continuously with ZERO timeout.
> This can be easily reproduced with a new integration test like this (passes
> for the legacy consumer but fails for the new consumer). We should add it as
> part of the fix, for better coverage:
> {code:java}
> @ParameterizedTest(name =
> TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
> @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
> def testNoOffsetForPartitionExceptionOnPollZero(quorum: String,
> groupProtocol: String): Unit = {
> this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
> "none")
> val consumer = createConsumer(configOverrides = this.consumerConfig)
> consumer.assign(List(tp).asJava)
> // continuous poll should eventually fail because there is no offset
> reset strategy set (fail only when resetting positions after coordinator is
> known)
> TestUtils.tryUntilNoAssertionError() {
> assertThrows(classOf[NoOffsetForPartitionException], () =>
> consumer.poll(Duration.ZERO))
> }
> }
> {code}
> Also this is covered in the unit test
> [KafkaConsumerTest.testMissingOffsetNoResetPolicy|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L915],
> that is currently enabled only for the LegacyConsumer. After fixing this
> issue we should be able to enable it for the new consumer too.
> The issue seems to be around calling poll with ZERO timeout, that even when
> called continuously, the consumer is not able to
> initWithCommittedOffsetsIfNeeded, so the updateFetchPositions never makes it
> to
> [resetInitializingPositions|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1663],
> where the exception is thrown.
>
> There is the related issue https://issues.apache.org/jira/browse/KAFKA-16637,
> but filing this one to provide more context and point out the test failures
> and suggested new tests,. All fail even with the current patch in KAFKA-16637
> so needs investigation.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[ https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17850215#comment-17850215 ] Kirk True commented on KAFKA-16792: --- These tests work with KAFKA-16200: - testResetUsingAutoResetPolicy - testFetchStableOffsetThrowInCommitted (wasn't in the above list) - testFetchStableOffsetThrowInPosition (wasn't in the above list) The following still don't work: - testCurrentLag - testListOffsetShouldUpdateSubscriptions - testPollReturnsRecords - testResetToCommittedOffset > Enable consumer unit tests that fail to fetch offsets only for new consumer > with poll(0) > > > Key: KAFKA-16792 > URL: https://issues.apache.org/jira/browse/KAFKA-16792 > Project: Kafka > Issue Type: Sub-task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Kirk True >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > > Enable the following unit tests for the new async consumer in > KafkaConsumerTest: > - testFetchStableOffsetThrowInPoll > - testCurrentLag > - testListOffsetShouldUpdateSubscriptions > - testResetToCommittedOffset > - testResetUsingAutoResetPolicy > - testPollReturnsRecords > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (KAFKA-16792) Enable consumer unit tests that fail to fetch offsets only for new consumer with poll(0)
[ https://issues.apache.org/jira/browse/KAFKA-16792?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True reassigned KAFKA-16792: - Assignee: Kirk True > Enable consumer unit tests that fail to fetch offsets only for new consumer > with poll(0) > > > Key: KAFKA-16792 > URL: https://issues.apache.org/jira/browse/KAFKA-16792 > Project: Kafka > Issue Type: Sub-task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Kirk True >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > > Enable the following unit tests for the new async consumer in > KafkaConsumerTest: > - testFetchStableOffsetThrowInPoll > - testCurrentLag > - testListOffsetShouldUpdateSubscriptions > - testResetToCommittedOffset > - testResetUsingAutoResetPolicy > - testPollReturnsRecords > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16200) Enforce that RequestManager implementations respect user-provided timeout
[
https://issues.apache.org/jira/browse/KAFKA-16200?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16200:
--
Description:
The intention of the {{CompletableApplicationEvent}} is for a {{Consumer}} to
block waiting for the event to complete. The application thread will block for
the timeout, but there is not yet a consistent manner in which network request
time outs are handled.
Enforce at the request manager layer that timeouts are respected per the design
in KAFKA-15848.
was:
The intention of the {{CompletableApplicationEvent}} is for a {{Consumer}} to
block waiting for the event to complete. The application thread will block for
the timeout, but there is not yet a consistent manner in which events are timed
out.
Enforce at the request manager layer that timeouts are respected per the design
in KAFKA-15848.
> Enforce that RequestManager implementations respect user-provided timeout
> -
>
> Key: KAFKA-16200
> URL: https://issues.apache.org/jira/browse/KAFKA-16200
> Project: Kafka
> Issue Type: Improvement
> Components: clients, consumer
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: consumer-threading-refactor, timeout
> Fix For: 3.8.0
>
>
> The intention of the {{CompletableApplicationEvent}} is for a {{Consumer}} to
> block waiting for the event to complete. The application thread will block
> for the timeout, but there is not yet a consistent manner in which network
> request time outs are handled.
> Enforce at the request manager layer that timeouts are respected per the
> design in KAFKA-15848.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Assigned] (KAFKA-15284) Implement ConsumerGroupProtocolVersionResolver to determine consumer group protocol
[
https://issues.apache.org/jira/browse/KAFKA-15284?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True reassigned KAFKA-15284:
-
Assignee: (was: Kirk True)
> Implement ConsumerGroupProtocolVersionResolver to determine consumer group
> protocol
> ---
>
> Key: KAFKA-15284
> URL: https://issues.apache.org/jira/browse/KAFKA-15284
> Project: Kafka
> Issue Type: New Feature
> Components: clients, consumer
>Reporter: Kirk True
>Priority: Major
> Labels: kip-848-client-support
> Fix For: 4.0.0
>
>
> At client initialization, we need to determine which of the
> {{ConsumerDelegate}} implementations to use:
> # {{LegacyKafkaConsumerDelegate}}
> # {{AsyncKafkaConsumerDelegate}}
> There are conditions defined by KIP-848 that determine client eligibility to
> use the new protocol. This will be modeled by the—deep
> breath—{{{}ConsumerGroupProtocolVersionResolver{}}}.
> Known tasks:
> * Determine at what point in the {{Consumer}} initialization the network
> communication should happen
> * Determine what RPCs to invoke in order to determine eligibility (API
> versions, IBP version, etc.)
> * Implement the network client lifecycle (startup, communication, shutdown,
> etc.)
> * Determine the fallback path in case the client is not eligible to use the
> protocol
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16639) AsyncKafkaConsumer#close does not send heartbeat to leave group
[ https://issues.apache.org/jira/browse/KAFKA-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16639: -- Priority: Critical (was: Major) > AsyncKafkaConsumer#close does not send heartbeat to leave group > --- > > Key: KAFKA-16639 > URL: https://issues.apache.org/jira/browse/KAFKA-16639 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: Chia-Ping Tsai >Assignee: Philip Nee >Priority: Critical > Labels: kip-848-client-support > Fix For: 3.8.0 > > > This bug can be reproduced by immediately closing a consumer which is just > created. > The root cause is that we skip the new heartbeat used to leave group when > there is a in-flight heartbeat request > ([https://github.com/apache/kafka/blob/5de5d967adffd864bad3ec729760a430253abf38/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java#L212).] > It seems to me the simple solution is that we create a heartbeat request when > meeting above situation and then send it by pollOnClose > ([https://github.com/apache/kafka/blob/5de5d967adffd864bad3ec729760a430253abf38/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java#L62).] > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16819) CoordinatorRequestManager seems to return 0ms during the coordinator discovery
[ https://issues.apache.org/jira/browse/KAFKA-16819?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16819: -- Labels: consumer-threading-refactor (was: ) > CoordinatorRequestManager seems to return 0ms during the coordinator discovery > -- > > Key: KAFKA-16819 > URL: https://issues.apache.org/jira/browse/KAFKA-16819 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: Philip Nee >Assignee: Philip Nee >Priority: Minor > Labels: consumer-threading-refactor > > In KAFKA-15250 we discovered the ConsumerNetworkThread is looping without > much backoff. The in-flight check PR fixed a lot of it; however, during the > coordinator discovery phase, CoordinatorRequestManager would keep on > returning 0 before the coordinator node was found. > > The impact is minor but we should be expecting the coordinator manager to > backoff until the exp backoff expired (so it should return around 100ms). -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16819) CoordinatorRequestManager seems to return 0ms during the coordinator discovery
[ https://issues.apache.org/jira/browse/KAFKA-16819?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16819: -- Component/s: clients > CoordinatorRequestManager seems to return 0ms during the coordinator discovery > -- > > Key: KAFKA-16819 > URL: https://issues.apache.org/jira/browse/KAFKA-16819 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: Philip Nee >Assignee: Philip Nee >Priority: Minor > > In KAFKA-15250 we discovered the ConsumerNetworkThread is looping without > much backoff. The in-flight check PR fixed a lot of it; however, during the > coordinator discovery phase, CoordinatorRequestManager would keep on > returning 0 before the coordinator node was found. > > The impact is minor but we should be expecting the coordinator manager to > backoff until the exp backoff expired (so it should return around 100ms). -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16818) Move event processing-related tests from ConsumerNetworkThreadTest to ApplicationEventProcessorTest
[
https://issues.apache.org/jira/browse/KAFKA-16818?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16818:
--
Summary: Move event processing-related tests from ConsumerNetworkThreadTest
to ApplicationEventProcessorTest (was: Move event-processing tests from
ConsumerNetworkThreadTest to ApplicationEventProcessorTest)
> Move event processing-related tests from ConsumerNetworkThreadTest to
> ApplicationEventProcessorTest
> ---
>
> Key: KAFKA-16818
> URL: https://issues.apache.org/jira/browse/KAFKA-16818
> Project: Kafka
> Issue Type: Improvement
> Components: clients, consumer, unit tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Minor
> Labels: consumer-threading-refactor
> Fix For: 3.8.0
>
>
> The {{ConsumerNetworkThreadTest}} currently has a number of tests which do
> the following:
> # Add event of type _T_ to the event queue
> # Call {{ConsumerNetworkThread.runOnce()}} to dequeue the events and call
> {{ApplicationEventProcessor.process()}}
> # Verify that the appropriate {{ApplicationEventProcessor}} process method
> was invoked for the event
> Those types of tests should be moved to {{{}ApplicationEventProcessorTest{}}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16818) Move event-processing tests from ConsumerNetworkThreadTest to ApplicationEventProcessorTest
Kirk True created KAFKA-16818:
-
Summary: Move event-processing tests from
ConsumerNetworkThreadTest to ApplicationEventProcessorTest
Key: KAFKA-16818
URL: https://issues.apache.org/jira/browse/KAFKA-16818
Project: Kafka
Issue Type: Improvement
Components: clients, consumer, unit tests
Affects Versions: 3.8.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
The {{ConsumerNetworkThreadTest}} currently has a number of tests which do the
following:
# Add event of type _T_ to the event queue
# Call {{ConsumerNetworkThread.runOnce()}} to dequeue the events and call
{{ApplicationEventProcessor.process()}}
# Verify that the appropriate {{ApplicationEventProcessor}} process method was
invoked for the event
Those types of tests should be moved to {{{}ApplicationEventProcessorTest{}}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Resolved] (KAFKA-16578) Revert changes to connect_distributed_test.py for the new async Consumer
[
https://issues.apache.org/jira/browse/KAFKA-16578?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True resolved KAFKA-16578.
---
Resolution: Won't Fix
Most of the {{connect_distributed_test.py}} system tests were fixed, and
{{test_exactly_once_source}} was reverted in a separate Jira/PR.
> Revert changes to connect_distributed_test.py for the new async Consumer
>
>
> Key: KAFKA-16578
> URL: https://issues.apache.org/jira/browse/KAFKA-16578
> Project: Kafka
> Issue Type: Task
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated
> a slew of system tests to run both the "old" and "new" implementations.
> KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
> could test the new consumer with Connect. However, we are not supporting
> Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the
> Connect system tests with the new {{AsyncKafkaConsumer}}, we get errors like
> the following:
> {code}
> test_id:
> kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 6 minutes 3.899 seconds
> InsufficientResourcesError('Not enough nodes available to allocate. linux
> nodes requested: 1. linux nodes available: 0')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
> line 919, in test_exactly_once_source
> consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
> self.source.topic, consumer_timeout_ms=1000, print_key=True)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
> line 97, in __init__
> BackgroundThreadService.__init__(self, context, num_nodes)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
> line 26, in __init__
> super(BackgroundThreadService, self).__init__(context, num_nodes,
> cluster_spec, *args, **kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
> line 107, in __init__
> self.allocate_nodes()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
> line 217, in allocate_nodes
> self.nodes = self.cluster.alloc(self.cluster_spec)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/cluster.py",
> line 54, in alloc
> allocated = self.do_alloc(cluster_spec)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/finite_subcluster.py",
> line 31, in do_alloc
> allocated = self._available_nodes.remove_spec(cluster_spec)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/node_container.py",
> line 117, in remove_spec
> raise InsufficientResourcesError("Not enough nodes available to allocate.
> " + msg)
> ducktape.cluster.node_container.InsufficientResourcesError: Not enough nodes
> available to allocate. linux nodes requested: 1. linux nodes available: 0
> {code}
> The task here is to revert the changes made in KAFKA-16272 [PR
> 15576|https://github.com/apache/kafka/pull/15576].
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16799) NetworkClientDelegate is not backing off if the node is not found
[
https://issues.apache.org/jira/browse/KAFKA-16799?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16799:
--
Labels: consumer-threading-refactor (was: )
> NetworkClientDelegate is not backing off if the node is not found
> -
>
> Key: KAFKA-16799
> URL: https://issues.apache.org/jira/browse/KAFKA-16799
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Reporter: Philip Nee
>Assignee: Philip Nee
>Priority: Major
> Labels: consumer-threading-refactor
>
> When performing stress testing, I found that AsycnKafkaConsumer's network
> client delegate isn't backing off if the node is not ready, causing a large
> number of:
> {code:java}
> 358 [2024-05-20 22:59:02,591] DEBUG [Consumer
> clientId=consumer.7136899e-0c20-4ccb-8ba3-497e9e683594-0,
> groupId=consumer-groups-test-5] Node is not ready, handle the request in the
> next event loop: node=b4-pkc-devcmkz697.us-west-2.aws.devel.cpd
> ev.cloud:9092 (id: 2147483643 rack: null),
> request=UnsentRequest{requestBuilder=ConsumerGroupHeartbeatRequestData(groupId='consumer-groups-test-5',
> memberId='', memberEpoch=0, instanceId=null, rackId=null,
> rebalanceTimeoutMs=10, subscri
> bedTopicNames=[_kengine-565-test-topic8081], serverAssignor=null,
> topicPartitions=[]),
> handler=org.apache.kafka.clients.consumer.internals.NetworkClientDelegate$FutureCompletionHandler@139a8761,
> node=Optional[b4-pkc-devcmkz697.us-west-2.aws .devel.cpdev.cloud:9092
> (id: 2147483643 rack: null)],
> timer=org.apache.kafka.common.utils.Timer@649fffad}
> (org.apache.kafka.clients.consumer.internals.NetworkClientDelegate:169) {code}
> show up in the log.
> What should have happened is: 1. node is not ready 2. exponential back off 3.
> retry
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16799) NetworkClientDelegate is not backing off if the node is not found
[
https://issues.apache.org/jira/browse/KAFKA-16799?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16799:
--
Component/s: clients
> NetworkClientDelegate is not backing off if the node is not found
> -
>
> Key: KAFKA-16799
> URL: https://issues.apache.org/jira/browse/KAFKA-16799
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Reporter: Philip Nee
>Assignee: Philip Nee
>Priority: Major
>
> When performing stress testing, I found that AsycnKafkaConsumer's network
> client delegate isn't backing off if the node is not ready, causing a large
> number of:
> {code:java}
> 358 [2024-05-20 22:59:02,591] DEBUG [Consumer
> clientId=consumer.7136899e-0c20-4ccb-8ba3-497e9e683594-0,
> groupId=consumer-groups-test-5] Node is not ready, handle the request in the
> next event loop: node=b4-pkc-devcmkz697.us-west-2.aws.devel.cpd
> ev.cloud:9092 (id: 2147483643 rack: null),
> request=UnsentRequest{requestBuilder=ConsumerGroupHeartbeatRequestData(groupId='consumer-groups-test-5',
> memberId='', memberEpoch=0, instanceId=null, rackId=null,
> rebalanceTimeoutMs=10, subscri
> bedTopicNames=[_kengine-565-test-topic8081], serverAssignor=null,
> topicPartitions=[]),
> handler=org.apache.kafka.clients.consumer.internals.NetworkClientDelegate$FutureCompletionHandler@139a8761,
> node=Optional[b4-pkc-devcmkz697.us-west-2.aws .devel.cpdev.cloud:9092
> (id: 2147483643 rack: null)],
> timer=org.apache.kafka.common.utils.Timer@649fffad}
> (org.apache.kafka.clients.consumer.internals.NetworkClientDelegate:169) {code}
> show up in the log.
> What should have happened is: 1. node is not ready 2. exponential back off 3.
> retry
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (KAFKA-16272) Update connect_distributed_test.py to support KIP-848’s group protocol config
[
https://issues.apache.org/jira/browse/KAFKA-16272?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17847953#comment-17847953
]
Kirk True commented on KAFKA-16272:
---
[~sagarrao]—both of the linked PRs are merged. Can this be marked as Resolved?
Thanks!
> Update connect_distributed_test.py to support KIP-848’s group protocol config
> -
>
> Key: KAFKA-16272
> URL: https://issues.apache.org/jira/browse/KAFKA-16272
> Project: Kafka
> Issue Type: Test
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Sagar Rao
>Priority: Major
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> This task is to update the test method(s) in {{connect_distributed_test.py}}
> to support the {{group.protocol}} configuration introduced in
> [KIP-848|https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol]
> by adding an optional {{group_protocol}} argument to the tests and matrixes.
> See KAFKA-16231 as an example of how the test parameters can be changed.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16777) New consumer should throw NoOffsetForPartitionException on continuous poll zero if no reset strategy
[
https://issues.apache.org/jira/browse/KAFKA-16777?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16777:
--
Fix Version/s: 3.8.0
> New consumer should throw NoOffsetForPartitionException on continuous poll
> zero if no reset strategy
>
>
> Key: KAFKA-16777
> URL: https://issues.apache.org/jira/browse/KAFKA-16777
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Reporter: Lianet Magrans
>Priority: Major
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> If the consumer does not define an offset reset strategy, a call to poll
> should fail with NoOffsetForPartitionException. That works as expected on the
> new consumer when polling with a timeout > 0 (existing integration test
> [here|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/core/src/test/scala/integration/kafka/api/PlaintextConsumerFetchTest.scala#L36]),
> but fails when polling continuously with ZERO timeout.
> This can be easily reproduced with a new integration test like this (passes
> for the legacy consumer but fails for the new consumer). We should add it as
> part of the fix, for better coverage:
> {code:java}
> @ParameterizedTest(name =
> TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
> @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
> def testNoOffsetForPartitionExceptionOnPollZero(quorum: String,
> groupProtocol: String): Unit = {
> this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
> "none")
> val consumer = createConsumer(configOverrides = this.consumerConfig)
> consumer.assign(List(tp).asJava)
> // continuous poll should eventually fail because there is no offset
> reset strategy set (fail only when resetting positions after coordinator is
> known)
> TestUtils.tryUntilNoAssertionError() {
> assertThrows(classOf[NoOffsetForPartitionException], () =>
> consumer.poll(Duration.ZERO))
> }
> }
> {code}
> Also this is covered in the unit test
> [KafkaConsumerTest.testMissingOffsetNoResetPolicy|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L915],
> that is currently enabled only for the LegacyConsumer. After fixing this
> issue we should be able to enable it for the new consumer too.
> The issue seems to be around calling poll with ZERO timeout, that even when
> called continuously, the consumer is not able to
> initWithCommittedOffsetsIfNeeded, so the updateFetchPositions never makes it
> to
> [resetInitializingPositions|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1663],
> where the exception is thrown.
>
> There is the related issue https://issues.apache.org/jira/browse/KAFKA-16637,
> but filing this one to provide more context and point out the test failures
> and suggested new tests,. All fail even with the current patch in KAFKA-16637
> so needs investigation.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16758) Extend Consumer#close with option to leave the group or not
[ https://issues.apache.org/jira/browse/KAFKA-16758?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16758: -- Fix Version/s: 4.0.0 > Extend Consumer#close with option to leave the group or not > --- > > Key: KAFKA-16758 > URL: https://issues.apache.org/jira/browse/KAFKA-16758 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer >Reporter: A. Sophie Blee-Goldman >Assignee: Lianet Magrans >Priority: Major > Labels: needs-kip > Fix For: 4.0.0 > > > See comments on https://issues.apache.org/jira/browse/KAFKA-16514 for the > full context. > Essentially we would get rid of the "internal.leave.group.on.close" config > that is used as a backdoor by Kafka Streams right now to prevent closed > consumers from leaving the group, thus reducing unnecessary task movements > after a simple bounce. > This would be replaced by an actual public API that would allow the caller to > opt in or out to the LeaveGroup when close is called. This would be similar > to the KafkaStreams#close(CloseOptions) API, and in fact would be how that > API will be implemented (since it only works for static groups at the moment > as noted in KAFKA-16514 ) > This has several benefits over the current situation: > # It allows plain consumer apps to opt-out of leaving the group when closed, > which is currently not possible through any public API (only an internal > backdoor config) > # It enables the caller to dynamically select the appropriate action > depending on why the client is being closed – for example, you would not want > the consumer to leave the group during a simple restart, but would want it to > leave the group when shutting down the app or if scaling down the node. This > is not possible today, even with the internal config, since configs are > immutable > # It can be leveraged to "fix" the KafkaStreams#close(closeOptions) API so > that the user's choice to leave the group during close will be respected for > non-static members -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16766) New consumer offsetsForTimes timeout exception does not have the proper message
[ https://issues.apache.org/jira/browse/KAFKA-16766?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16766: -- Component/s: clients > New consumer offsetsForTimes timeout exception does not have the proper > message > --- > > Key: KAFKA-16766 > URL: https://issues.apache.org/jira/browse/KAFKA-16766 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Affects Versions: 3.7.0 >Reporter: Lianet Magrans >Priority: Major > Labels: kip-848-client-support > Fix For: 3.8.0 > > > If a call to consumer.offsetsForTimes times out, the new AsyncKafkaConsumer > will throw a org.apache.kafka.common.errors.TimeoutException as expected, but > with the following as message: "java.util.concurrent.TimeoutException". > We should provide a clearer message, and I would even say we keep the same > message that the LegacyConsumer shows in this case, ex: "Failed to get > offsets by times in 6ms". > To fix this we should consider catching the timeout exception in the consumer > when offsetsForTimes result times out > ([here|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1115]), > and propagate it with the message specific to offsetsForTimes. > Same situation exists for beginningOffsets and endOffsets. All 3 funcs show > the same timeout message in the LegacyConsumer (defined > [here|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/main/java/org/apache/kafka/clients/consumer/internals/OffsetFetcher.java#L182]), > but do not have a clear message in the Async, so we should fix them all 3. > With the fix, we should write tests for each func, like the ones defined for > the Legacy Consumer > ([here|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L3244-L3276]). > Note that we would need different tests, added to AsyncKafkaConsumerTest, > given that the async consumer issues a FindCoordinator request in this case, > but the AsyncConsumer does, so it does not account for that when matching > requests/responses in the current tests. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16764) New consumer should throw InvalidTopicException on poll when invalid topic in metadata
[
https://issues.apache.org/jira/browse/KAFKA-16764?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16764:
--
Component/s: clients
> New consumer should throw InvalidTopicException on poll when invalid topic in
> metadata
> --
>
> Key: KAFKA-16764
> URL: https://issues.apache.org/jira/browse/KAFKA-16764
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Affects Versions: 3.7.0
>Reporter: Lianet Magrans
>Assignee: appchemist
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> A call to consumer.poll should throw InvalidTopicException if an invalid
> topic is discovered in metadata. This can be easily reproduced by calling
> subscribe("invalid topic") and then poll, for example.The new consumer does
> not throw the expected InvalidTopicException like the LegacyKafkaConsumer
> does.
> The legacy consumer achieves this by checking for metadata exceptions on
> every iteration of the ConsumerNetworkClient (see
> [here|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java#L315])
> This is probably what makes that
> [testSubscriptionOnInvalidTopic|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L2956]
> fails for the new consumer. Once this bug is fixed, we should be able to
> enable that test for the new consumer.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Assigned] (KAFKA-16787) Remove TRACE level logging from AsyncKafkaConsumer hot path
[
https://issues.apache.org/jira/browse/KAFKA-16787?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True reassigned KAFKA-16787:
-
Assignee: Kirk True
> Remove TRACE level logging from AsyncKafkaConsumer hot path
> ---
>
> Key: KAFKA-16787
> URL: https://issues.apache.org/jira/browse/KAFKA-16787
> Project: Kafka
> Issue Type: Improvement
> Components: clients, consumer, logging
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor
> Fix For: 3.8.0
>
>
> There are a few places in the new {{AsyncKafkaConsumer}} in which we added
> TRACE-level logging to common paths (like the
> {{{}ApplicationEventHandler{}}}). The logging is so overboard that when
> running system tests, we occasionally run out of disk space on the test
> instances, causing spurious test failures.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16787) Remove TRACE level logging from AsyncKafkaConsumer hot path
[
https://issues.apache.org/jira/browse/KAFKA-16787?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16787:
--
Description: There are a few places in the new {{AsyncKafkaConsumer}} in
which we added TRACE-level logging to common paths (like the
{{{}ApplicationEventHandler{}}}). The logging is so overboard that when running
system tests, we occasionally run out of disk space on the test instances,
causing spurious test failures.
> Remove TRACE level logging from AsyncKafkaConsumer hot path
> ---
>
> Key: KAFKA-16787
> URL: https://issues.apache.org/jira/browse/KAFKA-16787
> Project: Kafka
> Issue Type: Improvement
> Components: clients, consumer, logging
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor
> Fix For: 3.8.0
>
>
> There are a few places in the new {{AsyncKafkaConsumer}} in which we added
> TRACE-level logging to common paths (like the
> {{{}ApplicationEventHandler{}}}). The logging is so overboard that when
> running system tests, we occasionally run out of disk space on the test
> instances, causing spurious test failures.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16787) Remove TRACE level logging from AsyncKafkaConsumer hot path
Kirk True created KAFKA-16787: - Summary: Remove TRACE level logging from AsyncKafkaConsumer hot path Key: KAFKA-16787 URL: https://issues.apache.org/jira/browse/KAFKA-16787 Project: Kafka Issue Type: Improvement Components: clients, consumer, logging Affects Versions: 3.7.0 Reporter: Kirk True Fix For: 3.8.0 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16787) Remove TRACE level logging from AsyncKafkaConsumer hot path
[ https://issues.apache.org/jira/browse/KAFKA-16787?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16787: -- Labels: consumer-threading-refactor (was: ) > Remove TRACE level logging from AsyncKafkaConsumer hot path > --- > > Key: KAFKA-16787 > URL: https://issues.apache.org/jira/browse/KAFKA-16787 > Project: Kafka > Issue Type: Improvement > Components: clients, consumer, logging >Affects Versions: 3.7.0 >Reporter: Kirk True >Priority: Critical > Labels: consumer-threading-refactor > Fix For: 3.8.0 > > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16786) New consumer subscribe should not require the deprecated partition.assignment.strategy
[ https://issues.apache.org/jira/browse/KAFKA-16786?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16786: -- Component/s: clients > New consumer subscribe should not require the deprecated > partition.assignment.strategy > -- > > Key: KAFKA-16786 > URL: https://issues.apache.org/jira/browse/KAFKA-16786 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Affects Versions: 3.7.0 >Reporter: Lianet Magrans >Priority: Major > Labels: kip-848-client-support > Fix For: 3.8 > > > The partition.assignment.strategy config is deprecated with the new consumer > group protocol KIP-848. With the new protocol, server side assignors are > supported for now, defined in the property > group.remote.assignor, and with default values selected by the broker, so > it's not even a required property. > The new AsyncKafkaConsumer supports the new protocol only, but it currently > throws an IllegalStateException if a call to subscribe is made and the > deprecated config partition.assignment.strategy is empty (see > [throwIfNoAssignorsConfigured|https://github.com/apache/kafka/blob/056d232f4e28bf8e67e00f8ed2c103fdb0f3b78e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1715]). > > We should remove the reference to ConsumerPartitionAssignor in the > AsyncKafkaConsumer, along with it's validation for non-empty on subscribe > (only use it has) -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16777) New consumer should throw NoOffsetForPartitionException on continuous poll zero if no reset strategy
[
https://issues.apache.org/jira/browse/KAFKA-16777?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16777:
--
Component/s: clients
> New consumer should throw NoOffsetForPartitionException on continuous poll
> zero if no reset strategy
>
>
> Key: KAFKA-16777
> URL: https://issues.apache.org/jira/browse/KAFKA-16777
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Reporter: Lianet Magrans
>Priority: Major
> Labels: kip-848-client-support
>
> If the consumer does not define an offset reset strategy, a call to poll
> should fail with NoOffsetForPartitionException. That works as expected on the
> new consumer when polling with a timeout > 0 (existing integration test
> [here|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/core/src/test/scala/integration/kafka/api/PlaintextConsumerFetchTest.scala#L36]),
> but fails when polling continuously with ZERO timeout.
> This can be easily reproduced with a new integration test like this (passes
> for the legacy consumer but fails for the new consumer). We should add it as
> part of the fix, for better coverage:
> {code:java}
> @ParameterizedTest(name =
> TestInfoUtils.TestWithParameterizedQuorumAndGroupProtocolNames)
> @MethodSource(Array("getTestQuorumAndGroupProtocolParametersAll"))
> def testNoOffsetForPartitionExceptionOnPollZero(quorum: String,
> groupProtocol: String): Unit = {
> this.consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
> "none")
> val consumer = createConsumer(configOverrides = this.consumerConfig)
> consumer.assign(List(tp).asJava)
> // continuous poll should eventually fail because there is no offset
> reset strategy set (fail only when resetting positions after coordinator is
> known)
> TestUtils.tryUntilNoAssertionError() {
> assertThrows(classOf[NoOffsetForPartitionException], () =>
> consumer.poll(Duration.ZERO))
> }
> }
> {code}
> Also this is covered in the unit test
> [KafkaConsumerTest.testMissingOffsetNoResetPolicy|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L915],
> that is currently enabled only for the LegacyConsumer. After fixing this
> issue we should be able to enable it for the new consumer too.
> The issue seems to be around calling poll with ZERO timeout, that even when
> continuously, the consumer is not able to initWithCommittedOffsetsIfNeeded,
> so the updateFetchPositions never makes it to
> [resetInitializingPositions|https://github.com/apache/kafka/blob/ba19eedb90fc3a3efc97714e3ea125647338fc66/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java#L1663],
> where the exception is thrown.
>
> There is the related issue https://issues.apache.org/jira/browse/KAFKA-16637,
> but filing this one to provide more context and point out the test failures
> and suggested new tests,. All fail even with the current patch in KAFKA-16637
> so needs investigation.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16764) New consumer should throw InvalidTopicException on poll when invalid topic in metadata
[
https://issues.apache.org/jira/browse/KAFKA-16764?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16764:
--
Fix Version/s: 3.8.0
> New consumer should throw InvalidTopicException on poll when invalid topic in
> metadata
> --
>
> Key: KAFKA-16764
> URL: https://issues.apache.org/jira/browse/KAFKA-16764
> Project: Kafka
> Issue Type: Bug
> Components: consumer
>Affects Versions: 3.7.0
>Reporter: Lianet Magrans
>Priority: Blocker
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
>
> A call to consumer.poll should throw InvalidTopicException if an invalid
> topic is discovered in metadata. This can be easily reproduced by calling
> subscribe("invalid topic") and then poll, for example.The new consumer does
> not throw the expected InvalidTopicException like the LegacyKafkaConsumer
> does.
> The legacy consumer achieves this by checking for metadata exceptions on
> every iteration of the ConsumerNetworkClient (see
> [here|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java#L315])
> This is probably what makes that
> [testSubscriptionOnInvalidTopic|https://github.com/apache/kafka/blob/0e023e1f736ea03568032cc282803df8e61eb451/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java#L2956]
> fails for the new consumer. Once this bug is fixed, we should be able to
> enable that test for the new consumer.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-15284) Implement ConsumerGroupProtocolVersionResolver to determine consumer group protocol
[
https://issues.apache.org/jira/browse/KAFKA-15284?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-15284:
--
Fix Version/s: 4.0.0
> Implement ConsumerGroupProtocolVersionResolver to determine consumer group
> protocol
> ---
>
> Key: KAFKA-15284
> URL: https://issues.apache.org/jira/browse/KAFKA-15284
> Project: Kafka
> Issue Type: New Feature
> Components: clients, consumer
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Major
> Labels: kip-848-client-support
> Fix For: 4.0.0
>
>
> At client initialization, we need to determine which of the
> {{ConsumerDelegate}} implementations to use:
> # {{LegacyKafkaConsumerDelegate}}
> # {{AsyncKafkaConsumerDelegate}}
> There are conditions defined by KIP-848 that determine client eligibility to
> use the new protocol. This will be modeled by the—deep
> breath—{{{}ConsumerGroupProtocolVersionResolver{}}}.
> Known tasks:
> * Determine at what point in the {{Consumer}} initialization the network
> communication should happen
> * Determine what RPCs to invoke in order to determine eligibility (API
> versions, IBP version, etc.)
> * Implement the network client lifecycle (startup, communication, shutdown,
> etc.)
> * Determine the fallback path in case the client is not eligible to use the
> protocol
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-15697) Add local assignor and ensure it cannot be used with server side assignor
[ https://issues.apache.org/jira/browse/KAFKA-15697?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-15697: -- Fix Version/s: 4.0.0 > Add local assignor and ensure it cannot be used with server side assignor > - > > Key: KAFKA-15697 > URL: https://issues.apache.org/jira/browse/KAFKA-15697 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer >Reporter: Philip Nee >Priority: Major > Labels: kip-848-client-support > Fix For: 4.0.0 > > > When we start supporting local/client-side assignor, we should: > # Add the config to ConsumerConfig > # Examine where should we implement to logic to ensure it is not used along > side with the server side assignor, i.e. you can only specify local or remote > assignor, or non. > ## If both assignors are specified: Throw illegalArgumentException -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-15282) Implement client support for KIP-848 client-side assignors
[
https://issues.apache.org/jira/browse/KAFKA-15282?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-15282:
--
Fix Version/s: 4.0.0
> Implement client support for KIP-848 client-side assignors
> --
>
> Key: KAFKA-15282
> URL: https://issues.apache.org/jira/browse/KAFKA-15282
> Project: Kafka
> Issue Type: New Feature
> Components: clients, consumer
>Reporter: Kirk True
>Priority: Major
> Labels: kip-848-client-support
> Fix For: 4.0.0
>
>
> The client-side assignor provides the logic for the partition assignments
> instead of on the server. Client-side assignment is the main approach used by
> the “old protocol” for divvying up partitions. While the “new protocol”
> favors server-side assignors, the client-side assignor will continue to be
> used for backward compatibility, including KSQL, Connect, etc.
> Note: I _*think*_ that the client-side assignor logic and the reconciliation
> logic can remain separate from each other. We should strive to keep the two
> pieces unencumbered, unless it’s unavoidable.
> This task includes:
> * Validate the client’s configuration for assignor selection
> * Integrate with the new {{PartitionAssignor}} interface to invoke the logic
> from the user-provided assignor implementation
> * Implement the necessary logic around the request/response from the
> {{ConsumerGroupPrepareAssignment}} RPC call using the information from the
> {{PartitionAssignor}} above
> * Implement the necessary logic around the request/response from the
> {{ConsumerGroupInstallAssignment}} RPC call, again using the information
> calculated by the {{PartitionAssignor}}
> This task is part of the work to implement support for the new KIP-848
> consumer group protocol.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-15279) Implement client support for KIP-848 client-side assigner RPCs
[ https://issues.apache.org/jira/browse/KAFKA-15279?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-15279: -- Fix Version/s: 4.0.0 > Implement client support for KIP-848 client-side assigner RPCs > -- > > Key: KAFKA-15279 > URL: https://issues.apache.org/jira/browse/KAFKA-15279 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer >Reporter: Kirk True >Priority: Major > Labels: kip-848-client-support > Fix For: 4.0.0 > > > The protocol introduces three new RPCs that the client uses to communicate > with the broker: > # > [ConsumerGroupHeartbeat|https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol#KIP848:TheNextGenerationoftheConsumerRebalanceProtocol-ConsumerGroupHeartbeatAPI] > # > [ConsumerGroupPrepareAssignment|https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol#KIP848:TheNextGenerationoftheConsumerRebalanceProtocol-ConsumerGroupPrepareAssignmentAPI] > # > [ConsumerGroupInstallAssignment|https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol#KIP848:TheNextGenerationoftheConsumerRebalanceProtocol-ConsumerGroupInstallAssignmentAPI] > Support for ConsumerGroupHeartbeat is handled by KAFKA-15278. This task is to > implement the ConsumerGroupAssignmentRequestManager to handle the second and > third RPCs on the above list. > This task is part of the work to implement support for the new KIP-848 > consumer group protocol. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (KAFKA-16406) Split long-running consumer integration test
[ https://issues.apache.org/jira/browse/KAFKA-16406?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17846354#comment-17846354 ] Kirk True commented on KAFKA-16406: --- [~lianetm]—both PRs are closed. Can this be marked as resolved? > Split long-running consumer integration test > > > Key: KAFKA-16406 > URL: https://issues.apache.org/jira/browse/KAFKA-16406 > Project: Kafka > Issue Type: Task >Reporter: Lianet Magrans >Assignee: Lianet Magrans >Priority: Major > Fix For: 3.8.0 > > > PlaintextConsumerTest contains integration tests for the consumer. Since the > introduction of the new consumer group protocol (KIP-848) and the new > KafkaConsumer, this test has been parametrized to run with multiple > combinations, making sure we test the logic for the old and new coordinator, > as well as for the legacy and new KafkaConsumer. > This led to this being one of the longest-running integration tests, so in > the aim of reducing the impact on the build times we could split it to allow > for parallelization. The tests covers multiple areas of the consumer logic, > in a single file, so splitting based on the high-level features being tested > would be sensible and achieve the result wanted. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16406) Split long-running consumer integration test
[ https://issues.apache.org/jira/browse/KAFKA-16406?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16406: -- Fix Version/s: 3.8.0 > Split long-running consumer integration test > > > Key: KAFKA-16406 > URL: https://issues.apache.org/jira/browse/KAFKA-16406 > Project: Kafka > Issue Type: Task >Reporter: Lianet Magrans >Assignee: Lianet Magrans >Priority: Major > Fix For: 3.8.0 > > > PlaintextConsumerTest contains integration tests for the consumer. Since the > introduction of the new consumer group protocol (KIP-848) and the new > KafkaConsumer, this test has been parametrized to run with multiple > combinations, making sure we test the logic for the old and new coordinator, > as well as for the legacy and new KafkaConsumer. > This led to this being one of the longest-running integration tests, so in > the aim of reducing the impact on the build times we could split it to allow > for parallelization. The tests covers multiple areas of the consumer logic, > in a single file, so splitting based on the high-level features being tested > would be sensible and achieve the result wanted. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16599) LegacyConsumer: Always await async commit callbacks in commitSync and close
[
https://issues.apache.org/jira/browse/KAFKA-16599?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16599:
--
Fix Version/s: 3.8.0
> LegacyConsumer: Always await async commit callbacks in commitSync and close
> ---
>
> Key: KAFKA-16599
> URL: https://issues.apache.org/jira/browse/KAFKA-16599
> Project: Kafka
> Issue Type: Task
> Components: clients, consumer
>Reporter: Lucas Brutschy
>Assignee: Lucas Brutschy
>Priority: Major
> Fix For: 3.8.0
>
>
> The javadoc for KafkaConsumer.commitSync says:
> {code:java}
> Note that asynchronous offset commits sent previously with the {@link
> #commitAsync(OffsetCommitCallback)}
> (or similar) are guaranteed to have their callbacks invoked prior to
> completion of this method.
> {code}
> This is not always true in the legacy consumer, when the set of offsets is
> empty, the execution of the commit callback is not always awaited. There are
> also various races possible that can avoid callback handler execution.
> Similarly, there is code in the legacy consumer to await the completion of
> the commit callback before closing, however, the code doesn't cover all cases
> and the behavior is therefore inconsistent. While the Javadoc doesn't
> explicitly promise callback execution, it promises "completing commits",
> which one would reasonably expect to include callback execution. Either way,
> the current behavior of the legacy consumer is inconsistent.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16639) AsyncKafkaConsumer#close does not send heartbeat to leave group
[ https://issues.apache.org/jira/browse/KAFKA-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16639: -- Fix Version/s: 3.8.0 > AsyncKafkaConsumer#close does not send heartbeat to leave group > --- > > Key: KAFKA-16639 > URL: https://issues.apache.org/jira/browse/KAFKA-16639 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: Chia-Ping Tsai >Assignee: Philip Nee >Priority: Major > Labels: kip-848-client-support > Fix For: 3.8.0 > > > This bug can be reproduced by immediately closing a consumer which is just > created. > The root cause is that we skip the new heartbeat used to leave group when > there is a in-flight heartbeat request > ([https://github.com/apache/kafka/blob/5de5d967adffd864bad3ec729760a430253abf38/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java#L212).] > It seems to me the simple solution is that we create a heartbeat request when > meeting above situation and then send it by pollOnClose > ([https://github.com/apache/kafka/blob/5de5d967adffd864bad3ec729760a430253abf38/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java#L62).] > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16675) Move rebalance callback test for positions to callbacks test file
[ https://issues.apache.org/jira/browse/KAFKA-16675?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16675: -- Fix Version/s: 4.0.0 > Move rebalance callback test for positions to callbacks test file > - > > Key: KAFKA-16675 > URL: https://issues.apache.org/jira/browse/KAFKA-16675 > Project: Kafka > Issue Type: Task > Components: consumer >Reporter: Lianet Magrans >Assignee: Lianet Magrans >Priority: Minor > Labels: kip-848-client-support > Fix For: 4.0.0 > > > Integration test > testGetPositionOfNewlyAssignedPartitionFromPartitionsAssignedCallback was > added to the PlaintextConsumerTest.scala in this PR > https://github.com/apache/kafka/pull/15856, as there was no specific file for > testing callbacks at the moment. Another PR is in-flight, adding the file for > callback-related tests, https://github.com/apache/kafka/pull/15408. Once > 15408 gets merged, we should move > testGetPositionOfNewlyAssignedPartitionFromPartitionsAssignedCallback to it. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16758) Extend Consumer#close with option to leave the group or not
[ https://issues.apache.org/jira/browse/KAFKA-16758?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16758: -- Component/s: clients > Extend Consumer#close with option to leave the group or not > --- > > Key: KAFKA-16758 > URL: https://issues.apache.org/jira/browse/KAFKA-16758 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer >Reporter: A. Sophie Blee-Goldman >Priority: Major > Labels: needs-kip > > See comments on https://issues.apache.org/jira/browse/KAFKA-16514 for the > full context. > Essentially we would get rid of the "internal.leave.group.on.close" config > that is used as a backdoor by Kafka Streams right now to prevent closed > consumers from leaving the group, thus reducing unnecessary task movements > after a simple bounce. > This would be replaced by an actual public API that would allow the caller to > opt in or out to the LeaveGroup when close is called. This would be similar > to the KafkaStreams#close(CloseOptions) API, and in fact would be how that > API will be implemented (since it only works for static groups at the moment > as noted in KAFKA-16514 ) > This has several benefits over the current situation: > # It allows plain consumer apps to opt-out of leaving the group when closed, > which is currently not possible through any public API (only an internal > backdoor config) > # It enables the caller to dynamically select the appropriate action > depending on why the client is being closed – for example, you would not want > the consumer to leave the group during a simple restart, but would want it to > leave the group when shutting down the app or if scaling down the node. This > is not possible today, even with the internal config, since configs are > immutable > # It can be leveraged to "fix" the KafkaStreams#close(closeOptions) API so > that the user's choice to leave the group during close will be respected for > non-static members -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16737) Clean up KafkaConsumerTest TODOs enabling tests for new consumer
[ https://issues.apache.org/jira/browse/KAFKA-16737?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16737: -- Component/s: clients > Clean up KafkaConsumerTest TODOs enabling tests for new consumer > > > Key: KAFKA-16737 > URL: https://issues.apache.org/jira/browse/KAFKA-16737 > Project: Kafka > Issue Type: Task > Components: clients, consumer >Affects Versions: 3.7.0 >Reporter: Lianet Magrans >Assignee: Lianet Magrans >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > > KafkaConsumerTest.java contains lots of TODOs (50+) related to tests that are > only enabled for the CLASSIC protocol and should be reviewed and enabled for > the new CONSUMER group protocol when applicable. Some tests also have TODOs > to enable them for the new consumer when certain features/bugs are addressed. > The new protocol and consumer implementation have evolved a lot since those > TODOs where added, so we should review them all, enable tests for the new > protocol when applicable and removing the TODOs from the code. Note that > there is another AsyncKafkaConsumerTest.java, testing logic specific to the > internals of the new consumer, but still many tests in the KafkaConsumerTest > apply to both the new and legacy consumer, and we should enable them for > both. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16637) AsyncKafkaConsumer removes offset fetch responses from cache too aggressively
[ https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16637: -- Summary: AsyncKafkaConsumer removes offset fetch responses from cache too aggressively (was: KIP-848 does not work well) > AsyncKafkaConsumer removes offset fetch responses from cache too aggressively > - > > Key: KAFKA-16637 > URL: https://issues.apache.org/jira/browse/KAFKA-16637 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: sanghyeok An >Assignee: Kirk True >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > Attachments: image-2024-04-30-08-33-06-367.png, > image-2024-04-30-08-33-50-435.png > > > I want to test next generation of the consumer rebalance protocol > ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)] > > However, it does not works well. > You can check my condition. > > *Docker-compose.yaml* > [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml] > > *Consumer Code* > [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java] > > *Consumer logs* > [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector > - initializing Kafka metrics collector > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: > 2ae524ed625438c5 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: > 1714309299215 > [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer - > [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1 > [consumer_background_thread] INFO org.apache.kafka.clients.Metadata - > [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk) > Stuck In here... > > *Broker logs* > broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > stuck in here -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16637) KIP-848 does not work well
[ https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16637: -- Priority: Blocker (was: Minor) > KIP-848 does not work well > -- > > Key: KAFKA-16637 > URL: https://issues.apache.org/jira/browse/KAFKA-16637 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: sanghyeok An >Assignee: Kirk True >Priority: Blocker > Labels: kip-848-client-support > Fix For: 3.8.0 > > Attachments: image-2024-04-30-08-33-06-367.png, > image-2024-04-30-08-33-50-435.png > > > I want to test next generation of the consumer rebalance protocol > ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)] > > However, it does not works well. > You can check my condition. > > *Docker-compose.yaml* > [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml] > > *Consumer Code* > [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java] > > *Consumer logs* > [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector > - initializing Kafka metrics collector > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: > 2ae524ed625438c5 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: > 1714309299215 > [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer - > [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1 > [consumer_background_thread] INFO org.apache.kafka.clients.Metadata - > [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk) > Stuck In here... > > *Broker logs* > broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > stuck in here -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (KAFKA-16427) KafkaConsumer#position() does not respect timeout when group protocol is CONSUMER
[
https://issues.apache.org/jira/browse/KAFKA-16427?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17842817#comment-17842817
]
Kirk True commented on KAFKA-16427:
---
This bug can be triggered with an embarrassingly simple integration test (n)
But it appears this bug can be fixed with an embarrassingly simple change, so
(y)
> KafkaConsumer#position() does not respect timeout when group protocol is
> CONSUMER
> -
>
> Key: KAFKA-16427
> URL: https://issues.apache.org/jira/browse/KAFKA-16427
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Affects Versions: 3.7.0
>Reporter: Alyssa Huang
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, timeout
> Fix For: 3.8.0
>
>
> When
> `long position(TopicPartition partition, final Duration timeout);`
> is called on an unknown topic partition (and auto creation is disabled), the
> method fails to adhere to the timeout supplied.
> e.g. the following warning is logged continuously as metadata fetches are
> retried
> [2024-03-26 11:03:48,589] WARN [Consumer clientId=ConsumerTestConsumer,
> groupId=my-test] Error while fetching metadata with correlation id 200 :
> \{nonexistingTopic=UNKNOWN_TOPIC_OR_PARTITION}
> (org.apache.kafka.clients.NetworkClient:1313)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16272) Update connect_distributed_test.py to support KIP-848’s group protocol config
[
https://issues.apache.org/jira/browse/KAFKA-16272?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16272:
--
Fix Version/s: 3.8.0
(was: 4.0.0)
> Update connect_distributed_test.py to support KIP-848’s group protocol config
> -
>
> Key: KAFKA-16272
> URL: https://issues.apache.org/jira/browse/KAFKA-16272
> Project: Kafka
> Issue Type: Test
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Sagar Rao
>Priority: Major
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> This task is to update the test method(s) in {{connect_distributed_test.py}}
> to support the {{group.protocol}} configuration introduced in
> [KIP-848|https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol]
> by adding an optional {{group_protocol}} argument to the tests and matrixes.
> See KAFKA-16231 as an example of how the test parameters can be changed.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16272) Update connect_distributed_test.py to support KIP-848’s group protocol config
[
https://issues.apache.org/jira/browse/KAFKA-16272?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16272:
--
Priority: Major (was: Blocker)
> Update connect_distributed_test.py to support KIP-848’s group protocol config
> -
>
> Key: KAFKA-16272
> URL: https://issues.apache.org/jira/browse/KAFKA-16272
> Project: Kafka
> Issue Type: Test
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Sagar Rao
>Priority: Major
> Labels: kip-848-client-support, system-tests
> Fix For: 4.0.0
>
>
> This task is to update the test method(s) in {{connect_distributed_test.py}}
> to support the {{group.protocol}} configuration introduced in
> [KIP-848|https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+The+Next+Generation+of+the+Consumer+Rebalance+Protocol]
> by adding an optional {{group_protocol}} argument to the tests and matrixes.
> See KAFKA-16231 as an example of how the test parameters can be changed.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-15561) Client support for new SubscriptionPattern based subscription
[ https://issues.apache.org/jira/browse/KAFKA-15561?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-15561: -- Fix Version/s: 3.8.0 (was: 4.0.0) > Client support for new SubscriptionPattern based subscription > - > > Key: KAFKA-15561 > URL: https://issues.apache.org/jira/browse/KAFKA-15561 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Phuc Hong Tran >Priority: Blocker > Labels: kip-848-client-support, regex > Fix For: 3.8.0 > > > New consumer should support subscribe with the new SubscriptionPattern > introduced in the new consumer group protocol. When subscribing with this > regex, the client should provide the regex in the HB request on the > SubscribedTopicRegex field, delegating the resolution to the server. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (KAFKA-16637) KIP-848 does not work well
[
https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17842530#comment-17842530
]
Kirk True commented on KAFKA-16637:
---
[~chickenchickenlove]—thanks for filing this. There are two existing
improvements (KAFKA-15974 and KAFKA-16200) that fix timing issues in the new
consumer. However, even when testing your case on a temporary branch that
includes fixes for both of those issues, the problem still showed up.
This issue is related to an optimization for offset fetch logic.
When a user calls {{Consumer.poll()}}, among other things, the consumer
performs a network request to fetch any previously-committed offsets so it can
determine from where to start fetching new records. When the user passes in a
timeout of zero, it's almost always the case that the offset fetch network
request will not be performed within 0 milliseconds. However, the consumer
still sends out the request and handles the response when it is received,
usually a few milliseconds later. In this first attempt, the lookup fails and
the {{poll()}} loops back around. Given that this timeout is the common case,
the consumer caches the offset fetch response/result from the first attempt
(even though it timed out) because it knows that the _next_ call to {{poll()}}
is going to attempt the exact same operation. When it is later attempted a
second time, the response is already there from the first attempt such that the
consumer doesn't need to perform a network request.
The existing consumer has implemented this caching in
[PendingCommittedOffsetRequest|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java#L132].
The new consumer has implemented it in
[CommitRequestManager|https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java#L510].
The core issue is the new consumer implementation is clearing out the first
attempt's cached result too aggressively. The effect being that the second (and
subsequent) attempts fail to find any previous attempt's cached result, and all
submit network requests, which all fail. Thus the consumer never makes any
headway.
> KIP-848 does not work well
> --
>
> Key: KAFKA-16637
> URL: https://issues.apache.org/jira/browse/KAFKA-16637
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Reporter: sanghyeok An
>Assignee: Kirk True
>Priority: Minor
> Labels: kip-848-client-support
> Fix For: 3.8.0
>
> Attachments: image-2024-04-30-08-33-06-367.png,
> image-2024-04-30-08-33-50-435.png
>
>
> I want to test next generation of the consumer rebalance protocol
> ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)]
>
> However, it does not works well.
> You can check my condition.
>
> *Docker-compose.yaml*
> [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml]
>
> *Consumer Code*
> [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java]
>
> *Consumer logs*
> [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector
> - initializing Kafka metrics collector
> [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0
> [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId:
> 2ae524ed625438c5
> [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs:
> 1714309299215
> [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer -
> [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1
> [consumer_background_thread] INFO org.apache.kafka.clients.Metadata -
> [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk)
> Stuck In here...
>
> *Broker logs*
> broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for
> Set(__consumer_offsets) to the active controller.
> (kafka.server.DefaultAutoTopicCreationManager)
> broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for
> Set(__consumer_offsets) to the active controller.
> (kafka.server.DefaultAutoTopicCreationManager)
> broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for
> Set(__consumer_offsets) to the active controller.
> (kafka.server.DefaultAutoTopicCreationManager)
> broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for
> Set(__consumer_offsets) to the active controller.
> (kafka.server.DefaultAutoTopicCreationManager)
> broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for
> Set(__consumer_offsets) to
[jira] [Updated] (KAFKA-16290) Investigate propagating subscription state updates via queues
[ https://issues.apache.org/jira/browse/KAFKA-16290?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16290: -- Priority: Major (was: Critical) > Investigate propagating subscription state updates via queues > - > > Key: KAFKA-16290 > URL: https://issues.apache.org/jira/browse/KAFKA-16290 > Project: Kafka > Issue Type: Task > Components: clients, consumer >Reporter: Lucas Brutschy >Assignee: Bruno Cadonna >Priority: Major > Labels: consumer-threading-refactor, kip-848, > kip-848-client-support > Fix For: 4.0.0 > > > We are mostly using the queues for interaction between application thread and > background thread, but the subscription object is shared between the threads, > and it is updated directly without going through the queues. > The way we allow updates to the subscription state from both threads is > definitely not right, and will bring trouble. Places like the assign() is > probably the most obvious, where we send an event to the background to > commit, but then update the subscription in the foreground right away. > It seems sensible to aim for having all updates to the subscription state in > the background, triggered from the app thread via events (and I think we > already have related events for all updates, just that the subscription state > was left out in the app thread). -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16110) Implement consumer performance tests
[ https://issues.apache.org/jira/browse/KAFKA-16110?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16110: -- Priority: Major (was: Blocker) > Implement consumer performance tests > > > Key: KAFKA-16110 > URL: https://issues.apache.org/jira/browse/KAFKA-16110 > Project: Kafka > Issue Type: New Feature > Components: clients, consumer >Reporter: Kirk True >Assignee: Philip Nee >Priority: Major > Labels: consumer-threading-refactor, performance-benchmark > Fix For: 4.0.0 > > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16557) Fix OffsetFetchRequestState.toString()
[
https://issues.apache.org/jira/browse/KAFKA-16557?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16557:
--
Priority: Minor (was: Major)
> Fix OffsetFetchRequestState.toString()
> --
>
> Key: KAFKA-16557
> URL: https://issues.apache.org/jira/browse/KAFKA-16557
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Minor
> Labels: consumer-threading-refactor, logging
> Fix For: 3.8.0
>
>
> The code incorrectly overrides the {{toString()}} method instead of
> overriding {{{}toStringBase(){}}}. This affects debugging and troubleshooting
> consumer issues.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16558) Implement HeartbeatRequestState.toStringBase()
[
https://issues.apache.org/jira/browse/KAFKA-16558?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16558:
--
Priority: Minor (was: Major)
> Implement HeartbeatRequestState.toStringBase()
> --
>
> Key: KAFKA-16558
> URL: https://issues.apache.org/jira/browse/KAFKA-16558
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Minor
> Labels: consumer-threading-refactor, logging
> Fix For: 3.8.0
>
>
> The inner class {{HeartbeatRequestState}} does not override the
> {{toStringBase()}} method. This affects debugging and troubleshooting
> consumer issues.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16623) KafkaAsyncConsumer system tests warn about revoking partitions that weren't previously assigned
[
https://issues.apache.org/jira/browse/KAFKA-16623?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16623:
--
Description:
When running system tests for the KafkaAsyncConsumer, we occasionally see this
warning:
{noformat}
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 38, in _protected_worker
self._worker(idx, node)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 304, in _worker
handler.handle_partitions_revoked(event, node, self.logger)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 163, in handle_partitions_revoked
(tp, node.account.hostname)
AssertionError: Topic partition TopicPartition(topic='test_topic', partition=0)
cannot be revoked from worker20 as it was not previously assigned to that
consumer
{noformat}
In test_fencing_static_consumer, there are two sets of consumers that use group
instance IDs: the initial set and the "conflict" set. It appears that one of
the "conflicting" consumers hijacks the partition ownership from the
coordinator's perspective when the initial consumer leaves the group.
was:
When running system tests for the KafkaAsyncConsumer, we occasionally see this
warning:
{noformat}
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 38, in _protected_worker
self._worker(idx, node)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 304, in _worker
handler.handle_partitions_revoked(event, node, self.logger)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 163, in handle_partitions_revoked
(tp, node.account.hostname)
AssertionError: Topic partition TopicPartition(topic='test_topic', partition=0)
cannot be revoked from worker20 as it was not previously assigned to that
consumer
{noformat}
In
> KafkaAsyncConsumer system tests warn about revoking partitions that weren't
> previously assigned
> ---
>
> Key: KAFKA-16623
> URL: https://issues.apache.org/jira/browse/KAFKA-16623
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, kip-848-client-support,
> system-tests
> Fix For: 3.8.0
>
>
> When running system tests for the KafkaAsyncConsumer, we occasionally see
> this warning:
> {noformat}
> File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
> self.run()
> File "/usr/lib/python3.7/threading.py", line 865, in run
> self._target(*self._args, **self._kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
> line 38, in _protected_worker
> self._worker(idx, node)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 304, in _worker
> handler.handle_partitions_revoked(event, node, self.logger)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 163, in handle_partitions_revoked
> (tp, node.account.hostname)
> AssertionError: Topic partition TopicPartition(topic='test_topic',
> partition=0) cannot be revoked from worker20 as it was not previously
> assigned to that consumer
> {noformat}
> In test_fencing_static_consumer, there are two sets of consumers that use
> group instance IDs: the initial set and the "conflict" set. It appears that
> one of the "conflicting" consumers hijacks the partition ownership from the
> coordinator's perspective when the initial consumer leaves the group.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16623) KafkaAsyncConsumer system tests warn about revoking partitions that weren't previously assigned
[
https://issues.apache.org/jira/browse/KAFKA-16623?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16623:
--
Description:
When running system tests for the KafkaAsyncConsumer, we occasionally see this
warning:
{noformat}
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 38, in _protected_worker
self._worker(idx, node)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 304, in _worker
handler.handle_partitions_revoked(event, node, self.logger)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 163, in handle_partitions_revoked
(tp, node.account.hostname)
AssertionError: Topic partition TopicPartition(topic='test_topic', partition=0)
cannot be revoked from worker20 as it was not previously assigned to that
consumer
{noformat}
In
was:
When running system tests for the KafkaAsyncConsumer, we occasionally see this
warning:
{noformat}
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 38, in _protected_worker
self._worker(idx, node)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 304, in _worker
handler.handle_partitions_revoked(event, node, self.logger)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 163, in handle_partitions_revoked
(tp, node.account.hostname)
AssertionError: Topic partition TopicPartition(topic='test_topic', partition=0)
cannot be revoked from worker20 as it was not previously assigned to that
consumer
{noformat}
It is unclear what is causing this.
> KafkaAsyncConsumer system tests warn about revoking partitions that weren't
> previously assigned
> ---
>
> Key: KAFKA-16623
> URL: https://issues.apache.org/jira/browse/KAFKA-16623
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, kip-848-client-support,
> system-tests
> Fix For: 3.8.0
>
>
> When running system tests for the KafkaAsyncConsumer, we occasionally see
> this warning:
> {noformat}
> File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
> self.run()
> File "/usr/lib/python3.7/threading.py", line 865, in run
> self._target(*self._args, **self._kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
> line 38, in _protected_worker
> self._worker(idx, node)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 304, in _worker
> handler.handle_partitions_revoked(event, node, self.logger)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 163, in handle_partitions_revoked
> (tp, node.account.hostname)
> AssertionError: Topic partition TopicPartition(topic='test_topic',
> partition=0) cannot be revoked from worker20 as it was not previously
> assigned to that consumer
> {noformat}
> In
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (KAFKA-16623) KafkaAsyncConsumer system tests warn about revoking partitions that weren't previously assigned
[
https://issues.apache.org/jira/browse/KAFKA-16623?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17842453#comment-17842453
]
Kirk True commented on KAFKA-16623:
---
If I add the following code to {{test_fencing_static_consumer}} before stopping
the consumer, the test runs:
{code:python}
# Make sure the conflicting consumers are all dead, but then go
ahead and stop them all to ensure
# that everything is cleanly stopped, otherwise we may get
spurious errors
assert len(conflict_consumer.dead_nodes()) ==
len(conflict_consumer.nodes), "Conflicting consumers should all have received
errors on startup and quit"
conflict_consumer.stop_all()
wait_until(lambda: len(conflict_consumer.dead_nodes()) ==
len(conflict_consumer.nodes),
timeout_sec=self.session_timeout_sec+5,
err_msg="Timed out waiting for the conflict consumer
to shutdown")
{code}
I'm still investigating, though, as I'm not sure if this is a "fix" or if it
"masks" a real issue.
> KafkaAsyncConsumer system tests warn about revoking partitions that weren't
> previously assigned
> ---
>
> Key: KAFKA-16623
> URL: https://issues.apache.org/jira/browse/KAFKA-16623
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, kip-848-client-support,
> system-tests
> Fix For: 3.8.0
>
>
> When running system tests for the KafkaAsyncConsumer, we occasionally see
> this warning:
> {noformat}
> File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
> self.run()
> File "/usr/lib/python3.7/threading.py", line 865, in run
> self._target(*self._args, **self._kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
> line 38, in _protected_worker
> self._worker(idx, node)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 304, in _worker
> handler.handle_partitions_revoked(event, node, self.logger)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 163, in handle_partitions_revoked
> (tp, node.account.hostname)
> AssertionError: Topic partition TopicPartition(topic='test_topic',
> partition=0) cannot be revoked from worker20 as it was not previously
> assigned to that consumer
> {noformat}
> It is unclear what is causing this.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16637) KIP-848 does not work well
[ https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16637: -- Fix Version/s: 3.8.0 > KIP-848 does not work well > -- > > Key: KAFKA-16637 > URL: https://issues.apache.org/jira/browse/KAFKA-16637 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: sanghyeok An >Assignee: Kirk True >Priority: Minor > Labels: kip-848-client-support > Fix For: 3.8.0 > > > I want to test next generation of the consumer rebalance protocol > ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)] > > However, it does not works well. > You can check my condition. > > *Docker-compose.yaml* > [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml] > > *Consumer Code* > [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java] > > *Consumer logs* > [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector > - initializing Kafka metrics collector > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: > 2ae524ed625438c5 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: > 1714309299215 > [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer - > [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1 > [consumer_background_thread] INFO org.apache.kafka.clients.Metadata - > [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk) > Stuck In here... > > *Broker logs* > broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > stuck in here -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16637) KIP-848 does not work well
[ https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16637: -- Component/s: clients consumer > KIP-848 does not work well > -- > > Key: KAFKA-16637 > URL: https://issues.apache.org/jira/browse/KAFKA-16637 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: sanghyeok An >Assignee: Kirk True >Priority: Minor > > I want to test next generation of the consumer rebalance protocol > ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)] > > However, it does not works well. > You can check my condition. > > *Docker-compose.yaml* > [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml] > > *Consumer Code* > [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java] > > *Consumer logs* > [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector > - initializing Kafka metrics collector > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: > 2ae524ed625438c5 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: > 1714309299215 > [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer - > [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1 > [consumer_background_thread] INFO org.apache.kafka.clients.Metadata - > [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk) > Stuck In here... > > *Broker logs* > broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > stuck in here -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16637) KIP-848 does not work well
[ https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16637: -- Labels: kip-848-client-support (was: ) > KIP-848 does not work well > -- > > Key: KAFKA-16637 > URL: https://issues.apache.org/jira/browse/KAFKA-16637 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: sanghyeok An >Assignee: Kirk True >Priority: Minor > Labels: kip-848-client-support > > I want to test next generation of the consumer rebalance protocol > ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)] > > However, it does not works well. > You can check my condition. > > *Docker-compose.yaml* > [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml] > > *Consumer Code* > [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java] > > *Consumer logs* > [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector > - initializing Kafka metrics collector > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: > 2ae524ed625438c5 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: > 1714309299215 > [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer - > [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1 > [consumer_background_thread] INFO org.apache.kafka.clients.Metadata - > [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk) > Stuck In here... > > *Broker logs* > broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > stuck in here -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (KAFKA-16637) KIP-848 does not work well
[ https://issues.apache.org/jira/browse/KAFKA-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True reassigned KAFKA-16637: - Assignee: Kirk True > KIP-848 does not work well > -- > > Key: KAFKA-16637 > URL: https://issues.apache.org/jira/browse/KAFKA-16637 > Project: Kafka > Issue Type: Bug >Reporter: sanghyeok An >Assignee: Kirk True >Priority: Minor > > I want to test next generation of the consumer rebalance protocol > ([https://cwiki.apache.org/confluence/display/KAFKA/The+Next+Generation+of+the+Consumer+Rebalance+Protocol+%28KIP-848%29+-+Early+Access+Release+Notes)] > > However, it does not works well. > You can check my condition. > > *Docker-compose.yaml* > [https://github.com/chickenchickenlove/kraft-test/blob/main/docker-compose/docker-compose.yaml] > > *Consumer Code* > [https://github.com/chickenchickenlove/kraft-test/blob/main/src/main/java/org/example/Main.java] > > *Consumer logs* > [main] INFO org.apache.kafka.common.telemetry.internals.KafkaMetricsCollector > - initializing Kafka metrics collector > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 3.7.0 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: > 2ae524ed625438c5 > [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: > 1714309299215 > [main] INFO org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer - > [Consumer clientId=1-1, groupId=1] Subscribed to topic(s): test-topic1 > [consumer_background_thread] INFO org.apache.kafka.clients.Metadata - > [Consumer clientId=1-1, groupId=1] Cluster ID: Some(MkU3OEVBNTcwNTJENDM2Qk) > Stuck In here... > > *Broker logs* > broker | [2024-04-28 12:42:27,751] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:27,801] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,211] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,259] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > broker | [2024-04-28 12:42:28,727] INFO Sent auto-creation request for > Set(__consumer_offsets) to the active controller. > (kafka.server.DefaultAutoTopicCreationManager) > stuck in here -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (KAFKA-16642) Update KafkaConsumerTest to show parameters in test lists
Kirk True created KAFKA-16642:
-
Summary: Update KafkaConsumerTest to show parameters in test lists
Key: KAFKA-16642
URL: https://issues.apache.org/jira/browse/KAFKA-16642
Project: Kafka
Issue Type: Bug
Components: clients, consumer, unit tests
Affects Versions: 3.7.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
{{KafkaConsumerTest}} was recently updated to make many of its tests
parameterized to exercise both the {{CLASSIC}} and {{CONSUMER}} group
protocols. However, in some of the tools in which [lists of tests are
provided|https://ge.apache.org/scans/tests?search.names=Git%20branch=P28D=kafka=America%2FLos_Angeles=trunk=org.apache.kafka.clients.consumer.KafkaConsumerTest=FLAKY],
say, for analysis, the group protocol information is not exposed. For example,
one test ({{{}testReturnRecordsDuringRebalance{}}}) is flaky, but it's
difficult to know at a glance which group protocol is causing the problem
because the list simply shows:
{quote}{{testReturnRecordsDuringRebalance(GroupProtocol)[1]}}
{quote}
Ideally, it would expose more information, such as:
{quote}{{testReturnRecordsDuringRebalance(GroupProtocol=CONSUMER)}}
{quote}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16642) Update KafkaConsumerTest to show parameters in test lists
[
https://issues.apache.org/jira/browse/KAFKA-16642?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16642:
--
Priority: Minor (was: Major)
> Update KafkaConsumerTest to show parameters in test lists
> -
>
> Key: KAFKA-16642
> URL: https://issues.apache.org/jira/browse/KAFKA-16642
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, unit tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Minor
> Labels: consumer-threading-refactor
> Fix For: 3.8.0
>
>
> {{KafkaConsumerTest}} was recently updated to make many of its tests
> parameterized to exercise both the {{CLASSIC}} and {{CONSUMER}} group
> protocols. However, in some of the tools in which [lists of tests are
> provided|https://ge.apache.org/scans/tests?search.names=Git%20branch=P28D=kafka=America%2FLos_Angeles=trunk=org.apache.kafka.clients.consumer.KafkaConsumerTest=FLAKY],
> say, for analysis, the group protocol information is not exposed. For
> example, one test ({{{}testReturnRecordsDuringRebalance{}}}) is flaky, but
> it's difficult to know at a glance which group protocol is causing the
> problem because the list simply shows:
> {quote}{{testReturnRecordsDuringRebalance(GroupProtocol)[1]}}
> {quote}
> Ideally, it would expose more information, such as:
> {quote}{{testReturnRecordsDuringRebalance(GroupProtocol=CONSUMER)}}
> {quote}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16576) New consumer fails with assert in consumer_test.py’s test_consumer_failure system test
[
https://issues.apache.org/jira/browse/KAFKA-16576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16576:
--
Priority: Minor (was: Blocker)
> New consumer fails with assert in consumer_test.py’s test_consumer_failure
> system test
> --
>
> Key: KAFKA-16576
> URL: https://issues.apache.org/jira/browse/KAFKA-16576
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Minor
> Labels: flaky-test, kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 42.582 seconds
> AssertionError()
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 399, in test_consumer_failure
> assert partition_owner is not None
> AssertionError
> Notify
> {code}
> Affected tests:
> * {{test_consumer_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Assigned] (KAFKA-16576) New consumer fails with assert in consumer_test.py’s test_consumer_failure system test
[
https://issues.apache.org/jira/browse/KAFKA-16576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True reassigned KAFKA-16576:
-
Assignee: (was: Kirk True)
> New consumer fails with assert in consumer_test.py’s test_consumer_failure
> system test
> --
>
> Key: KAFKA-16576
> URL: https://issues.apache.org/jira/browse/KAFKA-16576
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Priority: Minor
> Labels: flaky-test, kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 42.582 seconds
> AssertionError()
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 399, in test_consumer_failure
> assert partition_owner is not None
> AssertionError
> Notify
> {code}
> Affected tests:
> * {{test_consumer_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Resolved] (KAFKA-16460) New consumer times out consuming records in consumer_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16460?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True resolved KAFKA-16460.
---
Resolution: Duplicate
> New consumer times out consuming records in consumer_test.py system test
>
>
> Key: KAFKA-16460
> URL: https://issues.apache.org/jira/browse/KAFKA-16460
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test fails with the following errors:
> {quote}
> * Timed out waiting for consumption
> {quote}
> Affected tests:
> * {{test_broker_failure}}
> * {{test_consumer_bounce}}
> * {{test_static_consumer_bounce}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16639) AsyncKafkaConsumer#close does not send heartbeat to leave group
[ https://issues.apache.org/jira/browse/KAFKA-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16639: -- Component/s: clients consumer > AsyncKafkaConsumer#close does not send heartbeat to leave group > --- > > Key: KAFKA-16639 > URL: https://issues.apache.org/jira/browse/KAFKA-16639 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Reporter: Chia-Ping Tsai >Assignee: Chia-Ping Tsai >Priority: Major > Labels: kip-848-client-support > > This bug can be reproduced by immediately closing a consumer which is just > created. > The root cause is that we skip the new heartbeat used to leave group when > there is a in-flight heartbeat request > ([https://github.com/apache/kafka/blob/5de5d967adffd864bad3ec729760a430253abf38/clients/src/main/java/org/apache/kafka/clients/consumer/internals/HeartbeatRequestManager.java#L212).] > It seems to me the simple solution is that we create a heartbeat request when > meeting above situation and then send it by pollOnClose > ([https://github.com/apache/kafka/blob/5de5d967adffd864bad3ec729760a430253abf38/clients/src/main/java/org/apache/kafka/clients/consumer/internals/RequestManager.java#L62).] > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Resolved] (KAFKA-16609) Update parse_describe_topic to support new topic describe output
[
https://issues.apache.org/jira/browse/KAFKA-16609?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True resolved KAFKA-16609.
---
Reviewer: Lucas Brutschy
Resolution: Fixed
> Update parse_describe_topic to support new topic describe output
>
>
> Key: KAFKA-16609
> URL: https://issues.apache.org/jira/browse/KAFKA-16609
> Project: Kafka
> Issue Type: Bug
> Components: admin, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Major
> Labels: system-test-failure
> Fix For: 3.8.0
>
>
> It appears that recent changes to the describe topic output has broken the
> system test's ability to parse the output.
> {noformat}
> test_id:
> kafkatest.tests.core.reassign_partitions_test.ReassignPartitionsTest.test_reassign_partitions.bounce_brokers=False.reassign_from_offset_zero=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 50.333 seconds
> IndexError('list index out of range')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/reassign_partitions_test.py",
> line 175, in test_reassign_partitions
> self.run_produce_consume_validate(core_test_action=lambda:
> self.reassign_partitions(bounce_brokers))
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/produce_consume_validate.py",
> line 105, in run_produce_consume_validate
> core_test_action(*args)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/reassign_partitions_test.py",
> line 175, in
> self.run_produce_consume_validate(core_test_action=lambda:
> self.reassign_partitions(bounce_brokers))
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/reassign_partitions_test.py",
> line 82, in reassign_partitions
> partition_info =
> self.kafka.parse_describe_topic(self.kafka.describe_topic(self.topic))
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/kafka/kafka.py",
> line 1400, in parse_describe_topic
> fields = list(map(lambda x: x.split(" ")[1], fields))
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/kafka/kafka.py",
> line 1400, in
> fields = list(map(lambda x: x.split(" ")[1], fields))
> IndexError: list index out of range
> {noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16623) KafkaAsyncConsumer system tests warn about revoking partitions that weren't previously assigned
Kirk True created KAFKA-16623:
-
Summary: KafkaAsyncConsumer system tests warn about revoking
partitions that weren't previously assigned
Key: KAFKA-16623
URL: https://issues.apache.org/jira/browse/KAFKA-16623
Project: Kafka
Issue Type: Bug
Components: clients, consumer, system tests
Affects Versions: 3.8.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
When running system tests for the KafkaAsyncConsumer, we occasionally see this
warning:
{noformat}
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 38, in _protected_worker
self._worker(idx, node)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 304, in _worker
handler.handle_partitions_revoked(event, node, self.logger)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 163, in handle_partitions_revoked
(tp, node.account.hostname)
AssertionError: Topic partition TopicPartition(topic='test_topic', partition=0)
cannot be revoked from worker20 as it was not previously assigned to that
consumer
{noformat}
It is unclear what is causing this.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16608) AsyncKafkaConsumer doesn't honor interrupted thread status on KafkaConsumer.poll(Duration)
[ https://issues.apache.org/jira/browse/KAFKA-16608?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16608: -- Summary: AsyncKafkaConsumer doesn't honor interrupted thread status on KafkaConsumer.poll(Duration) (was: AsyncKafkaConsumer doesn't honour interrupted thread status on KafkaConsumer.poll(Duration)) > AsyncKafkaConsumer doesn't honor interrupted thread status on > KafkaConsumer.poll(Duration) > -- > > Key: KAFKA-16608 > URL: https://issues.apache.org/jira/browse/KAFKA-16608 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Affects Versions: 3.8.0 >Reporter: Andrew Schofield >Assignee: Andrew Schofield >Priority: Minor > > The behaviour for KafkaConsumer.poll(Duration) when the calling thread is in > interrupted state is to throw InterruptException. The AsyncKafkaConsumer > doesn't do this. It only throws that exception if the interruption occurs > while it is waiting. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (KAFKA-16609) Update parse_describe_topic to support new topic describe output
Kirk True created KAFKA-16609:
-
Summary: Update parse_describe_topic to support new topic describe
output
Key: KAFKA-16609
URL: https://issues.apache.org/jira/browse/KAFKA-16609
Project: Kafka
Issue Type: Bug
Components: admin, system tests
Affects Versions: 3.8.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
It appears that recent changes to the describe topic output has broken the
system test's ability to parse the output.
{noformat}
test_id:
kafkatest.tests.core.reassign_partitions_test.ReassignPartitionsTest.test_reassign_partitions.bounce_brokers=False.reassign_from_offset_zero=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 50.333 seconds
IndexError('list index out of range')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/reassign_partitions_test.py",
line 175, in test_reassign_partitions
self.run_produce_consume_validate(core_test_action=lambda:
self.reassign_partitions(bounce_brokers))
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/produce_consume_validate.py",
line 105, in run_produce_consume_validate
core_test_action(*args)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/reassign_partitions_test.py",
line 175, in
self.run_produce_consume_validate(core_test_action=lambda:
self.reassign_partitions(bounce_brokers))
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/reassign_partitions_test.py",
line 82, in reassign_partitions
partition_info =
self.kafka.parse_describe_topic(self.kafka.describe_topic(self.topic))
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/kafka/kafka.py",
line 1400, in parse_describe_topic
fields = list(map(lambda x: x.split(" ")[1], fields))
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/kafka/kafka.py",
line 1400, in
fields = list(map(lambda x: x.split(" ")[1], fields))
IndexError: list index out of range
{noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16608) AsyncKafkaConsumer doesn't honour interrupted thread status on KafkaConsumer.poll(Duration)
[ https://issues.apache.org/jira/browse/KAFKA-16608?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16608: -- Component/s: consumer > AsyncKafkaConsumer doesn't honour interrupted thread status on > KafkaConsumer.poll(Duration) > --- > > Key: KAFKA-16608 > URL: https://issues.apache.org/jira/browse/KAFKA-16608 > Project: Kafka > Issue Type: Bug > Components: clients, consumer >Affects Versions: 3.8.0 >Reporter: Andrew Schofield >Priority: Minor > > The behaviour for KafkaConsumer.poll(Duration) when the calling thread is in > interrupted state is to throw InterruptException. The AsyncKafkaConsumer > doesn't do this. It only throws that exception if the interruption occurs > while it is waiting. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (KAFKA-16217) Transactional producer stuck in IllegalStateException during close
[
https://issues.apache.org/jira/browse/KAFKA-16217?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17840191#comment-17840191
]
Kirk True commented on KAFKA-16217:
---
[~calvinliu]—I noticed the PR request 15541 is merged, but the status of this
still "In Progress." The bug lists the fix versions as 3.8.0, 3.7.1, and 3.6.3.
Was the change in PR 15541 back-ported to those earlier versions?
Thanks!
> Transactional producer stuck in IllegalStateException during close
> --
>
> Key: KAFKA-16217
> URL: https://issues.apache.org/jira/browse/KAFKA-16217
> Project: Kafka
> Issue Type: Bug
> Components: clients, producer
>Affects Versions: 3.7.0, 3.6.1
>Reporter: Calvin Liu
>Assignee: Calvin Liu
>Priority: Major
> Labels: transactions
> Fix For: 3.8.0, 3.7.1, 3.6.3
>
>
> The producer is stuck during the close. It keeps retrying to abort the
> transaction but it never succeeds.
> {code:java}
> [ERROR] 2024-02-01 17:21:22,804 [kafka-producer-network-thread |
> producer-transaction-bench-transaction-id-f60SGdyRQGGFjdgg3vUgKg]
> org.apache.kafka.clients.producer.internals.Sender run - [Producer
> clientId=producer-transaction-ben
> ch-transaction-id-f60SGdyRQGGFjdgg3vUgKg,
> transactionalId=transaction-bench-transaction-id-f60SGdyRQGGFjdgg3vUgKg]
> Error in kafka producer I/O thread while aborting transaction:
> java.lang.IllegalStateException: Cannot attempt operation `abortTransaction`
> because the previous call to `commitTransaction` timed out and must be retried
> at
> org.apache.kafka.clients.producer.internals.TransactionManager.handleCachedTransactionRequestResult(TransactionManager.java:1138)
> at
> org.apache.kafka.clients.producer.internals.TransactionManager.beginAbort(TransactionManager.java:323)
> at
> org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:274)
> at java.base/java.lang.Thread.run(Thread.java:1583)
> at org.apache.kafka.common.utils.KafkaThread.run(KafkaThread.java:66)
> {code}
> With the additional log, I found the root cause. If the producer is in a bad
> transaction state(in my case, the TransactionManager.pendingTransition was
> set to commitTransaction and did not get cleaned), then the producer calls
> close and tries to abort the existing transaction, the producer will get
> stuck in the transaction abortion. It is related to the fix
> [https://github.com/apache/kafka/pull/13591].
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Resolved] (KAFKA-16462) New consumer fails with timeout in security_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16462?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True resolved KAFKA-16462.
---
Resolution: Duplicate
> New consumer fails with timeout in security_test.py system test
> ---
>
> Key: KAFKA-16462
> URL: https://issues.apache.org/jira/browse/KAFKA-16462
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{security_test.py}} system test fails with the following error:
> {noformat}
> test_id:
> kafkatest.tests.core.security_test.SecurityTest.test_client_ssl_endpoint_validation_failure.security_protocol=SSL.interbroker_security_protocol=PLAINTEXT.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 1 minute 30.885 seconds
> TimeoutError('')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/security_test.py",
> line 142, in test_client_ssl_endpoint_validation_failure
> wait_until(lambda: self.producer_consumer_have_expected_error(error),
> timeout_sec=30)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/utils/util.py",
> line 58, in wait_until
> raise TimeoutError(err_msg() if callable(err_msg) else err_msg) from
> last_exception
> ducktape.errors.TimeoutError
> {noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16461) New consumer fails to consume records in security_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16461?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16461:
--
Description:
The {{security_test.py}} system test fails with the following error:
{quote}
* Consumer failed to consume up to offsets
{quote}
Affected test:
* {{test_client_ssl_endpoint_validation_failure}}
Cause:
The system test was failing because the {{VerifiableConsumer}} hit a
{{NullPointerException}} during startup. The reason for the NPE was an attempt
to put a {{null}} as the value of {{--group-remote-assignor}} in the
{{Consumer}}'s configuration.
was:
The {{security_test.py}} system test fails with the following error:
{quote}
* Consumer failed to consume up to offsets
{quote}
Affected test:
* {{test_client_ssl_endpoint_validation_failure}}
> New consumer fails to consume records in security_test.py system test
> -
>
> Key: KAFKA-16461
> URL: https://issues.apache.org/jira/browse/KAFKA-16461
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{security_test.py}} system test fails with the following error:
> {quote}
> * Consumer failed to consume up to offsets
> {quote}
> Affected test:
> * {{test_client_ssl_endpoint_validation_failure}}
> Cause:
> The system test was failing because the {{VerifiableConsumer}} hit a
> {{NullPointerException}} during startup. The reason for the NPE was an
> attempt to put a {{null}} as the value of {{--group-remote-assignor}} in the
> {{Consumer}}'s configuration.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Resolved] (KAFKA-16464) New consumer fails with timeout in replication_replica_failure_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16464?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True resolved KAFKA-16464.
---
Resolution: Duplicate
> New consumer fails with timeout in replication_replica_failure_test.py system
> test
> --
>
> Key: KAFKA-16464
> URL: https://issues.apache.org/jira/browse/KAFKA-16464
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{replication_replica_failure_test.py}} system test fails with the
> following error:
> {noformat}
> test_id:
> kafkatest.tests.core.replication_replica_failure_test.ReplicationReplicaFailureTest.test_replication_with_replica_failure.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 1 minute 20.972 seconds
> TimeoutError('Timed out after 30s while awaiting initial record delivery
> of 5 records')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/core/replication_replica_failure_test.py",
> line 97, in test_replication_with_replica_failure
> self.await_startup()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/end_to_end.py",
> line 125, in await_startup
> (timeout_sec, min_records))
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/utils/util.py",
> line 58, in wait_until
> raise TimeoutError(err_msg() if callable(err_msg) else err_msg) from
> last_exception
> ducktape.errors.TimeoutError: Timed out after 30s while awaiting initial
> record delivery of 5 records
> {noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16565) IncrementalAssignmentConsumerEventHandler throws error when attempting to remove a partition that isn't assigned
[
https://issues.apache.org/jira/browse/KAFKA-16565?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16565:
--
Reviewer: Lucas Brutschy
Description:
In {{verifiable_consumer.py}}, the
{{IncrementalAssignmentConsumerEventHandler}} contains this logic:
{code:python}
def handle_partitions_revoked(self, event):
self.revoked_count += 1
self.state = ConsumerState.Rebalancing
self.position = {}
for topic_partition in event["partitions"]:
topic = topic_partition["topic"]
partition = topic_partition["partition"]
self.assignment.remove(TopicPartition(topic, partition))
{code}
If the {{self.assignment.remove()}} call is passed a {{TopicPartition}} that
isn't in the list, a generic Python list error is thrown. For now, we should
first check that the {{TopicPartition}} is in the list with an {{assert}} that
provides better information .
was:
In {{{}verifiable_consumer.py{}}}, the Incremental
{code:java}
def handle_partitions_revoked(self, event):
self.revoked_count += 1
self.state = ConsumerState.Rebalancing
self.position = {}
for topic_partition in event["partitions"]:
topic = topic_partition["topic"]
partition = topic_partition["partition"]
self.assignment.remove(TopicPartition(topic, partition))
{code}
If the {{self.assignment.remove()}} call is passed a {{TopicPartition}} that
isn't in the list, an error is thrown. For now, we should first check that the
{{TopicPartition}} is in the list, and if not, log a warning or something.
> IncrementalAssignmentConsumerEventHandler throws error when attempting to
> remove a partition that isn't assigned
>
>
> Key: KAFKA-16565
> URL: https://issues.apache.org/jira/browse/KAFKA-16565
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, kip-848-client-support,
> system-tests
> Fix For: 3.8.0
>
>
> In {{verifiable_consumer.py}}, the
> {{IncrementalAssignmentConsumerEventHandler}} contains this logic:
> {code:python}
> def handle_partitions_revoked(self, event):
> self.revoked_count += 1
> self.state = ConsumerState.Rebalancing
> self.position = {}
> for topic_partition in event["partitions"]:
> topic = topic_partition["topic"]
> partition = topic_partition["partition"]
> self.assignment.remove(TopicPartition(topic, partition))
> {code}
> If the {{self.assignment.remove()}} call is passed a {{TopicPartition}} that
> isn't in the list, a generic Python list error is thrown. For now, we should
> first check that the {{TopicPartition}} is in the list with an {{assert}}
> that provides better information .
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16461) New consumer fails to consume records in security_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16461?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16461:
--
Reviewer: Lucas Brutschy
> New consumer fails to consume records in security_test.py system test
> -
>
> Key: KAFKA-16461
> URL: https://issues.apache.org/jira/browse/KAFKA-16461
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{security_test.py}} system test fails with the following error:
> {quote}
> * Consumer failed to consume up to offsets
> {quote}
> Affected test:
> * {{test_client_ssl_endpoint_validation_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16103) AsyncKafkaConsumer: Always await async commit callbacks in commitSync and close
[ https://issues.apache.org/jira/browse/KAFKA-16103?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kirk True updated KAFKA-16103: -- Summary: AsyncKafkaConsumer: Always await async commit callbacks in commitSync and close (was: Review client logic for triggering offset commit callbacks) > AsyncKafkaConsumer: Always await async commit callbacks in commitSync and > close > --- > > Key: KAFKA-16103 > URL: https://issues.apache.org/jira/browse/KAFKA-16103 > Project: Kafka > Issue Type: Sub-task > Components: clients, consumer >Reporter: Lianet Magrans >Assignee: Lucas Brutschy >Priority: Critical > Labels: kip-848-client-support, offset > Fix For: 3.8.0 > > > Review logic for triggering commit callbacks, ensuring that all callbacks are > triggered before returning from commitSync -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (KAFKA-16577) New consumer fails with stop within allotted timeout in consumer_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16577?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16577:
--
Description:
The {{consumer_test.py}} system test intermittently fails with the following
error:
{code}
test_id:
kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 2 minutes 35.925 seconds
AssertionError('Node ubuntu@worker5: did not stop within the specified
timeout of 30 seconds')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 176, in test_consumer_bounce
self.rolling_bounce_consumers(consumer, clean_shutdown=clean_shutdown)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 39, in rolling_bounce_consumers
consumer.stop_node(node, clean_shutdown)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 419, in stop_node
(str(node.account), str(self.stop_timeout_sec))
AssertionError: Node ubuntu@worker5: did not stop within the specified timeout
of 30 seconds
{code}
Affected tests:
* {{test_broker_failure}}
* {{test_consumer_bounce}}
was:
The {{consumer_test.py}} system test intermittently fails with the following
error:
{code}
test_id:
kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 2 minutes 35.925 seconds
AssertionError('Node ubuntu@worker5: did not stop within the specified
timeout of 30 seconds')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 176, in test_consumer_bounce
self.rolling_bounce_consumers(consumer, clean_shutdown=clean_shutdown)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 39, in rolling_bounce_consumers
consumer.stop_node(node, clean_shutdown)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 419, in stop_node
(str(node.account), str(self.stop_timeout_sec))
AssertionError: Node ubuntu@worker5: did not stop within the specified timeout
of 30 seconds
{code}
Affected tests:
* {{test_consumer_bounce}}
> New consumer fails with stop within allotted timeout in consumer_test.py
> system test
>
>
> Key: KAFKA-16577
> URL: https://issues.apache.org/jira/browse/KAFKA-16577
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: flaky-test, kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
>
[jira] [Updated] (KAFKA-16577) New consumer fails with stop within allotted timeout in consumer_test.py system test
[
https://issues.apache.org/jira/browse/KAFKA-16577?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16577:
--
Summary: New consumer fails with stop within allotted timeout in
consumer_test.py system test (was: New consumer fails with stop within
allotted timeout in consumer_test.py’s test_consumer_bounce system test)
> New consumer fails with stop within allotted timeout in consumer_test.py
> system test
>
>
> Key: KAFKA-16577
> URL: https://issues.apache.org/jira/browse/KAFKA-16577
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: flaky-test, kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 2 minutes 35.925 seconds
> AssertionError('Node ubuntu@worker5: did not stop within the specified
> timeout of 30 seconds')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 176, in test_consumer_bounce
> self.rolling_bounce_consumers(consumer, clean_shutdown=clean_shutdown)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 39, in rolling_bounce_consumers
> consumer.stop_node(node, clean_shutdown)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 419, in stop_node
> (str(node.account), str(self.stop_timeout_sec))
> AssertionError: Node ubuntu@worker5: did not stop within the specified
> timeout of 30 seconds
> {code}
> Affected tests:
> * {{test_consumer_bounce}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16579) Revert changes to consumer_rolling_upgrade_test.py for the new async Consumer
[
https://issues.apache.org/jira/browse/KAFKA-16579?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16579:
--
Description:
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16271 updated the system tests in {{consumer_rolling_upgrade_test.py}} so
it could test the new consumer. However, the test is tailored specifically to
the "old" Consumer's protocol and assignment strategy upgrade.
Unsurprisingly, when we run those system tests with the new
{{AsyncKafkaConsumer}}, we get errors like the following:
{code}
test_id:
kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 29.634 seconds
AssertionError("Mismatched assignment: {frozenset(),
frozenset({TopicPartition(topic='test_topic', partition=0),
TopicPartition(topic='test_topic', partition=1)})}")
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_rolling_upgrade_test.py",
line 77, in rolling_update_test
self._verify_range_assignment(consumer)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_rolling_upgrade_test.py",
line 41, in _verify_range_assignment
"Mismatched assignment: %s" % assignment
AssertionError: Mismatched assignment: {frozenset(),
frozenset({TopicPartition(topic='test_topic', partition=0),
TopicPartition(topic='test_topic', partition=1)})}
{code}
The task here is to revert the changes made in KAFKA-16271.
was:
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16271 updated the system tests in {{consumer_rolling_upgrade_test.py}} so
it could test the new consumer. However, the test is tailored specifically to
the "old" Consumer's protocol and assignment strategy upgrade.
Unsurprisingly, when we run those system tests with the new
{{AsyncKafkaConsumer}}, we get errors like the following:
{code:java}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
line 919, in test_exactly_once_source
consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
self.source.topic, consumer_timeout_ms=1000, print_key=True)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
line 97, in __init__
BackgroundThreadService.__init__(self, context, num_nodes)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 26, in __init__
super(BackgroundThreadService, self).__init__(context, num_nodes,
cluster_spec, *args, **kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 107, in __init__
self.allocate_nodes()
File
[jira] [Updated] (KAFKA-16579) Revert changes to consumer_rolling_upgrade_test.py for the new async Consumer
[
https://issues.apache.org/jira/browse/KAFKA-16579?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16579:
--
Description:
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16271 updated the system tests in {{consumer_rolling_upgrade_test.py}} so
it could test the new consumer. However, the test is tailored specifically to
the "old" Consumer's protocol and assignment strategy upgrade.
Unsurprisingly, when we run those system tests with the new
{{AsyncKafkaConsumer}}, we get errors like the following:
{code:java}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
line 919, in test_exactly_once_source
consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
self.source.topic, consumer_timeout_ms=1000, print_key=True)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
line 97, in __init__
BackgroundThreadService.__init__(self, context, num_nodes)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 26, in __init__
super(BackgroundThreadService, self).__init__(context, num_nodes,
cluster_spec, *args, **kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 107, in __init__
self.allocate_nodes()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 217, in allocate_nodes
self.nodes = self.cluster.alloc(self.cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/cluster.py",
line 54, in alloc
allocated = self.do_alloc(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/finite_subcluster.py",
line 31, in do_alloc
allocated = self._available_nodes.remove_spec(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/node_container.py",
line 117, in remove_spec
raise InsufficientResourcesError("Not enough nodes available to allocate. "
+ msg)
ducktape.cluster.node_container.InsufficientResourcesError: Not enough nodes
available to allocate. linux nodes requested: 1. linux nodes available: 0
{code}
The task here is to revert the changes made in KAFKA-16272 [PR
15576|https://github.com/apache/kafka/pull/15576].
was:
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
could test the new consumer with Connect. However, we are not supporting
Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the Connect
system tests with the new {{{}AsyncKafkaConsumer{}}}, we get errors like the
following:
{code:java}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
[jira] [Updated] (KAFKA-16578) Revert changes to connect_distributed_test.py for the new async Consumer
[
https://issues.apache.org/jira/browse/KAFKA-16578?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16578:
--
Description:
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
could test the new consumer with Connect. However, we are not supporting
Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the Connect
system tests with the new {{AsyncKafkaConsumer}}, we get errors like the
following:
{code}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
line 919, in test_exactly_once_source
consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
self.source.topic, consumer_timeout_ms=1000, print_key=True)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
line 97, in __init__
BackgroundThreadService.__init__(self, context, num_nodes)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 26, in __init__
super(BackgroundThreadService, self).__init__(context, num_nodes,
cluster_spec, *args, **kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 107, in __init__
self.allocate_nodes()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 217, in allocate_nodes
self.nodes = self.cluster.alloc(self.cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/cluster.py",
line 54, in alloc
allocated = self.do_alloc(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/finite_subcluster.py",
line 31, in do_alloc
allocated = self._available_nodes.remove_spec(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/node_container.py",
line 117, in remove_spec
raise InsufficientResourcesError("Not enough nodes available to allocate. "
+ msg)
ducktape.cluster.node_container.InsufficientResourcesError: Not enough nodes
available to allocate. linux nodes requested: 1. linux nodes available: 0
{code}
The task here is to revert the changes made in KAFKA-16272 [PR
15576|https://github.com/apache/kafka/pull/15576].
was:
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
could test the new consumer with Connect. However, we are not supporting
Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the Connect
system tests with the new {{{}AsyncKafkaConsumer{}}}, we get errors like the
following:
{code:java}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
[jira] [Created] (KAFKA-16579) Revert changes to consumer_rolling_upgrade_test.py for the new async Consumer
Kirk True created KAFKA-16579:
-
Summary: Revert changes to consumer_rolling_upgrade_test.py for
the new async Consumer
Key: KAFKA-16579
URL: https://issues.apache.org/jira/browse/KAFKA-16579
Project: Kafka
Issue Type: Task
Components: clients, consumer, system tests
Affects Versions: 3.8.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
could test the new consumer with Connect. However, we are not supporting
Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the Connect
system tests with the new {{{}AsyncKafkaConsumer{}}}, we get errors like the
following:
{code:java}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
line 919, in test_exactly_once_source
consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
self.source.topic, consumer_timeout_ms=1000, print_key=True)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
line 97, in __init__
BackgroundThreadService.__init__(self, context, num_nodes)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 26, in __init__
super(BackgroundThreadService, self).__init__(context, num_nodes,
cluster_spec, *args, **kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 107, in __init__
self.allocate_nodes()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 217, in allocate_nodes
self.nodes = self.cluster.alloc(self.cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/cluster.py",
line 54, in alloc
allocated = self.do_alloc(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/finite_subcluster.py",
line 31, in do_alloc
allocated = self._available_nodes.remove_spec(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/node_container.py",
line 117, in remove_spec
raise InsufficientResourcesError("Not enough nodes available to allocate. "
+ msg)
ducktape.cluster.node_container.InsufficientResourcesError: Not enough nodes
available to allocate. linux nodes requested: 1. linux nodes available: 0
{code}
The task here is to revert the changes made in KAFKA-16272 [PR
15576|https://github.com/apache/kafka/pull/15576].
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Assigned] (KAFKA-16579) Revert changes to consumer_rolling_upgrade_test.py for the new async Consumer
[
https://issues.apache.org/jira/browse/KAFKA-16579?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True reassigned KAFKA-16579:
-
Assignee: Philip Nee (was: Kirk True)
> Revert changes to consumer_rolling_upgrade_test.py for the new async Consumer
> -
>
> Key: KAFKA-16579
> URL: https://issues.apache.org/jira/browse/KAFKA-16579
> Project: Kafka
> Issue Type: Task
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Philip Nee
>Priority: Critical
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated
> a slew of system tests to run both the "old" and "new" implementations.
> KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
> could test the new consumer with Connect. However, we are not supporting
> Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the
> Connect system tests with the new {{{}AsyncKafkaConsumer{}}}, we get errors
> like the following:
> {code:java}
> test_id:
> kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 6 minutes 3.899 seconds
> InsufficientResourcesError('Not enough nodes available to allocate. linux
> nodes requested: 1. linux nodes available: 0')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
> line 919, in test_exactly_once_source
> consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
> self.source.topic, consumer_timeout_ms=1000, print_key=True)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
> line 97, in __init__
> BackgroundThreadService.__init__(self, context, num_nodes)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
> line 26, in __init__
> super(BackgroundThreadService, self).__init__(context, num_nodes,
> cluster_spec, *args, **kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
> line 107, in __init__
> self.allocate_nodes()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
> line 217, in allocate_nodes
> self.nodes = self.cluster.alloc(self.cluster_spec)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/cluster.py",
> line 54, in alloc
> allocated = self.do_alloc(cluster_spec)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/finite_subcluster.py",
> line 31, in do_alloc
> allocated = self._available_nodes.remove_spec(cluster_spec)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/node_container.py",
> line 117, in remove_spec
> raise InsufficientResourcesError("Not enough nodes available to allocate.
> " + msg)
> ducktape.cluster.node_container.InsufficientResourcesError: Not enough nodes
> available to allocate. linux nodes requested: 1. linux nodes available: 0
> {code}
> The task here is to revert the changes made in KAFKA-16272 [PR
> 15576|https://github.com/apache/kafka/pull/15576].
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16578) Revert changes to connect_distributed_test.py for the new async Consumer
Kirk True created KAFKA-16578:
-
Summary: Revert changes to connect_distributed_test.py for the new
async Consumer
Key: KAFKA-16578
URL: https://issues.apache.org/jira/browse/KAFKA-16578
Project: Kafka
Issue Type: Task
Components: clients, consumer, system tests
Affects Versions: 3.8.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
To test the new, asynchronous Kafka {{Consumer}} implementation, we migrated a
slew of system tests to run both the "old" and "new" implementations.
KAFKA-16272 updated the system tests in {{connect_distributed_test.py}} so it
could test the new consumer with Connect. However, we are not supporting
Connect with the new consumer in 3.8.0. Unsurprisingly, when we run the Connect
system tests with the new {{{}AsyncKafkaConsumer{}}}, we get errors like the
following:
{code:java}
test_id:
kafkatest.tests.connect.connect_distributed_test.ConnectDistributedTest.test_exactly_once_source.clean=False.connect_protocol=eager.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 6 minutes 3.899 seconds
InsufficientResourcesError('Not enough nodes available to allocate. linux
nodes requested: 1. linux nodes available: 0')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/connect/connect_distributed_test.py",
line 919, in test_exactly_once_source
consumer_validator = ConsoleConsumer(self.test_context, 1, self.kafka,
self.source.topic, consumer_timeout_ms=1000, print_key=True)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/console_consumer.py",
line 97, in __init__
BackgroundThreadService.__init__(self, context, num_nodes)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/background_thread.py",
line 26, in __init__
super(BackgroundThreadService, self).__init__(context, num_nodes,
cluster_spec, *args, **kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 107, in __init__
self.allocate_nodes()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/services/service.py",
line 217, in allocate_nodes
self.nodes = self.cluster.alloc(self.cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/cluster.py",
line 54, in alloc
allocated = self.do_alloc(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/finite_subcluster.py",
line 31, in do_alloc
allocated = self._available_nodes.remove_spec(cluster_spec)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/cluster/node_container.py",
line 117, in remove_spec
raise InsufficientResourcesError("Not enough nodes available to allocate. "
+ msg)
ducktape.cluster.node_container.InsufficientResourcesError: Not enough nodes
available to allocate. linux nodes requested: 1. linux nodes available: 0
{code}
The task here is to revert the changes made in KAFKA-16272 [PR
15576|https://github.com/apache/kafka/pull/15576].
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16576) New consumer fails with assert in consumer_test.py’s test_consumer_failure system test
[
https://issues.apache.org/jira/browse/KAFKA-16576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16576:
--
Labels: flaky-test kip-848-client-support system-tests (was:
kip-848-client-support system-tests)
> New consumer fails with assert in consumer_test.py’s test_consumer_failure
> system test
> --
>
> Key: KAFKA-16576
> URL: https://issues.apache.org/jira/browse/KAFKA-16576
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: flaky-test, kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 42.582 seconds
> AssertionError()
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 399, in test_consumer_failure
> assert partition_owner is not None
> AssertionError
> Notify
> {code}
> Affected tests:
> * {{test_consumer_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16577) New consumer fails with stop within allotted timeout in consumer_test.py’s test_consumer_bounce system test
[
https://issues.apache.org/jira/browse/KAFKA-16577?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16577:
--
Labels: flaky-test kip-848-client-support system-tests (was:
kip-848-client-support system-tests)
> New consumer fails with stop within allotted timeout in consumer_test.py’s
> test_consumer_bounce system test
> ---
>
> Key: KAFKA-16577
> URL: https://issues.apache.org/jira/browse/KAFKA-16577
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: flaky-test, kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 2 minutes 35.925 seconds
> AssertionError('Node ubuntu@worker5: did not stop within the specified
> timeout of 30 seconds')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 176, in test_consumer_bounce
> self.rolling_bounce_consumers(consumer, clean_shutdown=clean_shutdown)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 39, in rolling_bounce_consumers
> consumer.stop_node(node, clean_shutdown)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
> line 419, in stop_node
> (str(node.account), str(self.stop_timeout_sec))
> AssertionError: Node ubuntu@worker5: did not stop within the specified
> timeout of 30 seconds
> {code}
> Affected tests:
> * {{test_consumer_bounce}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16577) New consumer fails with stop within allotted timeout in consumer_test.py’s test_consumer_bounce system test
[
https://issues.apache.org/jira/browse/KAFKA-16577?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16577:
--
Description:
The {{consumer_test.py}} system test intermittently fails with the following
error:
{code}
test_id:
kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 2 minutes 35.925 seconds
AssertionError('Node ubuntu@worker5: did not stop within the specified
timeout of 30 seconds')
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 176, in test_consumer_bounce
self.rolling_bounce_consumers(consumer, clean_shutdown=clean_shutdown)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 39, in rolling_bounce_consumers
consumer.stop_node(node, clean_shutdown)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/services/verifiable_consumer.py",
line 419, in stop_node
(str(node.account), str(self.stop_timeout_sec))
AssertionError: Node ubuntu@worker5: did not stop within the specified timeout
of 30 seconds
{code}
Affected tests:
* {{test_consumer_bounce}}
was:
The {{consumer_test.py}} system test intermittently fails with the following
error:
{code}
test_id:
kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 42.582 seconds
AssertionError()
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 399, in test_consumer_failure
assert partition_owner is not None
AssertionError
Notify
{code}
Affected tests:
* {{test_consumer_failure}}
> New consumer fails with stop within allotted timeout in consumer_test.py’s
> test_consumer_bounce system test
> ---
>
> Key: KAFKA-16577
> URL: https://issues.apache.org/jira/browse/KAFKA-16577
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_bounce.clean_shutdown=True.bounce_mode=rolling.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 2 minutes 35.925 seconds
> AssertionError('Node ubuntu@worker5: did not stop within the specified
> timeout of 30 seconds')
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return
[jira] [Updated] (KAFKA-16577) New consumer fails with stop within allotted timeout in consumer_test.py’s test_consumer_bounce system test
[
https://issues.apache.org/jira/browse/KAFKA-16577?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16577:
--
Summary: New consumer fails with stop within allotted timeout in
consumer_test.py’s test_consumer_bounce system test (was: New consumer fails
with stop timeout in consumer_test.py’s test_consumer_bounce system test)
> New consumer fails with stop within allotted timeout in consumer_test.py’s
> test_consumer_bounce system test
> ---
>
> Key: KAFKA-16577
> URL: https://issues.apache.org/jira/browse/KAFKA-16577
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 42.582 seconds
> AssertionError()
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 399, in test_consumer_failure
> assert partition_owner is not None
> AssertionError
> Notify
> {code}
> Affected tests:
> * {{test_consumer_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16577) New consumer fails with stop timeout in consumer_test.py’s test_consumer_bounce system test
Kirk True created KAFKA-16577:
-
Summary: New consumer fails with stop timeout in
consumer_test.py’s test_consumer_bounce system test
Key: KAFKA-16577
URL: https://issues.apache.org/jira/browse/KAFKA-16577
Project: Kafka
Issue Type: Bug
Components: clients, consumer, system tests
Affects Versions: 3.7.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
The {{consumer_test.py}} system test intermittently fails with the following
error:
{code}
test_id:
kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 42.582 seconds
AssertionError()
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 399, in test_consumer_failure
assert partition_owner is not None
AssertionError
Notify
{code}
Affected tests:
* {{test_consumer_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (KAFKA-16576) New consumer fails with assert in consumer_test.py’s test_consumer_failure system test
[
https://issues.apache.org/jira/browse/KAFKA-16576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16576:
--
Description:
The {{consumer_test.py}} system test intermittently fails with the following
error:
{code}
test_id:
kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
status: FAIL
run time: 42.582 seconds
AssertionError()
Traceback (most recent call last):
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 184, in _do_run
data = self.run_test()
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
line 262, in run_test
return self.test_context.function(self.test)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
line 433, in wrapper
return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
File
"/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
line 399, in test_consumer_failure
assert partition_owner is not None
AssertionError
Notify
{code}
Affected tests:
* {{test_consumer_failure}}
was:
The {{consumer_test.py}} system test fails with the following errors:
{quote}
* Timed out waiting for consumption
{quote}
Affected tests:
* {{test_broker_failure}}
* {{test_consumer_bounce}}
* {{test_static_consumer_bounce}}
> New consumer fails with assert in consumer_test.py’s test_consumer_failure
> system test
> --
>
> Key: KAFKA-16576
> URL: https://issues.apache.org/jira/browse/KAFKA-16576
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.7.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> The {{consumer_test.py}} system test intermittently fails with the following
> error:
> {code}
> test_id:
> kafkatest.tests.client.consumer_test.OffsetValidationTest.test_consumer_failure.clean_shutdown=True.enable_autocommit=True.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 42.582 seconds
> AssertionError()
> Traceback (most recent call last):
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 184, in _do_run
> data = self.run_test()
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/tests/runner_client.py",
> line 262, in run_test
> return self.test_context.function(self.test)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/venv/lib/python3.7/site-packages/ducktape/mark/_mark.py",
> line 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/home/jenkins/workspace/system-test-kafka-branch-builder/kafka/tests/kafkatest/tests/client/consumer_test.py",
> line 399, in test_consumer_failure
> assert partition_owner is not None
> AssertionError
> Notify
> {code}
> Affected tests:
> * {{test_consumer_failure}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16576) New consumer fails with assert in consumer_test.py’s test_consumer_failure system test
Kirk True created KAFKA-16576:
-
Summary: New consumer fails with assert in consumer_test.py’s
test_consumer_failure system test
Key: KAFKA-16576
URL: https://issues.apache.org/jira/browse/KAFKA-16576
Project: Kafka
Issue Type: Bug
Components: clients, consumer, system tests
Affects Versions: 3.7.0
Reporter: Kirk True
Assignee: Kirk True
Fix For: 3.8.0
The {{consumer_test.py}} system test fails with the following errors:
{quote}
* Timed out waiting for consumption
{quote}
Affected tests:
* {{test_broker_failure}}
* {{test_consumer_bounce}}
* {{test_static_consumer_bounce}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Resolved] (KAFKA-16405) Mismatch assignment error when running consumer rolling upgrade system tests
[
https://issues.apache.org/jira/browse/KAFKA-16405?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True resolved KAFKA-16405.
---
Reviewer: Lucas Brutschy
Resolution: Fixed
> Mismatch assignment error when running consumer rolling upgrade system tests
>
>
> Key: KAFKA-16405
> URL: https://issues.apache.org/jira/browse/KAFKA-16405
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Reporter: Philip Nee
>Assignee: Philip Nee
>Priority: Blocker
> Labels: kip-848-client-support, system-tests
> Fix For: 3.8.0
>
>
> relevant to [https://github.com/apache/kafka/pull/15578]
>
> We are seeing:
> {code:java}
>
> SESSION REPORT (ALL TESTS)
> ducktape version: 0.11.4
> session_id: 2024-03-21--001
> run time: 3 minutes 24.632 seconds
> tests run:7
> passed: 5
> flaky:0
> failed: 2
> ignored: 0
>
> test_id:
> kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=COMBINED_KRAFT.use_new_coordinator=True.group_protocol=classic
> status: PASS
> run time: 24.599 seconds
>
> test_id:
> kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=COMBINED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 26.638 seconds
> AssertionError("Mismatched assignment: {frozenset(),
> frozenset({TopicPartition(topic='test_topic', partition=3),
> TopicPartition(topic='test_topic', partition=0),
> TopicPartition(topic='test_topic', partition=1),
> TopicPartition(topic='test_topic', partition=2)})}")
> Traceback (most recent call last):
> File
> "/usr/local/lib/python3.9/dist-packages/ducktape/tests/runner_client.py",
> line 186, in _do_run
> data = self.run_test()
> File
> "/usr/local/lib/python3.9/dist-packages/ducktape/tests/runner_client.py",
> line 246, in run_test
> return self.test_context.function(self.test)
> File "/usr/local/lib/python3.9/dist-packages/ducktape/mark/_mark.py", line
> 433, in wrapper
> return functools.partial(f, *args, **kwargs)(*w_args, **w_kwargs)
> File
> "/opt/kafka-dev/tests/kafkatest/tests/client/consumer_rolling_upgrade_test.py",
> line 77, in rolling_update_test
> self._verify_range_assignment(consumer)
> File
> "/opt/kafka-dev/tests/kafkatest/tests/client/consumer_rolling_upgrade_test.py",
> line 38, in _verify_range_assignment
> assert assignment == set([
> AssertionError: Mismatched assignment: {frozenset(),
> frozenset({TopicPartition(topic='test_topic', partition=3),
> TopicPartition(topic='test_topic', partition=0),
> TopicPartition(topic='test_topic', partition=1),
> TopicPartition(topic='test_topic', partition=2)})}
>
> test_id:
> kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=False
> status: PASS
> run time: 29.815 seconds
>
> test_id:
> kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True
> status: PASS
> run time: 29.766 seconds
>
> test_id:
> kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=classic
> status: PASS
> run time: 30.086 seconds
>
> test_id:
> kafkatest.tests.client.consumer_rolling_upgrade_test.ConsumerRollingUpgradeTest.rolling_update_test.metadata_quorum=ISOLATED_KRAFT.use_new_coordinator=True.group_protocol=consumer
> status: FAIL
> run time: 35.965 seconds
> AssertionError("Mismatched assignment: {frozenset(),
> frozenset({TopicPartition(topic='test_topic', partition=3),
> TopicPartition(topic='test_topic', partition=0),
> TopicPartition(topic='test_topic', partition=1),
> TopicPartition(topic='test_topic', partition=2)})}")
> Traceback (most recent call last):
> File
> "/usr/local/lib/python3.9/dist-packages/ducktape/tests/runner_client.py",
> line 186, in _do_run
>
[jira] [Updated] (KAFKA-16565) IncrementalAssignmentConsumerEventHandler throws error when attempting to remove a partition that isn't assigned
[
https://issues.apache.org/jira/browse/KAFKA-16565?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True updated KAFKA-16565:
--
Labels: consumer-threading-refactor kip-848-client-support system-tests
(was: )
> IncrementalAssignmentConsumerEventHandler throws error when attempting to
> remove a partition that isn't assigned
>
>
> Key: KAFKA-16565
> URL: https://issues.apache.org/jira/browse/KAFKA-16565
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, kip-848-client-support,
> system-tests
> Fix For: 3.8.0
>
>
> In {{{}verifiable_consumer.py{}}}, the Incremental
>
> {code:java}
> def handle_partitions_revoked(self, event):
> self.revoked_count += 1
> self.state = ConsumerState.Rebalancing
> self.position = {}
> for topic_partition in event["partitions"]:
> topic = topic_partition["topic"]
> partition = topic_partition["partition"]
> self.assignment.remove(TopicPartition(topic, partition))
> {code}
> If the {{self.assignment.remove()}} call is passed a {{TopicPartition}} that
> isn't in the list, an error is thrown. For now, we should first check that
> the {{TopicPartition}} is in the list, and if not, log a warning or something.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Assigned] (KAFKA-16565) IncrementalAssignmentConsumerEventHandler throws error when attempting to remove a partition that isn't assigned
[
https://issues.apache.org/jira/browse/KAFKA-16565?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kirk True reassigned KAFKA-16565:
-
Assignee: Kirk True
> IncrementalAssignmentConsumerEventHandler throws error when attempting to
> remove a partition that isn't assigned
>
>
> Key: KAFKA-16565
> URL: https://issues.apache.org/jira/browse/KAFKA-16565
> Project: Kafka
> Issue Type: Bug
> Components: clients, consumer, system tests
>Affects Versions: 3.8.0
>Reporter: Kirk True
>Assignee: Kirk True
>Priority: Critical
> Labels: consumer-threading-refactor, kip-848-client-support,
> system-tests
> Fix For: 3.8.0
>
>
> In {{{}verifiable_consumer.py{}}}, the Incremental
>
> {code:java}
> def handle_partitions_revoked(self, event):
> self.revoked_count += 1
> self.state = ConsumerState.Rebalancing
> self.position = {}
> for topic_partition in event["partitions"]:
> topic = topic_partition["topic"]
> partition = topic_partition["partition"]
> self.assignment.remove(TopicPartition(topic, partition))
> {code}
> If the {{self.assignment.remove()}} call is passed a {{TopicPartition}} that
> isn't in the list, an error is thrown. For now, we should first check that
> the {{TopicPartition}} is in the list, and if not, log a warning or something.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Created] (KAFKA-16565) IncrementalAssignmentConsumerEventHandler throws error when attempting to remove a partition that isn't assigned
Kirk True created KAFKA-16565:
-
Summary: IncrementalAssignmentConsumerEventHandler throws error
when attempting to remove a partition that isn't assigned
Key: KAFKA-16565
URL: https://issues.apache.org/jira/browse/KAFKA-16565
Project: Kafka
Issue Type: Bug
Components: clients, consumer, system tests
Affects Versions: 3.8.0
Reporter: Kirk True
Fix For: 3.8.0
In {{{}verifiable_consumer.py{}}}, the Incremental
{code:java}
def handle_partitions_revoked(self, event):
self.revoked_count += 1
self.state = ConsumerState.Rebalancing
self.position = {}
for topic_partition in event["partitions"]:
topic = topic_partition["topic"]
partition = topic_partition["partition"]
self.assignment.remove(TopicPartition(topic, partition))
{code}
If the {{self.assignment.remove()}} call is passed a {{TopicPartition}} that
isn't in the list, an error is thrown. For now, we should first check that the
{{TopicPartition}} is in the list, and if not, log a warning or something.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
