Add test for concurrently updatingMetadata and fetching snapshot/cluster
Reviewers: Jason Gustafson <jason@confluent.io>
Co-authored-by: Zhifeng Chen <ericzhifengchen@gmail.com>
Treats KAFKA-16277 - CooperativeStickyAssignor does not spread topics evenly among consumer group
Reviewers: Anna Sophie Blee-Goldman <ableegoldman@apache.org>
Introduce call to onPartitionsLost callback to release assignment when a consumer pro-actively leaves the group due to poll timer expired.
When the poll timer expires, the member sends a leave group request (reusing same existing LEAVING state and logic), and then transitions to STALE to release it assignment and wait for the poll timer reset. Once both conditions are met, the consumer transitions out of the STALE state to rejoin the group. Note that while on this STALE state, the member is not part of the group so it does not send heartbeats.
This PR also includes the fix to ensure that while STALE or in any other state where the member is not in the group, heartbeat responses that may be received are ignored.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
Introduces a new filter in ListTransactionsRequest API. This enables caller to filter on transactions that have been running for longer than a certain duration of time.
This PR includes the following changes:
bumps version for ListTransactionsRequest API to 1. Set the durationFilter to -1L when communicating with an older broker that does not support version 1.
bumps version for ListTransactionsResponse to 1 without changing the response structure.
adds durationFilter option to kafka-transactions.sh --list
Tests:
Client side test to build request with correct combination of duration filter and API version: testBuildRequestWithDurationFilter
Server side test to filter transactions based on duration: testListTransactionsFiltering
Added test case for kafka-transactions.sh change in TransactionsCommandTest
Reviewers: Justine Olshan <jolshan@confluent.io>, Raman Verma <rverma@confluent.io>
KRaft was only notifying listeners of the latest leader and epoch when the replica transition to a new state. This can result in the listener never getting notified if the registration happened after it had become a follower.
This problem doesn't exists for the active leader because the KRaft implementation attempts to notified the listener of the latest leader and epoch when the replica is the active leader.
This issue is fixed by notifying the listeners of the latest leader and epoch after processing the listener registration request.
Reviewers: Colin P. McCabe <cmccabe@apache.org>
Commit 1442862 introduced a test with an assertion inside a listener
callback. This improves the test by checking that the listener is
actually being executed, to avoid silent skipping of the assertion.
Reviewers: Matthias J. Sax <matthias@confluent.io>
The group coordinator expects the instance ID to always be sent when
leaving the group in a static membership configuration, see
ea94507679/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java (L814)
The failure was silent, because the group coordinator does not log
failed requests and the consumer doesn't wait for the heartbeat response
during close.
Reviewers: Matthias J. Sax <matthias@confluent.io>, Kirk True <ktrue@confluent.io>, Bruno Cadonna <cadonna@apache.org>
Fix to ensure that a consumer that has been fenced by the coordinator stops sending heartbeats while it is on the FENCED state releasing its assignment (waiting for the onPartitionsLost callback to complete). Once the callback completes, the member transitions to JOINING and it's then when it should resume sending heartbeats again.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
a3528a3 removed this log but not the test asserting it.
Builds are currently red because for some reason these tests can't retry. We should address that as a followup.
Reviewers: Greg Harris <greg.harris@aiven.io>, Matthias J. Sax <matthias@confluent.io>
The usual flow of updating the upgrade.html docs is to first do it in apache/kafka/trunk, then cherry-pick to the relative release branch and then copy into the kafka-site repo.
It seems like this was not done with a few commits updating the 3.6.1, 3.5.2 and 3.5.1, resulting in kafka-site's latest upgrade.html containing content that isn't here. This was caught while we were adding the 3.7 upgrade docs.
This patch reconciles both files by taking the extra changes from kafka-site and placing them here. This was done by simply comparing a diff of both changes and taking the ones that apply
Due to the asynchronous nature of the async consumer, it might happen that on the application thread the group metadata is not known after the first poll returns records. If the offsets of those records are then send to a transaction with
txnProducer.sendOffsetsToTransaction(offsetsToCommit, groupMetadata);
and then the transaction is committed, the group coordinator will raise an error saying that the member is not known since the member in groupMetadata is still -1 before the metadata is updated.
This commit avoids this error by not returning any records from poll() until the group metadata is updated, i.e., the member ID and the generation ID (a.k.a. member epoch) are known. This check is only done if group management is used.
Additionally, this commit resets the group metadata when the consumer unsubscribes.
Reviewer: Lucas Brutschy <lbrutschy@confluent.io>
We already record dropping record via metrics and logging at WARN level
is too noise. This PR removes the unnecessary logging.
Reviewers: Kalpesh Patel <kpatel@confluent.io>, Walker Carlson <wcarlson@confluent.io>, Lucas Brutschy <lbrutschy@confluent.io>
Variable metadataMock was removed by #15323 after the CI build of #15320 was run and before #15320 was merged.
Reviewers: Luke Chen <showuon@gmail.com>, Lucas Brutschy <lbrutschy@confluent.io>
This change modifies the commit manager for improved retry logic & fixing bugs:
- defines high level functions for each of the different types of commit: commitSync, commitAsync, autoCommitSync (used from consumer close), autoCommitAsync (on interval), autoCommitNow (before revocation).
- moves retry logic to these caller functions, keeping a common response error handling that propagates errors that each caller functions retry as it needs.
Fixes the following issues:
- auto-commit before revocation should retry with latest consumed offsets
- auto-commit before revocation should only reset the timer once, when the rebalance completes
- StaleMemberEpoch error (fatal) is considered retriable only when committing offsets before revocation, where it is retried with backoff if the member has a valid epoch. All other commits will fail fatally on stale epoch. Note that auto commit on the interval (autoCommitAsync) does not have any specific retry logic for the stale epoch, but will effectively retry on the next interval (as it does for any other fatal error)
- fix duplicated and noisy logs for auto-commit
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
A performance regression introduced in commit 5bc3aa4 reduces the write performance in RocksDB by ~3x. The bug is that we fail to pass the WriteOptions that disable the write-ahead log into the DB accessor.
For testing, the time to write 10 times 1 Million records into one RocksDB each were measured:
Before 5bc3aa4: 7954ms, 12933ms
After 5bc3aa4: 30345ms, 31992ms
After 5bc3aa4 with this fix: 8040ms, 10563ms
On current trunk with this fix: 9508ms, 10441ms
Reviewers: Bruno Cadonna <bruno@confluent.io>, Nick Telford <nick.telford@gmail.com>
There is a test already that checks that transactional messages are not
drained when partition is not added, this change just logically
completes the test to also show that messages can be drained after
parition is added.
Reviewers: Omnia Ibrahim <o.g.h.ibrahim@gmail.com>, Tzu-Li (Gordon) Tai <tzulitai@apache.org>, Justine Olshan <jolshan@confluent.io>
ConsistencyVectorIntegrationTest failed frequently because the return
Position from IQv2 is not immutable while the test assume immutability.
To return a Position with a QueryResult that does not change, we need to
deep copy the Position object.
Reviewers: John Roesler <john@confluent.io>, Lucas Brutschy <lbrutschy@confluent.io>
The consumer keeps a poll timer, which is used to ensure liveness of the application thread. The poll timer automatically updates while the Consumer.poll(Duration) method is blocked, while the newer consumer only updates the poll timer when a new call to Consumer.poll(Duration) is issued. This means that the kafka-console-consumer.sh tools, which uses a very long timeout by default, works differently with the new consumer, with the consumer proactively rejoining the group during long poll timeouts.
This change solves the problem by (a) repeatedly sending PollApplicationEvents to the background thread, not just on the first call of poll and (b) making sure that the application thread doesn't block for so long that it runs out of max.poll.interval.
An integration test is added to make sure that we do not rejoin the group when a long poll timeout is used with a low max.poll.interval.
Reviewers: Lianet Magrans <lianetmr@gmail.com>, Andrew Schofield <aschofield@confluent.io>, Bruno Cadonna <cadonna@apache.org>
The assertion to check BytesOut doesn't include replication was performed before replication occurred.
This PR fixed the position of the assertion.
Reviewers: Luke Chen <showuon@gmail.com>
In this modification, if ./gradlew systemTestLibs fails, the script will output an error message and terminate execution using the die function. This ensures that the script fails fast and prompts the user to address the error before continuing.
Reviewers: Luke Chen <showuon@gmail.com>
This PR cleans up: metrics, migration, network, raft, security, serializer, tools, utils, and zookeeper package classes
Mark methods and fields private where possible
Annotate public methods and fields
Remove unused classes and methods
Make sure Arrays are not printed with .toString
Optimize minor warnings
Reviewers: Mickael Maison <mickael.maison@gmail.com>
The wake-up mechanism in the new consumer is preventing from committing within a rebalance listener callback. The reason is that we are trying to register two wake-uppable actions at the same time.
The fix is to register the wake-uppable action more closely to where we are in fact blocking on it, so that the action is not registered when we execute rebalance listeneners and callback listeners.
Reviewers: Bruno Cadonna <cadonna@apache.org>
This PR improves TopicCommandIntegrationTest by :
- using TestUtils.createTopicWithAdmin
- replacing \n with lineSeperator
- using waitForAllReassignmentsToComplete
- adding more log when assertion fails
Reviewers: Luke Chen <showuon@gmail.com>, Justine Olshan <jolshan@confluent.io>
While reviewing https://github.com/apache/kafka/pull/15150, I found that our tests verifying the console output are really hard to read. Here is my proposal to make it better.
Reviewers: Ritika Reddy <rreddy@confluent.io>, Justine Olshan <jolshan@confluent.io>
Currently, when initializing StandardAuthorizer, it'll wait until all ACL loaded and complete the initialLoadFuture. So, checking logs, we'll see:
2023-12-06 14:07:50,325 INFO [StandardAuthorizer 1] Initialized with 6 acl(s). (org.apache.kafka.metadata.authorizer.StandardAuthorizerData) [kafka-1-metadata-loader-event-handler]
2023-12-06 14:07:50,325 INFO [StandardAuthorizer 1] Completed initial ACL load process. (org.apache.kafka.metadata.authorizer.StandardAuthorizerData) [kafka-1-metadata-loader-event-handler]
But then, when shutting down the node, we will also see this error:
2023-12-06 14:12:32,752 ERROR [StandardAuthorizer 1] Failed to complete initial ACL load process. (org.apache.kafka.metadata.authorizer.StandardAuthorizerData) [kafka-1-metadata-loader-event-handler]
java.util.concurrent.TimeoutException
at kafka.server.metadata.AclPublisher.close(AclPublisher.scala:98)
at org.apache.kafka.image.loader.MetadataLoader.closePublisher(MetadataLoader.java:568)
at org.apache.kafka.image.loader.MetadataLoader.lambda$removeAndClosePublisher$7(MetadataLoader.java:528)
at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:127)
at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:210)
at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:181)
at java.base/java.lang.Thread.run(Thread.java:840)
It's confusing. And it's because we'll try to complete authorizer initialLoad, and complete the initialLoadFuture if not done. But we'll log the error no matter it's completed or not. This patch improves the logging.
Reviewers: Josep Prat <josep.prat@aiven.io>
When completing the partition reassignment, the new ISR should have all the target replicas.
Reviewers: Justine Olshan <jolshan@confluent.io>, David Mao <dmao@confluent.io>
This trivial PR makes clear when it's the right time to switch from AclAuthorizer to StandardAuthorizer during the migration process.
Reviewers: Luke Chen <showuon@gmail.com>
During ZK migrating to KRaft, before entering dual-write mode, the KRaft controller will send RPCs (i.e. UpdateMetadataRequest, LeaderAndIsrRequest, and StopReplicaRequest) to the brokers. Currently, we use the inter broker listener to send the RPC to brokers from the controller. But in the doc, we didn't provide this info to users because the normal KRaft controller won't use inter.broker.listener.names.
This PR adds the missing config in the ZK migrating to KRaft doc.
Reviewers: Mickael Maison <mickael.maison@gmail.com>, Paolo Patierno <ppatierno@live.com>
The Python VerifiableConsumer now passes in the --group-protocol and --group-remote-assignor command line arguments to VerifiableConsumer if the node is running 3.7.0+ and using the new consumer group.protocol.
Reviewers: Andrew Schofield <aschofield@confluent.io>, Lucas Brutschy <lbrutschy@confluent.io>
This fixes an invalid transition (leaving->stale) that was discovered in the system tests. The underlying issue was that the poll timer expiration logic was blindly forcing a transition to stale and sending a leave group, without considering that the member could be already leaving.
The fix included in this PR ensures that the poll timer expiration logic, whose purpose is to leave the group, is only applied if the member is not already leaving. Note that it also fixes the transition out of the STALE state, that should only happen when the poll timer is reset.
As a result of this changes:
If the poll timer expires while the member is not leaving, the poll timer expiration logic is applied: it will transition to stale, send a leave group, and remain in STALE state until the timer is reset. At that point the member will transition to JOINING to rejoin the group.
If the poll timer expires while the member is already leaving, the poll timer expiration logic does not apply, and just lets the HB continue. Not that this would be the case of member in PREPARE_LEAVING waiting for callbacks to complete (needs to continue sending HB), or LEAVING (needs to send the last HB to leave).
Reviewers: Kirk True <ktrue@confluent.io>, Andrew Schofield <aschofield@confluent.io>, Lucas Brutschy <lbrutschy@confluent.io>
This integration test is now passing, presumably based on recent related changes. Re-enabling to ensure it is included in the test suite to catch any regressions.
Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
As this [JIRA](https://issues.apache.org/jira/browse/KAFKA-16226) explains, there is increased synchronization between application-thread, and the background thread as the background thread started to synchronized methods Metadata.currentLeader() in [original PR](https://github.com/apache/kafka/pull/14384). So this PR does the following changes
1. Changes background thread, i.e. RecordAccumulator's partitionReady(), and drainBatchesForOneNode(), to not use `Metadata.currentLeader()`. Instead rely on `MetadataCache` that is immutable. So access to it is unsynchronized.
2. This PR repurposes `MetadataCache` as an immutable snapshot of Metadata. This is a wrapper around public `Cluster`. `MetadataCache`'s API/functionality should be extended for internal client usage Vs public `Cluster`. For example, this PR adds `MetadataCache.leaderEpochFor()`
3. Rename `MetadataCache` to `MetadataSnapshot` to make it explicit its immutable.
**Note both `Cluster` and `MetadataCache` are not syncronized, hence reduce synchronization from the hot path for high partition counts.**
Reviewers: Jason Gustafson <jason@confluent.io>
Spotted the following log:
```
[2024-02-14 09:59:30,103] INFO [GroupCoordinator id=1] Finished loading of metadata from 39 in __consumer_offsets-4ms with epoch 2 where 39ms was spent in the scheduler. Loaded 0 records which total to 0 bytes. (org.apache.kafka.coordinator.group.runtime.CoordinatorRuntime)
```
The partition and the time are incorrect. This patch fixes it.
Reviewers: Jeff Kim <jeff.kim@confluent.io>, Bruno Cadonna <bruno@confluent.io>
The main bug causing this test to fail as described in the ticket was already fixed.
The test is still flaky if unchanged, because in the new consumer, the assignment can
change in between two polls. Interceptors are only executed inside poll (and have to be,
since they must run as part of the application thread), so we need to modify the
integration test to call poll once after observing that the assignment changed.
Reviewers: Bruno Cadonna <bruno@confluent.io>
`GroupMetadataManagerTest` class got a little under control. We have too many things defined in it. As a first steps, this patch extracts all the inner classes. It also extracts all the helper methods. However, the logic is not changed at all.
Reviewers: Omnia Ibrahim <o.g.h.ibrahim@gmail.com>, Justine Olshan <jolshan@confluent.io>
Fix the flakiness of LogDirFailureTest by setting a separate metadata.log.dir for brokers in KRAFT mode.
The test was flaky because as we call causeLogDirFailure some times we impact the first log.dir which also is KafkaConfig.metadataLogDir as we don't have metadata.log.dir. So to fix the flakiness we need to explicitly set metadata.log.dir to diff log dir than the ones we could potentially fail for the tests.
This is part 1 of the fixes. Delivering them separately as the other issues were not as clear cut.
Reviewers: Gaurav Narula <gaurav_narula2@apple.com>, Justine Olshan <jolshan@confluent.io>, Greg Harris <greg.harris@aiven.io>
When the consumer enters state LEAVING, it sets the epoch to the leave epoch,
such as -1. When the timing is right, we may get a heartbeat response after
entering the state LEAVING, which resets the epoch to the member epoch on
the server. The result is that the consumer never leaves the group.
Seems like c6f4c604d8 changed the timing inside
the consumer to relatively frequently triggers this problem inside
`DescribeConsumerGroupTest`.
We fix it by ignoring any heartbeat responses when we are in state LEAVING.
Reviewers: David Jacot <djacot@confluent.io>
Mayank Shekhar Narula
Cameron Redpath
Lianet Magrans
Anton Agestam
Yang Yu
José Armando García Sancio
Lucas Brutschy
Daan Gerits
Justine Olshan
Stanislav Kozlovski
Josep Prat
Bruno Cadonna
Matthias J. Sax
Anton Liauchuk
Paolo Patierno
Artem Livshits
Omnia Ibrahim
runom
Ahmed Sobeh
Owen Leung
David Jacot
Luke Chen
Calvin Liu
Kirk True
Minha, Jeong
Mickael Maison
Gantigmaa Selenge