Reviewers: Luke Chen <[email protected]>

In comparator, objects that are not equal need to have a stable order otherwise, binary search may not find the objects. Improve the producer batch comparator

…12017)

With KAFKA-13527 / KIP-784 we introduced a new top-level error code for
the DescribeLogDirs API for versions 3 and above. However, the change
regressed the error handling for versions less than 3 since the response
converter fails to write the non-zero error code out (rightly) for
versions lower than 3 and drops the response to the client which
eventually times out instead of receiving an empty log dirs response and
processing that as a Cluster Auth failure.

With this change, the API conditionally propagates the error code out to
the client if the request API version is 3 and above. This keeps the
semantics of the error handling the same for all versions and restores
the behavior for older versions.

See current behavior in the broker log:
```bash
ERROR] 2022-04-08 01:22:56,406 [data-plane-kafka-request-handler-10] kafka.server.KafkaApis - [KafkaApi-0] Unexpected error handling request RequestHeader(apiKey=DESCRIBE_LOG_DIRS, apiVersion=0, clientId=sarama, correlationId=1) -- DescribeLogDirsRequestData(topics=null)
org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default errorCode at version 0
[ERROR] 2022-04-08 01:22:56,407 [data-plane-kafka-request-handler-10] kafka.server.KafkaRequestHandler - [Kafka Request Handler 10 on Broker 0], Exception when handling request
org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default errorCode at version 0
```

Reviewers: Ismael Juma <[email protected]>

This regressed in ca375d8 due to a typo. We need tests
for our builds. :)

I verified that passing the commitId via `-PcommitId=123`
works correctly.

Reviewers: Ismael Juma <[email protected]>

Reviewers: Luke Chen <[email protected]>

…or dynamically configured reporters (#11998)

MetricsReporter.contextChange contract states the method should always be called first before MetricsReporter.init is called. This is done correctly for reporters enabled by default (e.g. JmxReporter) but not for metrics reporters configured dynamically.

This fixes the call ordering for dynamically configured metrics reporter and updates tests to enforce ordering.

Reviewers: David Jacot <[email protected]>

Reviewers: Luke Chen <[email protected]>

#11842)

This PR allows to limit the output batches while they are inspected via the kafka-dump-log.sh script.

The idea is to take samples from the logsegments without affecting a production cluster as the current script will read the whole files, this could create issues related to performance.

Please see the KIP-824

Reviewers: Jun Rao <[email protected]>

The test expects that the connection becomes idle before the mock time is moved forward, but the processor thread runs concurrently and may run some activity on the connection after the mock time is moved forward, thus the connection never expires.

The solution is to wait until the message is received on the socket, and only then wait until the connection is unmuted (it's not enough to wait for unmuted without waiting for message being received on the socket, because the channel might have not been muted yet).

Reviewers: David Jacot <[email protected]>

* gradle: 7.3.3 -> 7.4.2
  Configuration cache improvements and several other improvements.
  https://docs.gradle.org/7.4.2/release-notes.html
* dependencycheck gradle plugin: 6.5.3 -> 7.0.3
  Minor fixes.
* spotbugs gradle plugin: 5.0.5 -> 5.0.6
  Minor fixes.
  https://github.com/spotbugs/spotbugs-gradle-plugin/releases/tag/5.0.6
* jmh: 1.34 -> 1.35
  Fixes and profiler improvements.
  https://mail.openjdk.java.net/pipermail/jmh-dev/2022-March/003422.html
* jqwik: 1.6.3 -> 1.6.5
  Various tweaks and some breaking changes that don't seem to affect us.
  https://github.com/jlink/jqwik/releases/tag/1.6.4
  https://github.com/jlink/jqwik/releases/tag/1.6.5
* mockito: 4.3.1 -> 4.4.0
  Add feature to verify static methods calls in order and minor fixes/improvements.
  https://github.com/mockito/mockito/releases/tag/v4.4.0

Reviewers: Manikumar Reddy <[email protected]>

Reviewers: Mickael Maison <[email protected]>, Bruno Cadonna <[email protected]>

…ble (#11997)

As part of KIP-376 we had ConsumerInterceptor implement AutoCloseable
but forgot to do the same for ProducerInterceptor. This fixes the
inconsistency and also addresses KAFKA-6204 at the same time.

Reviewers: John Roesler <[email protected]>

…entalCooperativeAssignor (#11974)


Reviewers: Mickael Maison <[email protected]>

…ctionManager` (#11991)

Fixes a bug in the comparator used to sort producer inflight batches for a topic partition. This can cause batches in the map `inflightBatchesBySequence` to be removed incorrectly: i.e. one batch may be removed by another batch with the same sequence number. This leads to an `IllegalStateException` when the inflight request finally returns. This patch fixes the comparator to check equality of the `ProducerBatch` instances if the base sequences match.

Reviewers: Jason Gustafson <[email protected]>

…replica selection (#11965)

The current preferred read replica selection logic relies on `partition.leaderReplicaIdOpt` to determine if the selection must be run. The issue is that `partition.leaderReplicaIdOpt` is defined for both the leader and the followers thus the logic is ran all the time. The impact is not too bad as the leader is selected most of the time when the logic is ran by the follower and the leader is filtered out. However there are cases where the selection on a follower could redirect the consumer to another follower under certain rare conditions. For instance with the `RackAwareReplicaSelector `, the follower must have stale replica states from a previous leadership and must have other followers in the same rack for instance. Other implementation of the selection logic could be more impacted.

This patch ensures that the preferred read replica selection is only ran by the leader.

Reviewers: David Jacot <[email protected]>

…l batch (#11995)

Fixes a regression introduced in #11452. Following [KIP-480](https://cwiki.apache.org/confluence/display/KAFKA/KIP-480%3A+Sticky+Partitioner), the `Partitioner` will receive a callback when a batch has been completed so that it can choose another partition. Because of this, we have to wait until the batch has been successfully appended to the accumulator before adding the partition in `TransactionManager.maybeAddPartition`. This is still safe because the `Sender` cannot dequeue a batch from the accumulator until it has been added to the transaction successfully.

Reviewers: Artem Livshits <[email protected]>, David Jacot <[email protected]>,  Tom Bentley <[email protected]>

…` and `forgottenTopics` (#11981)

Fix FetchResponse#`fetchData` and `forgottenTopics`: Assignment of lazy-initialized members should be the last step with double-checked locking

Reviewers: Luke Chen <[email protected]>

…11992)


Reviewers: Mickael Maison <[email protected]>

Reviewers: David Jacot <[email protected]>

Reviewers: David Jacot <[email protected]>

Reviewers: David Jacot <[email protected]>

Previously, when in KRaft mode, CreateTopics did not return the active configurations for the
topic(s) it had just created. This PR addresses that gap. We will now return these topic
configuration(s) when the user has DESCRIBE_CONFIGS permission. (In the case where the user does
not have this permission, we will omit the configurations and set TopicErrorCode. We will also omit
the number of partitions and replication factor data as well.)

For historical reasons, we use different names to refer to each topic configuration when it is set
in the broker context, as opposed to the topic context. For example, the topic configuration
"segment.ms" corresponds to the broker configuration "log.roll.ms". Additionally, some broker
configurations have synonyms. For example, the broker configuration "log.roll.hours" can be used to
set the log roll time instead of "log.roll.ms". In order to track all of this, this PR adds a
table in LogConfig.scala which maps each topic configuration to an ordered list of ConfigSynonym
classes. (This table is then passed to KafkaConfigSchema as a constructor argument.)

Some synonyms require transformations. For example, in order to convert from "log.roll.hours" to
"segment.ms", we must convert hours to milliseconds. (Note that our assumption right now is that
topic configurations do not have synonyms, only broker configurations. If this changes, we will
need to add some logic to handle it.)

This PR makes the 8-argument constructor for ConfigEntry public. We need this in order to make full
use of ConfigEntry outside of the admin namespace. This change is probably inevitable in general
since otherwise we cannot easily test the output from various admin APIs in junit tests outside the
admin package.

Testing:

This PR adds PlaintextAdminIntegrationTest#testCreateTopicsReturnsConfigs. This test validates
some of the configurations that it gets back from the call to CreateTopics, rather than just checking
if it got back a non-empty map like some of the existing tests. In order to test the
configuration override logic, testCreateDeleteTopics now sets up some custom static and dynamic
configurations.

In QuorumTestHarness, we now allow tests to configure what the ID of the controller should be. This
allows us to set dynamic configurations for the controller in testCreateDeleteTopics. We will have
a more complete fix for setting dynamic configuations on the controller later.

This PR changes ConfigurationControlManager so that it is created via a Builder. This will make it
easier to add more parameters to its constructor without having to update every piece of test code
that uses it. It will also make the test code easier to read.

Reviewers: David Arthur <[email protected]>

Add a note in `control.plane.listener.name` doc to mention the value can't be identical with `inter.broker.listener.name`.

Reviewers: Luke Chen <[email protected]>

`validateChars` and `BaseEnum` are used in old version of clients. Remove them.

Reviewers: Luke Chen <[email protected]>

…of active segment (#11950)

Reviewers: Kowshik Prakasam <[email protected]>, Jun Rao <[email protected]>

Part of KIP-825

Reviewers: Matthias J. Sax <[email protected]>

…11928)

Fix upperbound for sliding window, making it compatible with no grace period (kafka-13739)

Added unit test for early sliding window and "normal" sliding window for both events within one time difference (small input) and above window time difference (large input).

Fixing this window interval may slightly change stream behavior but probability to happen is extremely slow and may not have a huge impact on the result given.

Reviewers Leah Thomas <[email protected]>, Bill Bejeck <[email protected]>

…cher thread pool is resized (#11953)

Partitions are assigned to fetcher threads based on their hash modulo the number of fetcher threads. When we resize the fetcher thread pool, we basically re-distribute all the partitions based on the new fetcher thread pool size. The issue is that the logic that resizes the fetcher thread pool updates the `fetcherThreadMap` while iterating over it. The `Map` does not give any guarantee in this case - especially when the underlying map is re-hashed - and that led to not iterating over all the fetcher threads during the process and thus in leaving some partitions in the wrong fetcher threads.

Reviewers: Luke Chen <[email protected]>, David Jacot <[email protected]>

…upRequest (#11971)

KIP-800 introduced a mechanism to pass a reason in the join group request and in the leave group request. A default reason is used unless one is provided by the user. In this case, the custom reason is prefixed by the default one.

When we tried to used this in Kafka Streams, we noted a significant degradation of the performances, see #11873. It is not clear wether the prefixing is the root cause of the issue or not. To be on the safe side, I think that we should remove the prefixing. It does not bring much anyway as we are still able to distinguish a custom reason from the default one on the broker side.

This patch removes prefixing the user provided reasons. So if a the user provides a reason, the reason is used directly. If the reason is empty or null, the default reason is used.

Reviewers: Luke Chen <[email protected]>, <[email protected]>, Hao Li <[email protected]>

In KIP-815 we replaced KafkaConsumer with AdminClient in GetOffsetShell. In the previous implementation, partitions were just ignored if there is no offset for them, however, we will print -1 instead now, This PR fix this inconsistency.

Reviewers: David Jacot <[email protected]>, Luke Chen <[email protected]>

Reviewers: Luke Chen <[email protected]>

…issue (#11963)

In Fix FetchResponse#responseData, we did a double-checked lock for the responseData, but the assignment of lazy-initialized object(responseData) didn't assign in the last step, which would let other threads get the partial object. 

Reviewers: David Jacot <[email protected]>, Luke Chen <[email protected]>

…11959)

Since the topology-level cache size config only controls whether we disable the caching layer entirely for that topology, setting it to anything other than 0 has no effect. The actual cache memory is still just split evenly between the threads, and shared by all topologies.

It's possible we'll want to change this in the future, but for now we should make sure to log a warning so that users who do try to set this override to some nonzero value are made aware that it doesn't work like this.

Also includes some minor refactoring plus a fix for an off-by-one error in #11796

Reviewers: Luke Chen <[email protected]>, Walker Carlson <[email protected]>, Sagar Rao <[email protected]>

With major server components like the new quorum controller being moved outside of the `core` module, it is useful to have shared dependencies moved into `server-common`. An example of this is Yammer metrics which server components still rely heavily upon. All server components should have access to the default registry used by the broker so that new metrics can be registered and metric naming conventions should be standardized. This is particularly important in KRaft where we are attempting to recreate identically named metrics in the controller context. This patch takes a step in this direction. It moves `KafkaYammerMetrics` into `server-common` and it implements
standard metric naming utilities there. 

Reviewers: Manikumar Reddy <[email protected]>

…PATH and plugin.path by default (#11908)

With this change we stop including the non-production grade connectors that are meant to be used for demos and quick starts by default in the CLASSPATH and plugin.path of Connect deployments. The package of these connector will still be shipped with the Apache Kafka distribution and will be available for explicit inclusion. 

The changes have been tested through the system tests and the existing unit and integration tests. 

Reviewers: Mickael Maison <[email protected]>, Randall Hauch <[email protected]>