kafka.git
25 hours agoMINOR: Deflake OptimizedKTableIntegrationTest (#12186) trunk
John Roesler [Fri, 20 May 2022 14:17:39 +0000 (09:17 -0500)] 
MINOR: Deflake OptimizedKTableIntegrationTest (#12186)

This test has been flaky due to unexpected rebalances during the test.
This change fixes it by detecting an unexpected rebalance and retrying
the test logic (within a timeout).

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <guozhang@apache.org>

39 hours agoKAFKA-13889: Fix AclsDelta handling of REMOVE_ACCESS_CONTROL_ENTRY_RECORD (#12160)
andymg3 [Thu, 19 May 2022 23:24:34 +0000 (19:24 -0400)] 
KAFKA-13889: Fix AclsDelta handling of REMOVE_ACCESS_CONTROL_ENTRY_RECORD (#12160)

AclsDelta stores the pending deletion in the changes Map. This could override a creation that might have just happened. This is an issue because in BrokerMetadataPublisher this results in us making a removeAcl call which finally results in StandardAuthorizerData trying to remove an ACL that doesn't exist which throws an exception. If the ACCESS_CONTROL_ENTRY_RECORD event never got processed by BrokerMetadataPublisher then the ACL wont be in the Map yet.

The fix here is to remove the entry from the changes Map if the ACL doesnt exist in the image yet.

Reviewers: José Armando García Sancio <jsancio@users.noreply.github.com>

46 hours agoKAFKA-13863; Prevent null config value when create topic in KRaft mode (#12109)
dengziming [Thu, 19 May 2022 16:46:48 +0000 (00:46 +0800)] 
KAFKA-13863; Prevent null config value when create topic in KRaft mode (#12109)

This patch ensures consistent handling of null-valued topic configs between the zk and kraft controller. Prior to this patch, we returned INVALID_REQUEST in zk mode and it was not an error in kraft. After this patch, we return INVALID_CONFIG consistently for this case.

Reviewers: Jason Gustafson <jason@confluent.io>

2 days agoKAFKA-13837; Return an error from Fetch if follower is not a valid replica (#12150)
Jason Gustafson [Thu, 19 May 2022 03:58:20 +0000 (20:58 -0700)] 
KAFKA-13837; Return an error from Fetch if follower is not a valid replica (#12150)

When a partition leader receives a `Fetch` request from a replica which is not in the current replica set, the behavior today is to return a successful fetch response, but with empty data. This causes the follower to retry until metadata converges without updating any state on the leader side. It is clearer in this case to return an error, so that the metadata inconsistency is visible in logging and so that the follower backs off before retrying.

In this patch, we use `UNKNOWN_LEADER_EPOCH` when the `Fetch` request includes the current leader epoch. The way we see this is that the leader is validating the (replicaId, leaderEpoch) tuple. When the leader returns `UNKNOWN_LEADER_EPOCH`, it means that the leader does not expect the given leaderEpoch from that replica. If the request does not include a leader epoch, then we use `NOT_LEADER_OR_FOLLOWER`. We can take a similar interpretation for this case: the leader is rejecting the request because it does not think it should be the leader for that replica. But mainly these errors ensure that the follower will retry the request.

As a part of this patch, I have refactored the way that the leader updates follower fetch state. Previously, the process is a little convoluted. We send the fetch from `ReplicaManager` down to `Partition.readRecords`, then we iterate over the results and call `Partition.updateFollowerFetchState`. It is more straightforward to update state directly as a part of `readRecords`. All we need to do is pass through the `FetchParams`. This also prevents an unnecessary copy of the read results.

Reviewers: David Jacot <djacot@confluent.io>

2 days agoMINOR: Fix typo in ReplicaManagerTest (#12178)
bozhao12 [Thu, 19 May 2022 02:28:47 +0000 (10:28 +0800)] 
MINOR: Fix typo in ReplicaManagerTest (#12178)

Reviewer: Luke Chen <showuon@gmail.com>

2 days agoMINOR: Enable KRaft in `TransactionsTest` (#12176)
Jason Gustafson [Wed, 18 May 2022 21:07:59 +0000 (14:07 -0700)] 
MINOR: Enable KRaft in `TransactionsTest` (#12176)

Enable support for KRaft in `TransactionsTest`.

Reviewers: David Arthur <mumrah@gmail.com>

2 days agoKAFKA-13830 MetadataVersion integration for KRaft controller (#12050)
David Arthur [Wed, 18 May 2022 19:08:36 +0000 (15:08 -0400)] 
KAFKA-13830 MetadataVersion integration for KRaft controller (#12050)

This patch builds on #12072 and adds controller support for metadata.version. The kafka-storage tool now allows a
user to specify a specific metadata.version to bootstrap into the cluster, otherwise the latest version is used.

Upon the first leader election of the KRaft quroum, this initial metadata.version is written into the metadata log. When
writing snapshots, a FeatureLevelRecord for metadata.version will be written out ahead of other records so we can
decode things at the correct version level.

This also includes additional validation in the controller when setting feature levels. It will now check that a given
metadata.version is supportable by the quroum, not just the brokers.

Reviewers: José Armando García Sancio <jsancio@gmail.com>, Colin P. McCabe <cmccabe@apache.org>, dengziming <dengziming1993@gmail.com>, Alyssa Huang <ahuang@confluent.io>

2 days agoMINOR: Replace string literal with constant in RequestChannel (#12134)
runom [Wed, 18 May 2022 18:31:15 +0000 (03:31 +0900)] 
MINOR: Replace string literal with constant in RequestChannel (#12134)

Replace the "RequestsPerSec" literal value with the pre-existing constant `RequestsPerSec`.

Reviewers: Divij Vaidya <divijvaidya13@gmail.com>, Jason Gustafson <jason@confluent.io>

2 days agoMINOR: Enable some AdminClient integration tests (#12110)
dengziming [Wed, 18 May 2022 16:39:26 +0000 (00:39 +0800)] 
MINOR: Enable some AdminClient integration tests (#12110)

Enable KRaft in `AdminClientWithPoliciesIntegrationTes`t and `PlaintextAdminIntegrationTest`. There are some tests not enabled or not as expected yet:

- testNullConfigs, see KAFKA-13863
- testDescribeCluster and testMetadataRefresh, currently we don't get the real controller in KRaft mode so the test may not run as expected

This patch also changes the exception type raised from invalid `IncrementalAlterConfig` requests with the `SUBTRACT` and `APPEND` operations. When the configuration value type is not a list, we now raise `INVALID_CONFIG` instead of `INVALID_REQUEST`.

Reviewers: Luke Chen <showuon@gmail.com>, Jason Gustafson <jason@confluent.io>

3 days agoMINOR: Remove extraneous code in LocalLogManager (#12168)
David Arthur [Wed, 18 May 2022 14:58:08 +0000 (10:58 -0400)] 
MINOR: Remove extraneous code in LocalLogManager (#12168)

Reviewers: Kvicii <Karonazaba@gmail.com>, dengziming <dengziming1993@gmail.com>, Divij Vaidya <divijvaidya13@gmail.com>

3 days agoMINOR: Followers should not have any remote replica states left over from previous...
David Jacot [Wed, 18 May 2022 07:32:48 +0000 (09:32 +0200)] 
MINOR: Followers should not have any remote replica states left over from previous leadership (#12138)

This patch ensures that followers don't have any remote replica states left over from previous leadership.

Reviewers: Jason Gustafson <jason@confluent.io>

3 days agoMINOR: Remove redundant metric reset in KafkaController (#12158)
bozhao12 [Tue, 17 May 2022 22:40:05 +0000 (06:40 +0800)] 
MINOR: Remove redundant metric reset in KafkaController (#12158)

The following variables in `KafkaController` are used for metrics:
```
    offlinePartitionCount
    preferredReplicaImbalanceCount
    globalTopicCount
    globalPartitionCount
    topicsToDeleteCount
    replicasToDeleteCount
    ineligibleTopicsToDeleteCount
    ineligibleReplicasToDeleteCount
```
When the controller goes from active to non-active, these variables will be reset to 0. Currently, this is done explicitly in in `KafkaController.onControllerResignation()` and also after every loop iteration in `KafkaController.updateMetrics()` .
The first of these is redundant and can be removed. This patch fixes this and also simplifies `updateMetrics`.

Reviewers: Jason Gustafson <jason@confluent.io>

4 days agoFix DSL typo in streams docs (#12152)
Milind Mantri [Tue, 17 May 2022 09:19:26 +0000 (14:49 +0530)] 
Fix DSL typo in streams docs (#12152)

Reviewer: Bruno Cadonna <cadonna@apache.org>

4 days agoKAFKA-13905: Fix failing ServerShutdownTest.testCleanShutdownAfterFailedStartupDueToC...
dengziming [Tue, 17 May 2022 08:31:28 +0000 (16:31 +0800)] 
KAFKA-13905: Fix failing ServerShutdownTest.testCleanShutdownAfterFailedStartupDueToCorruptLogs (#12165)

Reviewers: Jason Gustafson <jason@confluent.io>, Luke Chen <showuon@gmail.com>

4 days agoMINOR: Rename remaining `zkVersion` to `partitionEpoch` in `PartitionTest` (#12147)
David Jacot [Tue, 17 May 2022 06:58:43 +0000 (08:58 +0200)] 
MINOR: Rename remaining `zkVersion` to `partitionEpoch` in `PartitionTest` (#12147)

Reviewers:  Kvicii <42023367+Kvicii@users.noreply.github.com>, dengziming <dengziming1993@gmail.com>, Jason Gustafson <jason@confluent.io>

4 days agoKAFKA-13899: Use INVALID_CONFIG error code consistently in AlterConfig APIs (#12162)
Jason Gustafson [Tue, 17 May 2022 00:41:23 +0000 (17:41 -0700)] 
KAFKA-13899: Use INVALID_CONFIG error code consistently in AlterConfig APIs (#12162)

In the AlterConfigs/IncrementalAlterConfigs zk handler, we return `INVALID_REQUEST` and `INVALID_CONFIG` inconsistently. The problem is in `LogConfig.validate`. We may either return `ConfigException` or `InvalidConfigException`. When the first of these is thrown, we catch it and convert to `INVALID_REQUEST`. If the latter is thrown, then we return `INVALID_CONFIG`. It seems more appropriate to return `INVALID_CONFIG` consistently, which is what the KRaft implementation already does this. This patch fixes this and converts a few integration tests to KRaft.

Reviewers: José Armando García Sancio <jsancio@users.noreply.github.com>

4 days agoMINOR: Clarify impact of num.replica.fetchers (#12153)
Joel Hamill [Mon, 16 May 2022 16:38:06 +0000 (09:38 -0700)] 
MINOR: Clarify impact of num.replica.fetchers (#12153)

The documentation for `num.replica.fetchers` should emphasize the fact that the count applies to each source broker individually. Also mention the tradeoff.

Reviewers: Jason Gustafson <jason@confluent.io>

4 days agoKAFKA-12635: Don't emit checkpoints for partitions without offset-syncs (#11748)
Mickael Maison [Mon, 16 May 2022 15:44:14 +0000 (17:44 +0200)] 
KAFKA-12635: Don't emit checkpoints for partitions without offset-syncs (#11748)

Reviewers: Luke Chen <showuon@gmail.com>,  Viktor Somogyi-Vass <viktorsomogyi@gmail.com>, Dániel Urbán <urb.daniel7@gmail.com>, Federico Valeri <fedevaleri@gmail.com>

5 days agoUpdate note on upgrade from log4j to reload4j (#12164)
xjin-Confluent [Mon, 16 May 2022 11:23:10 +0000 (04:23 -0700)] 
Update note on upgrade from log4j to reload4j (#12164)

Reviewers: Nikhil Bhatia <rite2nikhil@gmail.com>, Bruno Cadonna <cadonna@apache.org>

5 days agoKAFKA-13851: Add integration tests for DeleteRecords API (#12087)
Divij Vaidya [Mon, 16 May 2022 07:50:50 +0000 (10:50 +0300)] 
KAFKA-13851: Add integration tests for DeleteRecords API (#12087)

Reviewers: Luke Chen <showuon@gmail.com>, dengziming <dengziming1993@gmail.com>

5 days agoKAFKA-12703; Allow unencrypted private keys when using PEM files (#11916)
Dejan Maric [Mon, 16 May 2022 07:25:05 +0000 (09:25 +0200)] 
KAFKA-12703; Allow unencrypted private keys when using PEM files (#11916)

Reviewers: David Jacot <djacot@confluent.io>

7 days agoKAFKA-13785: [7/N][Emit final] emit final for sliding window (#12135)
Guozhang Wang [Sat, 14 May 2022 02:29:00 +0000 (19:29 -0700)] 
KAFKA-13785: [7/N][Emit final] emit final for sliding window (#12135)

This is a copy PR of #12037: Implementation to emit final for sliding window agg. This is authored by lihaosky.

Reviewers: Matthias J. Sax <matthias@confluent.io>

7 days agoMINOR: convert some tests to KRaft (#12155)
Colin Patrick McCabe [Sat, 14 May 2022 00:29:47 +0000 (17:29 -0700)] 
MINOR: convert some tests to KRaft (#12155)

Convert EndToEndClusterIdTest, ConsumerGroupCommandTest,
ListConsumerGroupTest, and LogOffsetTest to test KRaft mode.

Reviewers: Jason Gustafson <jason@confluent.io>, dengziming <dengziming1993@gmail.com>

7 days agoMINOR: Add upgrade tests for FK joins (#12122)
Alex Sorokoumov [Sat, 14 May 2022 00:21:27 +0000 (02:21 +0200)] 
MINOR: Add upgrade tests for FK joins (#12122)

Follow up PR for KAFKA-13769.

Reviewers: Matthias J. Sax <matthias@confluent.io>

7 days agoKAFKA-13746: Attempt to fix flaky test by waiting on metadata update (#12104)
vamossagar12 [Sat, 14 May 2022 00:09:47 +0000 (05:39 +0530)] 
KAFKA-13746: Attempt to fix flaky test by waiting on metadata update (#12104)

Reviewers: dengziming <dengziming1993@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

8 days agoMINOR: Update release versions for upgrade tests with 3.1.1 release (#12156)
Tom Bentley [Fri, 13 May 2022 08:32:41 +0000 (09:32 +0100)] 
MINOR: Update release versions for upgrade tests with 3.1.1 release (#12156)

Updates release versions in files that are used for upgrade test with the 3.1.1 release version.

Reviewers: Bruno Cadonna <bruno@confluent.io>

8 days agoKAFKA-13649: Implement early.start.listeners and fix StandardAuthorizer loading ...
Colin Patrick McCabe [Thu, 12 May 2022 21:48:33 +0000 (14:48 -0700)] 
KAFKA-13649: Implement early.start.listeners and fix StandardAuthorizer loading (#11969)

Since the StandardAuthorizer relies on the metadata log to store its ACLs, we need to be sure that
we have the latest metadata before allowing the authorizer to be used. However, if the authorizer
is not usable for controllers in the cluster, the latest metadata cannot be fetched, because
inter-node communication cannot occur. In the initial commit which introduced StandardAuthorizer,
we punted on the loading issue by allowing the authorizer to be used immediately. This commit fixes
that by implementing early.start.listeners as specified in KIP-801. This will allow in superusers
immediately, but throw the new AuthorizerNotReadyException if non-superusers try to use the
authorizer before StandardAuthorizer#completeInitialLoad is called.

For the broker, we call StandardAuthorizer#completeInitialLoad immediately after metadata catch-up
is complete, right before unfencing. For the controller, we call
StandardAuthorizer#completeInitialLoad when the node has caught up to the high water mark of the
cluster metadata partition.

This PR refactors the SocketServer so that it creates the configured acceptors and processors in
its constructor, rather than requiring a call to SocketServer#startup A new function,
SocketServer#enableRequestProcessing, then starts the threads and begins listening on the
configured ports. enableRequestProcessing uses an async model: we will start the acceptor and
processors associated with an endpoint as soon as that endpoint's authorizer future is completed.

Also fix a bug where the controller and listener were sharing an Authorizer when in co-located
mode, which was not intended.

Reviewers: Jason Gustafson <jason@confluent.io>

8 days agoMINOR; DeleteTopics version tests (#12141)
José Armando García Sancio [Thu, 12 May 2022 20:04:48 +0000 (13:04 -0700)] 
MINOR; DeleteTopics version tests (#12141)

Add a DeleteTopics test for all supported versions. Convert the
DeleteTopicsRequestTest to run against both ZK and KRaft mode.

Reviewers: Colin Patrick McCabe <cmccabe@apache.org>, dengziming <dengziming1993@gmail.com>

9 days agoMINOR: Fix flaky testDescribeUnderReplicatedPartitions (#12112)
dengziming [Thu, 12 May 2022 12:55:42 +0000 (20:55 +0800)] 
MINOR: Fix flaky testDescribeUnderReplicatedPartitions (#12112)

Currently, we are waiting for metadataCache to bookkeeper the partition info, this isn't enough, we should wait until the partition ISR is less than AR.

Reviewers: Luke Chen <showuon@gmail.com>, Divij Vaidya <divijvaidya13@gmail.com>

9 days agoMINOR: parameter name fix for maxScalacThreads (#12151)
jparag [Thu, 12 May 2022 06:51:27 +0000 (02:51 -0400)] 
MINOR: parameter name fix for maxScalacThreads (#12151)

There's a typo in build.gradle, and cause the `maxScalacThreads` parameter doesn't work as expected

9 days agoKAFKA-10000: Add all public-facing config properties (#11775)
Chris Egerton [Thu, 12 May 2022 06:45:53 +0000 (02:45 -0400)] 
KAFKA-10000: Add all public-facing config properties (#11775)

Reviewers: Luke Chen <showuon@gmail.com>, Tom Bentley <tbentley@redhat.com>, Andrew Eugene Choi <andrew.choi@uwaterloo.ca>

9 days agoMINOR: Remove kraft authorizer from list of missing features (#12146)
Ismael Juma [Wed, 11 May 2022 16:46:37 +0000 (09:46 -0700)] 
MINOR: Remove kraft authorizer from list of missing features (#12146)

Also tweak the list of missing features a bit

Reviewers: José Armando García Sancio <jsancio@users.noreply.github.com>

9 days agoMINOR: Make TopicPartitionBookkeeper and TopicPartitionEntry top level (#12097)
Ismael Juma [Wed, 11 May 2022 16:30:46 +0000 (09:30 -0700)] 
MINOR: Make TopicPartitionBookkeeper and TopicPartitionEntry top level (#12097)

This is the first step towards refactoring the `TransactionManager` so
that it's easier to understand and test. The high level idea is to push
down behavior to `TopicPartitionEntry` and `TopicPartitionBookkeeper`
and to encapsulate the state so that the mutations can only be done via
the appropriate methods.

Inner classes have no mechanism to limit access from the outer class,
which presents a challenge when mutability is widespread (like we do
here).

As a first step, we make `TopicPartitionBookkeeper` and
`TopicPartitionEntry` top level and rename them and a couple
of methods to make the intended usage clear and avoid
redundancy.

To make the review easier, we don't change anything else
except access changes required for the code to compile.
The next PR will contain the rest of the refactoring.

Reviewers: Jason Gustafson <jason@confluent.io>

10 days agoMINOR: reload4j build dependency fixes (#12144)
Ismael Juma [Wed, 11 May 2022 03:14:21 +0000 (20:14 -0700)] 
MINOR: reload4j build dependency fixes (#12144)

* Replace `log4j` with `reload4j` in `copyDependantLibs`. Since we have
  some projects that have an explicit `reload4j` dependency, it
  was included in the final release release tar - i.e. it was effectively
  a workaround for this bug.
* Exclude `log4j` and `slf4j-log4j12` transitive dependencies for
  `streams:upgrade-system-tests`. Versions 0100 and 0101
  had a transitive dependency to `log4j` and `slf4j-log4j12` via
  `zkclient` and `zookeeper`. This avoids classpath conflicts that lead
  to [NoSuchFieldError](https://github.com/qos-ch/reload4j/issues/41) in
  system tests.

Reviewers: Jason Gustafson <jason@confluent.io>

10 days ago KAFKA-13892: Fix bug where multiple remove records are generated for one ACL
andymg3 [Tue, 10 May 2022 22:26:57 +0000 (18:26 -0400)] 
 KAFKA-13892: Fix bug where multiple remove records are generated for one ACL

Fix a bug where multiple remove records could be generated for a single ACL. Previously, this happened
if the user submitted multiple filters to deleteAcls, and more than one matched.

Reviewers: Colin P. McCabe <cmccabe@apache.org>, Jason Gustafson <jason@confluent.io>

10 days agoMINOR: Create case class to encapsulate fetch parameters and simplify handling (...
Jason Gustafson [Tue, 10 May 2022 20:24:23 +0000 (13:24 -0700)] 
MINOR: Create case class to encapsulate fetch parameters and simplify handling (#12082)

This patch adds a new case class `FetchParams` which encapsulates the parameters of the fetch request. It then uses this class in `DelayedFetch` directly instead of `FetchMetadata`. The intent is to reduce the number of things we need to change whenever we need to pass through new parameters. The patch also cleans up `ReplicaManagerTest` for more consistent usage.

Reviewers: David Jacot <djacot@confluent.io>

10 days agoMINOR: install Exit.exit handler in BrokerMetadataPublisherTest (#12142)
Colin Patrick McCabe [Tue, 10 May 2022 19:44:39 +0000 (12:44 -0700)] 
MINOR: install Exit.exit handler in BrokerMetadataPublisherTest (#12142)

Reviewers: Jason Gustafson <jason@confluent.io>

10 days agoKAFKA-13862; Support Append/Subtract multiple config values in KRaft mode (#12108)
dengziming [Tue, 10 May 2022 19:41:17 +0000 (03:41 +0800)] 
KAFKA-13862; Support Append/Subtract multiple config values in KRaft mode (#12108)

We can append/subtract multiple config values in kraft mode using the `IncrementalAlterConfig` RPC. For example: append/subtract topic config "cleanup.policy" with value="delete,compact" will end up treating "delete,compact" as a value not 2 values. This patch fixes the problem. Additionally, it update the zk logic to correctly handle duplicate additions.

Reviewers: Akhilesh Chaganti <akhileshchg@users.noreply.github.com>, Jason Gustafson <jason@confluent.io>

11 days agoMINOR: Update release versions for upgrade tests with 3.2.0 release (#12143)
Bruno Cadonna [Tue, 10 May 2022 12:47:46 +0000 (14:47 +0200)] 
MINOR: Update release versions for upgrade tests with 3.2.0 release (#12143)

Updates release versions in files that are used for upgrade test with the 3.2.0 release version.

Reviewer: David Jacot <djacot@confluent.io>

11 days agoKAFKA-13879: Reconnect exponential backoff is ineffective in some cases (#12131)
chern [Tue, 10 May 2022 10:36:42 +0000 (03:36 -0700)] 
KAFKA-13879: Reconnect exponential backoff is ineffective in some cases (#12131)

When a client connects to a SSL listener using PLAINTEXT security protocol, after the TCP connection is setup, the client considers the channel setup is complete. In reality the channel setup is not complete yet. The client then resets reconnect exponential backoff and issues API version request. Since the broker expects SSL handshake, the API version request will cause the connection to disconnect. Client reconnects without exponential backoff since it has been reset.

This commit removes the reset of reconnect exponential backoff when sending API version request. In the good case where the channel setup is complete, reconnect exponential backoff will be reset when the node becomes ready, which is after getting the API version response. Inter-broker clients which do not send API version request and go directly to ready state continue to reset backoff before any  successful requests.

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

11 days agoMINOR: Small cleanups in connect/mirror (#12113)
Mickael Maison [Tue, 10 May 2022 06:49:56 +0000 (08:49 +0200)] 
MINOR: Small cleanups in connect/mirror (#12113)

Reviewers: Luke Chen <showuon@gmail.com>, Divij Vaidya <divijvaidya13@gmail.com>

11 days agoMINOR: fix typo for QUORUM_FETCH_TIMEOUT_MS_DOC (#12132)
Jacklee [Tue, 10 May 2022 02:47:23 +0000 (10:47 +0800)] 
MINOR: fix typo for QUORUM_FETCH_TIMEOUT_MS_DOC (#12132)

Reviewers: Luke Chen <showuon@gmail.com>

11 days agoKAFKA-13790; ReplicaManager should be robust to all partition updates from kraft...
David Jacot [Mon, 9 May 2022 18:47:14 +0000 (20:47 +0200)] 
KAFKA-13790; ReplicaManager should be robust to all partition updates from kraft metadata log (#12085)

This patch refactors the `Partition.makeLeader` and `Partition.makeFollower` to be robust to all partition updates from the KRaft metadata log. Particularly, it ensures the following invariants:

- A partition update is accepted if the partition epoch is equal or newer. The partition epoch is updated by the AlterPartition path as well so we accept an update from the metadata log with the same partition epoch in order to fully update the partition state.
- The leader epoch state offset is only updated when the leader epoch is bumped.
- The follower states are only updated when the leader epoch is bumped.
- Fetchers are only restarted when the leader epoch is bumped. This was already the case but this patch adds unit tests to prove/maintain it.

In the mean time, the patch unifies the state change logs to be similar in both ZK and KRaft world.

Reviewers: Jason Gustafson <jason@confluent.io>

11 days agoKAFKA-13793: Add validators for configs that lack validators (#12010)
RivenSun [Mon, 9 May 2022 18:29:17 +0000 (02:29 +0800)] 
KAFKA-13793: Add validators for configs that lack validators (#12010)

Reviewers: Mickael Maison <mickael.maison@gmail.com>, Luke Chen <showuon@gmail.com>, Chris Egerton <fearthecellos@gmail.com>, Christo Lolov <lolovc@amazon.com>, Divij Vaidya <divijvaidya13@gmail.com>

12 days agoKAFKA-13763: Refactor IncrementalCooperativeAssignor for improved unit testing (...
Chris Egerton [Mon, 9 May 2022 13:43:47 +0000 (09:43 -0400)] 
KAFKA-13763: Refactor IncrementalCooperativeAssignor for improved unit testing (#11983)

The goals here include:

1. Create an overloaded variant of the IncrementalCooperativeAssignor::performTaskAssignment method that is more testing friendly
2. Simplify the parameter list for the IncrementalCooperativeAssignor::handleLostAssignments method, which in turn simplifies the logic for testing this class
3. Capture repeated Java 8 streams logic in simple, reusable, easily-verifiable utility methods added to the ConnectUtils class

Reviewers: Luke Chen <showuon@gmail.com>

2 weeks agoKAFKA-10888: Sticky partition leads to uneven produce msg (#12049)
Artem Livshits [Fri, 6 May 2022 18:31:12 +0000 (11:31 -0700)] 
KAFKA-10888: Sticky partition leads to uneven produce msg (#12049)

The design is described in detail in KIP-794
https://cwiki.apache.org/confluence/display/KAFKA/KIP-794%3A+Strictly+Uniform+Sticky+Partitioner.

Implementation notes:

The default partitioning logic is moved to the BuiltInPartitioner class
(there is one object per topic).  The object keeps track of how many
bytes are produced per-partition and once the amount exceeds batch.size,
switches to the next partition (note that partition switch decision is
decoupled from batching).  The object also keeps track of probability
weights that are based on the queue sizes (the larger the queue size
is the less chance for the next partition to be chosen).  The queue
sizes are calculated in the RecordAccumulator in the `ready` method,
the method already enumerates all partitions so we just add some extra
logic into the existing O(N) method.  The partition switch decision may
take O(logN), where N is the number partitions per topic, but it happens
only once per batch.size (and the logic is avoided when all queues are
of equal size).  Produce bytes accounting logic is lock-free.

When partitioner.availability.timeout.ms is non-0, RecordAccumulator
keeps stats on "node latency" which is defined as the difference between
the last time the node had a batch waiting to be send and the last time
the node was ready to take a new batch.  If this difference exceeds
partitioner.availability.timeout.ms we don't switch to that partition
until the node is ready.

Reviewers: Jun Rao <junrao@gmail.com>

2 weeks agoMINOR: Fix RecordContext Javadoc (#12130)
John Roesler [Fri, 6 May 2022 16:31:51 +0000 (11:31 -0500)] 
MINOR: Fix RecordContext Javadoc (#12130)

A prior commit accidentally changed the javadoc for RecordContext.
In reality, it is not reachable from api.Processor, only Processor.

Reviewers: Guozhang Wang <guozhang@apache.org>

2 weeks agoMINOR: Fix link to old doc in quickstart (#12129)
Bruno Cadonna [Fri, 6 May 2022 09:43:30 +0000 (11:43 +0200)] 
MINOR: Fix link to old doc in quickstart (#12129)

In Kafka's quickstart a link points to the 2.5 Kafka Streams demo.
This PR fixes this link.

2 weeks agoKAFKA-13804: Output the reason why broker exit unexpectedly during startup (#12028)
Luke Chen [Fri, 6 May 2022 09:25:33 +0000 (17:25 +0800)] 
KAFKA-13804: Output the reason why broker exit unexpectedly during startup (#12028)

Reviewers: Mickael Maison <mickael.maison@gmail.com>

2 weeks agoKAFKA-10000: Add new source connector APIs related to exactly-once support (KIP-618...
Chris Egerton [Fri, 6 May 2022 09:13:59 +0000 (05:13 -0400)] 
KAFKA-10000: Add new source connector APIs related to exactly-once support (KIP-618) (#11773)

Reviewers: Mickael Maison <mickael.maison@gmail.com>, Reviewers: Tom Bentley <tbentley@redhat.com>, Hector Geraldino <hgeraldino@bloomberg.net>, Andrew Eugene Choi <andrew.choi@uwaterloo.ca>

2 weeks agoKAFKA-13785: [8/N][emit final] time-ordered session store (#12127)
Guozhang Wang [Thu, 5 May 2022 23:09:16 +0000 (16:09 -0700)] 
KAFKA-13785: [8/N][emit final] time-ordered session store (#12127)

Time ordered session store implementation. I introduced AbstractRocksDBTimeOrderedSegmentedBytesStore to make it generic for RocksDBTimeOrderedSessionSegmentedBytesStore and RocksDBTimeOrderedSegmentedBytesStore.

A few minor follow-up changes:

1. Avoid extra byte array allocation for fixed upper/lower range serialization.
2. Rename some class names to be more consistent.

Authored-by: Hao Li <1127478+lihaosky@users.noreply.github.com>
Reviewers: Guozhang Wang <wangguoz@gmail.com.com>, John Roesler <vvcephei@apache.org>

2 weeks agoKAFKA-10199: Implement adding active tasks to the state updater (#12128)
Bruno Cadonna [Thu, 5 May 2022 23:00:35 +0000 (01:00 +0200)] 
KAFKA-10199: Implement adding active tasks to the state updater (#12128)

This PR adds the default implementation of the state updater. The implementation only implements adding active tasks to the state updater.

Reviewers: Guozhang Wang <wangguoz@gmail.com>

2 weeks agoKAFKA-13815: Avoid reinitialization for a replica that is being deleted (#12029)
Lucas Wang [Wed, 4 May 2022 18:41:34 +0000 (11:41 -0700)] 
KAFKA-13815: Avoid reinitialization for a replica that is being deleted (#12029)

This PR tries to avoid the reinitialization of the leader epoch cache
and the partition metadata if the corresponding replica is being deleted.
With this change, the asyncDelete method can run more efficiently,
which means a StopReplica request with many partitions to be deleted can be
processed more quickly.

Reviewers: David Jacot <djacot@confluent.io>, Jun Rao <junrao@gmail.com>

2 weeks agoKAFKA-13861; Fix the validateOnly behavior for CreatePartitions requests in KRaft...
Akhilesh Chaganti [Wed, 4 May 2022 17:31:46 +0000 (10:31 -0700)] 
KAFKA-13861; Fix the validateOnly behavior for CreatePartitions requests in KRaft mode (#12106)

The KRaft implementation of the `CreatePartitions` ignores the `validateOnly` flag in the
request and creates the partitions if the validations are successful. Fixed the behavior
not to create partitions upon validation if the `validateOnly` flag is true.

Reviewers: Divij Vaidya <divijvaidya13@gmail.com>, dengziming <dengziming1993@gmail.com>, Jason Gustafson <jason@confluent.io>

2 weeks agoMINOR: Fix typos in configuration docs (#11874)
Joel Hamill [Wed, 4 May 2022 17:27:14 +0000 (10:27 -0700)] 
MINOR: Fix typos in configuration docs (#11874)

Reviewers: Chris Egerton, Weikang Sun, Andrew Eugene Choi, Luke Chen, Guozhang Wang

2 weeks agoMINOR: Improve docs about how to provide multiple log.dir (#12119)
Joel Hamill [Wed, 4 May 2022 03:15:29 +0000 (20:15 -0700)] 
MINOR: Improve docs about how to provide multiple log.dir (#12119)

Reviewer:  Luke Chen <showuon@gmail.com>

2 weeks agoMINOR: Remove duplicated test cases in MetadataVersionTest (#12116)
dengziming [Wed, 4 May 2022 03:10:39 +0000 (11:10 +0800)] 
MINOR: Remove duplicated test cases in MetadataVersionTest (#12116)

These tests belongs to ApiVersionsResponseTest, and accidentally copied them to MetadataVersionTest when working on #12072.

Reviewer: Luke Chen <showuon@gmail.com>

2 weeks agoKAFKA-13785: [6/N][Emit final] Copy: Emit final for TimeWindowedKStreamImpl (#12100)
Guozhang Wang [Tue, 3 May 2022 16:42:23 +0000 (09:42 -0700)] 
KAFKA-13785: [6/N][Emit final] Copy: Emit final for TimeWindowedKStreamImpl (#12100)

This is a copy PR of #11896, authored by @lihaosky (Hao Li): Initial implementation to emit final for TimeWindowedKStreamImpl. This PR is on top of #12030

Author: Hao Li
Reviewers: John Roesler <vvcephei@apache.org>

2 weeks agoMINOR: Note that slf4j-log4j in version 1.7.35+ should be used (#12114)
Bruno Cadonna [Tue, 3 May 2022 10:24:53 +0000 (12:24 +0200)] 
MINOR: Note that slf4j-log4j in version 1.7.35+ should be used (#12114)

Adds a note to the upgrade notes to use slf4j-log4j version
1.7.35+ [1] or slf4j-reload4j to avoid possible compatibility issues
originating from the logging framework [2].

[1] https://www.slf4j.org/manual.html#swapping
[2] https://www.slf4j.org/codes.html#no_tlm

Reviewer: Ismael Juma <ismael@juma.me.uk>

2 weeks agoKAFKA-13854 Refactor ApiVersion to MetadataVersion (#12072)
Alyssa Huang [Mon, 2 May 2022 23:27:52 +0000 (16:27 -0700)] 
KAFKA-13854 Refactor ApiVersion to MetadataVersion (#12072)

Refactoring ApiVersion to MetadataVersion to support both old IBP versioning and new KRaft versioning (feature flags)
for KIP-778.

IBP versions are now encoded as enum constants and explicitly prefixed w/ IBP_ instead of KAFKA_, and having a
LegacyApiVersion vs DefaultApiVersion was not necessary and replaced with appropriate parsing rules for extracting
the correct shortVersions/versions.

Co-authored-by: David Arthur <mumrah@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>, David Arthur <mumrah@gmail.com>, dengziming <dengziming1993@gmail.com>, Colin P. McCabe <cmccabe@apache.org>

2 weeks agoMINOR: fix ClientQuotasRequestTest (#12107)
Colin Patrick McCabe [Mon, 2 May 2022 16:06:16 +0000 (09:06 -0700)] 
MINOR: fix ClientQuotasRequestTest (#12107)

Fix ClientQuotasRequestTest.testAlterClientQuotasBadIp so that it uses actually unresolvable hostnames.
The previous choices "ip" and "abc-123" are now resolvable.

Reviewers: David Jacot <djacot@confluent.io>, Andrew Choi <andrew.choi@uwaterloo.ca>, Divij Vaidya <divijvaidya13@gmail.com>

2 weeks agoKAFKA-13865: Fix ResponseSendTimeMs metric in RequestChannel is removed twice (#12111)
bozhao12 [Mon, 2 May 2022 09:02:53 +0000 (17:02 +0800)] 
KAFKA-13865: Fix ResponseSendTimeMs metric in RequestChannel is removed twice (#12111)

Fix ResponseSendTimeMs metric in RequestChannel is removed twice

Reviewers: Luke Chen <showuon@gmail.com>

3 weeks agoMINOR: Fix event output inconsistencies in TransactionalMessageCopier (#12098)
Jason Gustafson [Fri, 29 Apr 2022 17:02:25 +0000 (10:02 -0700)] 
MINOR: Fix event output inconsistencies in TransactionalMessageCopier (#12098)

This patch fixes some strangeness and inconsistency in the messages written by `TransactionalMessageCopier` to stdout. Here is a sample of two messages.

Progress message:
```
{"consumed":33000,"stage":"ProcessLoop","totalProcessed":33000,"progress":"copier-0","time":"2022/04/24 05:40:31:649","remaining":333}
```
The `transactionalId` is set to the value of the `progress` key.

And a shutdown message:
```
{"consumed":33333,"shutdown_complete":"copier-0","totalProcessed":33333,"time":"2022/04/24 05:40:31:937","remaining":0}
```
The `transactionalId` this time is set to the `shutdown_complete` key and there is no `stage` key.

In this patch, we change the following:

1. Use a separate key for the `transactionalId`.
2. Drop the `progress` and `shutdown_complete` keys.
3. Use `stage=ShutdownComplete` in the shutdown message.
4. Modify `transactional_message_copier.py` system test service accordingly.

Reviewers: David Arthur <mumrah@gmail.com>

3 weeks agoMINOR: Improve document description in zero-copy (#12099)
RivenSun [Fri, 29 Apr 2022 15:44:39 +0000 (23:44 +0800)] 
MINOR: Improve document description in zero-copy (#12099)

Follow up of #12052 to improve the description.

Reviewer: David Jacot <djacot@confluent.io>

3 weeks agoKAFKA-12380 shutdown Executor in Connect's Worker when closed (#11955)
Rajani Karuturi [Fri, 29 Apr 2022 05:35:25 +0000 (11:05 +0530)] 
KAFKA-12380 shutdown Executor in Connect's Worker when closed (#11955)

When the worker is stopped, it does not shutdown this executor. This PR fixes the issue.

Reviewers: Luke Chen <showuon@gmail.com>

3 weeks agoKAFKA-13794: Fix comparator of inflightBatchesBySequence in TransactionsManager ...
Ismael Juma [Thu, 28 Apr 2022 13:13:23 +0000 (06:13 -0700)] 
KAFKA-13794: Fix comparator of inflightBatchesBySequence in TransactionsManager (round 3) (#12096)

Conceptually, the ordering is defined by the producer id, producer epoch
and the sequence number. This set should generally only have entries
for the same producer id and epoch, but there is one case where
we can have conflicting `remove` calls and hence we add this as
a temporary safe fix.

We'll follow-up with a fix that ensures the original intended invariant.

Reviewers: Jason Gustafson <jason@confluent.io>, David Jacot <djacot@confluent.io>, Luke Chen <showuon@gmail.com>

3 weeks agoMINOR: Fix minor typos in `PartitionChangeBuilder` (#12101)
David Jacot [Thu, 28 Apr 2022 03:19:34 +0000 (05:19 +0200)] 
MINOR: Fix minor typos in `PartitionChangeBuilder` (#12101)

Reviewers: Luke Chen <showuon@gmail.com>

3 weeks agoHOTFIX: Only measure in nano when producer metadata refresh is required (#12102)
Guozhang Wang [Wed, 27 Apr 2022 18:27:54 +0000 (11:27 -0700)] 
HOTFIX: Only measure in nano when producer metadata refresh is required (#12102)

We added the metadata wait time in total blocked time (#11805). But we added it in the critical path of send which is called per-record, whereas metadata refresh only happens rarely. This way the cost of time.nanos becomes unnecessarily significant as we call it twice per record.

This PR moves the call to inside the waitOnMetadata callee and only when we do need to wait for a metadata refresh round-trip (i.e. we are indeed blocking).

Reviewers: Matthias J. Sax <matthias@confluent.io>

3 weeks agoMINOR: fix html generation syntax errors (#12094)
Mike Tobola [Tue, 26 Apr 2022 23:51:12 +0000 (16:51 -0700)] 
MINOR: fix html generation syntax errors (#12094)

The html document generation has some errors in it, specifically related to protocols. The two issues identified and resolved are:

* Missing </tbody> closing tags added
* Invalid usage of a <p> tag as a wrapper element for <table> elements. Changed the <p> tag to be a <div>.

Tested by running ./gradlew siteDocsTar and observing that the output was properly formed.

Reviewers: Guozhang Wang <wangguoz@gmail.com>

3 weeks agoMINOR: Rename `AlterIsrManager` to `AlterPartitionManager` (#12089)
Jason Gustafson [Tue, 26 Apr 2022 16:34:18 +0000 (09:34 -0700)] 
MINOR: Rename `AlterIsrManager` to `AlterPartitionManager` (#12089)

Since we have changed the `AlterIsr` API to `AlterPartition`, it makes sense to rename `AlterIsrManager` as well and some of the associated classes.

Reviewers: dengziming <dengziming1993@gmail.com>, David Jacot <djacot@confluent.io>

3 weeks agoKAFKA-12841: Remove an additional call of onAcknowledgement (#12064)
Philip Nee [Mon, 25 Apr 2022 22:59:45 +0000 (15:59 -0700)] 
KAFKA-12841: Remove an additional call of onAcknowledgement (#12064)

The bug was introduced in #11689 that an additional onAcknowledgement was made using the InterceptorCallback class. This is undesirable since onSendError will attempt to call onAcknowledgement once more.

Reviewers: Jun Rao <junrao@gmail.com>

3 weeks agoMINOR; enable KRaft in ConfigCommandIntegrationTest (#11732)
Alyssa Huang [Mon, 25 Apr 2022 22:11:14 +0000 (15:11 -0700)] 
MINOR; enable KRaft in ConfigCommandIntegrationTest (#11732)

Adding KRaft and ZK params to ConfigCommandIntegrationTest wherever appropriate.

Reviewers: Kvicii <42023367+Kvicii@users.noreply.github.com>, dengziming <dengziming1993@gmail.com>, José Armando García Sancio <jsancio@users.noreply.github.com>

3 weeks agoMINOR: Refactor `kafka.cluster.Replica` (#12081)
David Jacot [Mon, 25 Apr 2022 20:43:32 +0000 (21:43 +0100)] 
MINOR: Refactor `kafka.cluster.Replica` (#12081)

This patch refactors kafka.cluster.Replica, it usages and tests. This is part of the work in KAFKA-13790.

Reviewers: Jason Gustafson <jason@confluent.io>

3 weeks agoMINOR: Change `AlterPartition` validation order in `KafkaController` (#12032)
Jason Gustafson [Mon, 25 Apr 2022 16:37:03 +0000 (09:37 -0700)] 
MINOR: Change `AlterPartition` validation order in `KafkaController` (#12032)

Currently we validate recovery state before checking leader epoch in `KafkaController`. It seems more intuitive to validate leader epoch first since the leader might be working with stale state, which is what we do in KRaft. This patch fixes this and adds a couple additional validations to make the behavior consistent.

Reviewers: José Armando García Sancio <jsancio@users.noreply.github.com>

3 weeks agoMINOR: Using enums for auto.offset.reset configuration (#12077)
RivenSun [Sun, 24 Apr 2022 12:54:44 +0000 (20:54 +0800)] 
MINOR: Using enums for auto.offset.reset configuration (#12077)

Using enums instead of Strings for auto.offset.reset configuration

Reviewers: Divij Vaidya <divijvaidya13@gmail.com>, Luke Chen <showuon@gmail.com

3 weeks agoKAFKA-13834: add test coverage for RecordAccumulatorTest (#12092)
ruanliang [Sun, 24 Apr 2022 09:06:19 +0000 (17:06 +0800)] 
KAFKA-13834: add test coverage for RecordAccumulatorTest (#12092)

Reviewers: Luke Chen <showuon@gmail.com>

3 weeks agoKAFKA-13852: Kafka Acl documentation bug for wildcard '*' (#12090)
Hongten [Sun, 24 Apr 2022 08:50:44 +0000 (16:50 +0800)] 
KAFKA-13852: Kafka Acl documentation bug for wildcard '*' (#12090)

The wildcard * in command without wrapped by single quote will be replaced into the file name under the current folder by bash. So we need to wrap with single quote. Update the doc and command option description.

Reviewers: dengziming <dengziming1993@gmail.com>, Luke Chen <showuon@gmail.com>

4 weeks agoMINOR: Rename `ZkVersion` to `PartitionEpoch` (#12071)
David Jacot [Fri, 22 Apr 2022 18:38:17 +0000 (20:38 +0200)] 
MINOR: Rename `ZkVersion` to `PartitionEpoch` (#12071)

This patch does some initial cleanups in the context of KAFKA-13790. Mainly, it renames `ZkVersion` field to `PartitionEpoch` in the `LeaderAndIsrRequest`, the `LeaderAndIsr` and the `Partition`.

Reviewers: Jason Gustafson <jason@confluent.io>, dengziming <dengziming1993@gmail.com>

4 weeks agoKAFKA-13841: Fix a case where we were unable to place on fenced brokers in KRaft...
Colin Patrick McCabe [Thu, 21 Apr 2022 21:58:02 +0000 (14:58 -0700)] 
KAFKA-13841: Fix a case where we were unable to place on fenced brokers in KRaft mode (#12075)

This PR fixes a case where we were unable to place on fenced brokers In KRaft mode. Specifically,
if we had a broker registration in the metadata log, but no associated heartbeat, previously the
HeartbeatManager would not track the fenced broker. This PR fixes this by adding this logic to the
metadata log replay path in ClusterControlManager.

Reviewers: David Arthur <mumrah@gmail.com>, dengziming <dengziming1993@gmail.com>

4 weeks agoMINOR: revert back to 60s session timeout for static membership test (#11881)
Luke Chen [Thu, 21 Apr 2022 18:51:31 +0000 (02:51 +0800)] 
MINOR: revert back to 60s session timeout for static membership test (#11881)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

4 weeks agoMINOR; Fix partition change record noop check (#12073)
José Armando García Sancio [Thu, 21 Apr 2022 16:05:46 +0000 (09:05 -0700)] 
MINOR; Fix partition change record noop check (#12073)

When LeaderRecoveryState was added to the PartitionChangeRecord, the
check for being a noop was not updated. This commit fixes that and
improves the associated test to avoid this oversight in the future.

Reviewers: Colin Patrick McCabe <cmccabe@apache.org>

4 weeks agoMINOR: Improve Gradle Caching and Fix Deprecations (#12003)
Clay Johnson [Thu, 21 Apr 2022 15:25:20 +0000 (10:25 -0500)] 
MINOR: Improve Gradle Caching and Fix Deprecations (#12003)

* Fix UP-TO-DATE check in `create*VersionFile` tasks

`create*VersionFile` tasks explicitly declared output UP-TO-DATE status
 as being false. This change properly sets the inputs to
`create*VersionFile` tasks to the `commitId` and `version` values and
sets `receiptFile` locally rather than in an extra property.

* Enable output caching for `process*Messages` tasks

`process*Messages` tasks did not have output caching enabled. This
change enables that caching, as well as setting a property name and
RELATIVE path sensitivity.

* Fix existing Gradle deprecations

Replaces `JavaExec#main` with `JavaExec#mainClass`

Replaces `Report#destination` with `Report#outputLocation`

Adds a `generator` configuration to projects that need to resolve
the `generator` project (rather than referencing the runtimeClasspath
of the `generator` project from other project contexts.

Reviewers: Mickael Maison <mickael.maison@gmail.com>


4 weeks agoKAFKA-13834: fix drain batch starving issue (#12066)
ruanliang [Thu, 21 Apr 2022 11:26:55 +0000 (19:26 +0800)] 
KAFKA-13834: fix drain batch starving issue (#12066)

In drainBatchesForOneNode method, there's possibility causing some partitions in a node will never get picked. Fix this issue by maintaining a drainIndex for each node.

Reviewers: Luke Chen <showuon@gmail.com>, RivenSun <91005273+RivenSun2@users.noreply.github.com>

4 weeks agoMINOR: cleanup for postProcessAndValidateIdempotenceConfigs method (#12069)
RivenSun [Thu, 21 Apr 2022 06:34:35 +0000 (14:34 +0800)] 
MINOR: cleanup for postProcessAndValidateIdempotenceConfigs method (#12069)

Reviewers: Luke Chen <showuon@gmail.com>

4 weeks agoHOTFIX: fix broken trunk due to conflicting and overlapping commits (#12074)
Matthias J. Sax [Wed, 20 Apr 2022 21:39:15 +0000 (14:39 -0700)] 
HOTFIX: fix broken trunk due to conflicting and overlapping commits (#12074)

Reviewers: Victoria Xia <victoria.xia@confluent.io>, David Arthur <mumrah@gmail.com>

4 weeks agoKAFKA-13588: consolidate `changelogFor` methods to simplify the generation of interna...
Sayantanu Dey [Wed, 20 Apr 2022 18:39:03 +0000 (00:09 +0530)] 
KAFKA-13588: consolidate `changelogFor` methods to simplify the generation of internal topic names (#11703)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

4 weeks agoKAFKA-13785: [5/N][emit final] cache for time ordered window store (#12030)
Hao Li [Wed, 20 Apr 2022 18:09:13 +0000 (11:09 -0700)] 
KAFKA-13785: [5/N][emit final] cache for time ordered window store (#12030)

A new cache for RocksDBTimeOrderedWindowStore. Need this because RocksDBTimeOrderedWindowStore's key ordering is different from CachingWindowStore which has issues for MergedSortedCacheWindowStoreIterator

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>

4 weeks agoKAFKA-13799: Improve documentation for Kafka zero-copy (#12052)
RivenSun [Wed, 20 Apr 2022 17:31:32 +0000 (01:31 +0800)] 
KAFKA-13799: Improve documentation for Kafka zero-copy (#12052)

Improve documentation for Kafka zero-copy. Kafka combines pagecache and zero-copy to greatly improve message consumption efficiency. But zero-copy only works in PlaintextTransportLayer.

Reviewers: Divij Vaidya <divijvaidya13@gmail.com>, Guozhang Wang <wangguoz@gmail.com>

4 weeks agoMINOR: Scala cleanups in core (#12058)
Mickael Maison [Wed, 20 Apr 2022 13:10:46 +0000 (15:10 +0200)] 
MINOR: Scala cleanups in core (#12058)

Reviewers: Luke Chen <showuon@gmail.com>, Divij Vaidya <diviv@amazon.com>

4 weeks agoKAFKA-10095: Add stricter assertion in LogCleanerManagerTest (#12004)
Divij Vaidya [Wed, 20 Apr 2022 10:45:03 +0000 (12:45 +0200)] 
KAFKA-10095: Add stricter assertion in LogCleanerManagerTest (#12004)

Reviewers: Mickael Maison <mickael.maison@gmail.com>

4 weeks agoKAFKA-13835: Fix two bugs related to dynamic broker configs in KRaft (#12063)
Colin Patrick McCabe [Tue, 19 Apr 2022 20:17:16 +0000 (13:17 -0700)] 
KAFKA-13835: Fix two bugs related to dynamic broker configs in KRaft (#12063)

Fix two bugs related to dynamic broker configs in KRaft. The first bug is that we are calling reloadUpdatedFilesWithoutConfigChange when a topic configuration is changed, but not when a
broker configuration is changed. This is backwards. This function must be called only for broker
configs, and never for topic configs or cluster configs.

The second bug is that there were several configurations such as max.connections which are related
to broker listeners, but which do not involve changing the registered listeners. We should support
these configurations in KRaft. This PR fixes the configuration change validation to support this case.

Reviewers: Jason Gustafson <jason@confluent.io>, Matthew de Detrich <mdedetrich@gmail.com>

4 weeks agoKAFKA-13654: Extend KStream process with new Processor API (#11993)
Jorge Esteban Quilcate Otoya [Tue, 19 Apr 2022 15:29:28 +0000 (16:29 +0100)] 
KAFKA-13654: Extend KStream process with new Processor API (#11993)

Updates the KStream process API to cover the use cases
of both process and transform, and deprecate the KStream transform API.

Implements KIP-820

Reviewer: John Roesler <vvcephei@apache.org>

4 weeks agoKAFKA-13832: Fix flaky testAlterAssignment (#12060)
dengziming [Tue, 19 Apr 2022 07:08:29 +0000 (15:08 +0800)] 
KAFKA-13832: Fix flaky testAlterAssignment (#12060)

In KRaft mode the metadata is not propagate in time, so we should should wait for it before make assertions.

Reviewers:  Luke Chen <showuon@gmail.com>

4 weeks agoKAFKA-8785: fix request timeout by waiting for metadata cache up-to-date (#11681)
Luke Chen [Tue, 19 Apr 2022 06:13:21 +0000 (14:13 +0800)] 
KAFKA-8785: fix request timeout by waiting for metadata cache up-to-date (#11681)

The reason why this test is flaky is because we have race condition at the beginning of the test, when brokers are staring up, and the adminClient is requesting for brokers metadata. Once the adminClient only got partial metadata, the test will fail, because in these tests, brokers will be shutdown to test leader election.

Fix this issue by explicitly waiting for metadata cache up-to-date in waitForReadyBrokers, and let admin client get created after waitForReadyBrokers.

Reviewers: Jason Gustafson <jason@confluent.io>, David Jacot <djacot@confluent.io>, dengziming <dengziming1993@gmail.com>

4 weeks agoMINOR: Fix TestDowngrade.test_upgrade_and_downgrade (#12027)
David Jacot [Mon, 18 Apr 2022 21:22:33 +0000 (23:22 +0200)] 
MINOR: Fix TestDowngrade.test_upgrade_and_downgrade (#12027)

The second validation does not verify the second bounce because the verified producer and the verified consumer are stopped in `self.run_validation`. This means that the second `run_validation` just spit out the same information as the first one. Instead, we should just run the validation at the end.

Reviewers: Jason Gustafson <jason@confluent.io>

4 weeks agoMINOR: Correct spelling errors in KafkaRaftClient (#12061)
RivenSun [Mon, 18 Apr 2022 17:58:57 +0000 (01:58 +0800)] 
MINOR: Correct spelling errors in KafkaRaftClient (#12061)

Correct spelling errors in KafkaRaftClient

Reviewers: José Armando García Sancio <jsancio@users.noreply.github.com>

5 weeks agoKAFKA-13807: Fix incrementalAlterConfig and refactor some things (#12033)
Colin Patrick McCabe [Fri, 15 Apr 2022 23:07:23 +0000 (16:07 -0700)] 
KAFKA-13807: Fix incrementalAlterConfig and refactor some things (#12033)

Ensure that we can set log.flush.interval.ms at the broker or cluster level via
IncrementalAlterConfigs. This was broken by KAFKA-13749, which added log.flush.interval.ms as the
second synonym rather than the first. Add a regression test to DynamicConfigChangeTest.

Create ControllerRequestContext and pass it to every controller API. This gives us a uniform way to
pass through information like the deadline (if there is one) and the Kafka principal which is
making the request (in the future we will want to log this information).

In ControllerApis, enforce a timeout for broker heartbeat requests which is equal to the heartbeat
request interval, to avoid heartbeats piling up on the controller queue. This should have been done
previously, but we overlooked it.

Add a builder for ClusterControlManager and ReplicationControlManager to avoid the need to deal
with a lot of churn (especially in test code) whenever a new constructor parameter gets added for
one of these.

In ControllerConfigurationValidator, create a separate function for when we just want to validate
that a ConfigResource is a valid target for DescribeConfigs. Previously we had been re-using the
validation code for IncrementalAlterConfigs, but this was messy.

Split out the replica placement code into a separate package and reorganize it a bit.

Reviewers: David Arthur <mumrah@gmail.com

5 weeks agoKAFKA-13769: Explicitly route FK join results to correct partitions (#11945)
Aleksandr Sorokoumov [Fri, 15 Apr 2022 18:28:43 +0000 (20:28 +0200)] 
KAFKA-13769: Explicitly route FK join results to correct partitions (#11945)

Prior to this commit FK response sink routed FK results to
SubscriptionResolverJoinProcessorSupplier using the primary key.

There are cases, where this behavior is incorrect. For example,
if KTable key serde differs from the data source serde which might
happen without a key changing operation.

Instead of determining the resolver partition by serializing the PK
this patch includes target partition in SubscriptionWrapper payloads.
Default FK response-sink partitioner extracts the correct partition
from the value and routes the message accordingly.

Reviewers: Matthias J. Sax <matthias@confluent.io>

5 weeks agoMINOR: Make link in quickstart dynamic (#12057)
Bruno Cadonna [Fri, 15 Apr 2022 14:22:16 +0000 (16:22 +0200)] 
MINOR: Make link in quickstart dynamic (#12057)

Reviewer: Matthias J. Sax <mjsax@apache.org>

5 weeks agoMINOR: update comment in LocalLog.replaceSegments() (#12054)
Xiaobing Fang [Fri, 15 Apr 2022 13:00:11 +0000 (21:00 +0800)] 
MINOR: update comment in LocalLog.replaceSegments() (#12054)

Reviewers: Luke Chen <showuon@gmail.com>