Nodes that are materialized should not forward requests to `enableSendingOldValues` to parent nodes, as they themselves can handle fulfilling this request. However, some instances of `KTableProcessorSupplier` were still forwarding requests to parent nodes, which was causing unnecessary materialization of table sources.

The following instances of `KTableProcessorSupplier` have been updated to not forward `enableSendingOldValues` to parent nodes if they themselves are materialized and can handle sending old values downstream:

 * `KTableFilter`
 * `KTableMapValues`
 * `KTableTransformValues`

Other instances of `KTableProcessorSupplier` have not be modified for reasons given below:
 * `KTableSuppressProcessorSupplier`: though it has a `storeName` field, it didn't seem right for this to handle sending old values itself. Its only job is to suppress output.
 * `KTableKTableAbstractJoin`: doesn't have a store name, i.e. it is never materialized, so can't handle the call itself.
 * `KTableKTableJoinMerger`: table-table joins already have materialized sources, which are sending old values. It would be an unnecessary performance hit to have this class do a lookup to retrieve the old value from its store.
 * `KTableReduce`: is always materialized and already handling the call without forwarding
 * `KTableAggregate`: is always materialized and already handling the call without forwarding

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

* rename whitelist/blacklist to include/exclude
* add utility methods to translate deprecated configs

Author: Xavier Léauté <xvrl@apache.org>

Reviewers: Gwen Shapira

Closes #9367 from xvrl/kafka-10570

rename internal classes, methods, and related constants for KIP-629

Author: Xavier Léauté <xvrl@apache.org>

Reviewers: Gwen Shapira

Closes #9400 from xvrl/rename-topic-includelist

cc gwenshap

Author: Xavier Léauté <xvrl@apache.org>

Reviewers: Gwen Shapira

Closes #9405 from xvrl/minor-kip-629-vars

Changes the Connect `ReplaceField` SMT's configuration properties, deprecating and replacing `blacklist` with `exclude`, and `whitelist` with `include`. The old configurations are still allowed (ensuring backward compatibility), but warning messages are written to the log to suggest users change to `include` and `exclude`.

This is part of KIP-629.

Author: Xavier Léauté <xvrl@apache.org>
Reviewer: Randall Hauch <rhauch@gmail.com>

…ected (#9353)

Before `AlterIsr` which was introduced in KIP-497, the controller would register watches in Zookeeper for each reassigning partition so that it could be notified immediately when the ISR was expanded and the reassignment could be completed. This notification is not needed with the latest IBP when `AlterIsr` is enabled because the controller will execute all ISR changes itself.

There is one subtle detail. If we are in the middle of a roll in order to bump the IBP, then it is possible for the controller to be on the latest IBP while some of the brokers are still on the older one. In this case, the brokers on the older IBP will not send `AlterIsr`, but we can still rely on the delayed notification through the `isr_notifications` path to complete reassignments. This seems like a reasonable tradeoff since it should be a short window before the roll is completed.

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

Simple javadoc fix that aligns the properties with the text. 

Reviewers: Konstantine Karantasis <k.karantasis@gmail.com>

#9428)

Fix a bug that was introduced by change 86013dc that resulted in incorrect behavior when
deleting through an iterator.

The bug is that the hash table relies on a denseness invariant... if you remove something,
you might have to move some other things. Calling removeElementAtSlot will do this.
Calling removeFromList is not enough.

Reviewers: Jason Gustafson <jason@confluent.io>

When using an error tracking system, two error log messages result into two different alerts.
It's best to group the logs and have one error with all the information.

For example when using with Sentry, this double line of log.error will create 2 different Issues. One can merge the issues but it will be simpler to have a single error log line.

Signed-off-by: Benoit Maggi <benoit.maggi@gmail.com>

Reviewers: Ewen Cheslack-Postava <me@ewencp.org>, Konstantine Karantasis <k.karantasis@gmail.com>

Fix comment typos.

Reviewers: Boyang Chen <boyang@confluent.io>, Lee Dongjin <dongjin@apache.org>

This release includes a key fix:
* Zinc leaks its dependencies to user classpath (gradle/gradle#14168)

Release notes:
https://docs.gradle.org/6.7/release-notes.html

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>

The leader epoch field is added in version 4, and the auto-generated protocol code would throw unsupported version exception if the field is set to any non-default values for version < 4. This would cause older versioned clients to never receive list-offset results.

Reviewers: Boyang Chen <boyang@confluent.io>

When a coordinator module is being elected / resigned, our log entry is usually associated with a background scheduler on loading / unloading entries and hence it is unclear at the exact time when the election or resignation happens, and we have to then compare with the KafkaAPI's log entry for leaderAndISR / StopReplica to infer the actual time. I think add a couple new log entries indicating the exact time when it happens is helpful.

Reviewers: Boyang Chen <boyang@confluent.io>, Lee Dongjin <dongjin@apache.org>, Bruno Cadonna <bruno@confluent.io>

…ng logic (#9436)

Reviewers: Jun Rao <junrao@gmail.com>

…on (#9149)

Reviewers: Konstantine Karantasis <k.karantasis@gmail.com>

ErrantRecordReporter uses a RetryWithToleranceOperator instance, which is necessarily stateful, having a ProcessingContext of which there's supposed to be one per task. That ProcessingContext is used by both RetryWithToleranceOperator.executeFailed() and execute(), so it's not enough to just synchronize executeFailed().

So make all public methods of RetryWithToleranceOperator synchronized so that RetryWithToleranceOperator is now threadsafe.

Tested with the addition of a multithreaded test case that fails consistently if the methods are not properly synchronized. 

Reviewers: Konstantine Karantasis <k.karantasis@gmail.com>

…ernal topic validation (#9432)

Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>

…signor (#9384)

Currently, we pass multiple object reference (AdminClient,TaskManager, and a few more) into StreamsPartitionAssignor. Furthermore, we (miss)use TaskManager#mainConsumer() to get access to the main consumer (we need to do this, to avoid a cyclic dependency).

This PR unifies how object references are passed into a single ReferenceContainer class to
 - not "miss use" the TaskManager as reference container
 - unify how object references are passes

Note: we need to use a reference container to avoid cyclic dependencies, instead of using a config for each passed reference individually.

Reviewers: John Roesler <john@confluent.io>

…ayedFetch (#9434)

In 2.7, we added lastFetchedEpoch to fetch requests and divergingEpoch to fetch responses. We are not using these for truncation yet, but in order to use these for truncation with IBP 2.7 onwards in the next release, we should make sure that we handle these in all the supporting classes even in 2.7.

Reviewers: Jason Gustafson <jason@confluent.io>

Author: Cyrus Vafadari <cyrus@confluent.io>
Reviewers: Chris Egerton <chrise@confluent.io>, Arjun Satish <arjun@confluent.io>, Randall Hauch <rhauch@gmail.com>

…ues for properties not in connector’s ConfigDef (#9425)

Connect should not always add an error to configuration values in validation results that don't have a `ConfigKey` defined in the connector's `ConfigDef`, and any errors on such configuration values included by the connector should be counted in the total number of errors. Added more unit tests for `AbstractHerder.generateResult(...)`.

Author: Randall Hauch <rhauch@gmail.com>
Reviewer: Konstantine Karantasis <konstantine@confluent.io>

Reviewers: Mickael Maison <mickael.maison@gmail.com>, Ryanne Dolan <ryannedolan@gmail.com>

the field metadata is nullable (see https://github.com/apache/kafka/blob/trunk/clients/src/main/resources/common/message/OffsetFetchResponse.json#L50)

Reviewers: David Jacot <david.jacot@gmail.com>

…ror (#7205)

Reviewers: Stanislav Kozlovski<stanislav_kozlovski@outlook.com>, William Hammond<william.t.hammond@gmail.com>, Chia-Ping Tsai<chia7712@gmail.com>

…tation (#9265)

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

…nt (#9397)

Other than a Stack Overflow comment (see https://stackoverflow.com/a/61738065) by Colin Patrick McCabe and a proposed design note on KIP-117 wiki, there is no source that verifies the thread-safety of KafkaAdminClient.

This patch updates JavaDoc of KafkaAdminClient to clarify its thread-safety.

Reviewers: Tom Bentley <tbentley@redhat.com>, Chia-Ping Tsai <chia7712@gmail.com>

Trigger task reconfiguration when:
- topic-partitions are created or deleted on source cluster
- topic-partitions are missing on target cluster

Authors: Mickael Maison <mickael.maison@gmail.com>, Edoardo Comar <ecomar@uk.ibm.com>
Reviewer: Randall Hauch <rhauch@gmail.com>

…9409)

This PR implements a basic CLI tool for feature versioning system. The KIP-584 write up has been updated to suit this PR. The following is implemented in this PR:

--describe:
Describe supported and finalized features.
Usage: $> ./bin/kafka-features.sh --bootstrap-server host1:port1, host2:port2 --describe [--from-controller] [--command-config <path_to_java_properties_file>]
Optionally, use the --from-controller option to get features from the controller.
--upgrade-all:
Upgrades all features known to the tool to their highest max version levels.
Usage: $> ./bin/kafka-features.sh --bootstrap-server host1:port1, host2:port2 --upgrade-all [--dry-run] [--command-config <path_to_java_properties_file>]
Optionally, use the --dry-run CLI option to preview the feature updates without actually applying them.
--downgrade-all:
Downgrades existing finalized features to the highest max version levels known to this tool.
Usage: $> ./bin/kafka-features.sh --bootstrap-server host1:port1, host2:port2 --downgrade-all [--dry-run] [--command-config <path_to_java_properties_file>].
Optionally, use the --dry-run CLI option to preview the feature updates without actually applying them.

Reviewers: Boyang Chen <boyang@confluent.io>, Jun Rao <junrao@gmail.com>

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Reviewers: Bruno Cadonna <bruno@confluent.io>, Chia-Ping Tsai <chia7712@gmail.com>

Co-authored-by: Chia-Ping Tsai <chia7712@gmail.com>

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

This reverts commit 8a59a22 since it breaks
client configurations like `bootstrap.servers=SASL_PLAINTEXT://localhost:49767`.
A KIP will be submitted to discuss the details and an adjusted change will
be submitted depending on the outcome of that.

… request (#9435)

Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>

… utils (#9714)

Connection id is now only present in `NetworkSend`, which is now
the class used by `Selector`/`NetworkClient`/`KafkaChannel` (which
works well since `NetworkReceive` is the class used for
received data).

The previous `NetworkSend` was also responsible for adding a size
prefix. This logic is already present in `SendBuilder`, but for the
minority of cases where `SendBuilder` is not used (including
a number of tests), we now have `ByteBufferSend.sizePrefixed()`.

With regards to the request/message utilities:
* Renamed `toByteBuffer`/`toBytes` in `MessageUtil` to
`toVersionPrefixedByteBuffer`/`toVersionPrefixedBytes` for clarity.
* Introduced new `MessageUtil.toByteBuffer` that does not include
the version as the prefix.
* Renamed `serializeBody` in `AbstractRequest/Response` to
`serialize` for symmetry with `parse`.
* Introduced `RequestTestUtils` and moved relevant methods from
`TestUtils`.
* Moved `serializeWithHeader` methods that were only used in
tests to `RequestTestUtils`.
* Deleted `MessageTestUtil`.

Finally, a couple of changes to simplify coding patterns:
* Added `flip()` and `buffer()` to `ByteBufferAccessor`.
* Added `MessageSizeAccumulator.sizeExcludingZeroCopy`.
* Used lambdas instead of `TestCondition`.
* Used `Arrays.copyOf` instead of `System.arraycopy` in `MessageUtil`.

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Jason Gustafson <jason@confluent.io>

That makes the behavior of print equal to pyhton2.

Reviewers: Guozhang Wang <wangguoz@gmail.com>

When there are no pending operations, the raft IO thread can block indefinitely waiting for a network event. We rely on asynchronous wakeups in order to break the blocking wait in order to respond to a scheduled append. The current logic already does this, but only for the case when the linger time has been completed during the call to `scheduleAppend`. It is possible instead that after making one call to `scheduleAppend` to start the linger timer, the application does not do any additional appends. In this case, we still need the IO thread to wakeup when the linger timer expires. This patch fixes the problem by ensuring that the IO thread gets woken up after the first append which begins the linger timer.

Reviewers: Guozhang Wang <wangguoz@gmail.com>

…edTest.test_bounce) (#9673)

In Python 3, `filter` functions return iterators rather than `list` so it can traverse only once. Hence, the following loop will only see "empty" and then validation fails.

```python
        src_messages = self.source.committed_messages() # return iterator
        sink_messages = self.sink.flushed_messages()) # return iterator
        for task in range(num_tasks):
            # only first task can "see" the result. following tasks see empty result
            src_seqnos = [msg['seqno'] for msg in src_messages if msg['task'] == task]
```

Reference: https://portingguide.readthedocs.io/en/latest/iterators.html#new-behavior-of-map-and-filter.

Reviewers: Jason Gustafson <jason@confluent.io>

FileChannel.close() does not guarantee modified buffer would be written on the file system. We are changing  it with force() semantics to enforce file buffer and metadata written to filesystem (FileChannel.force(true) updates buffer and metadata).

Reviewers: Jun Rao <junrao@gmail.com>, Jason Gustafson <jason@confluent.io>

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

As suggested, ensure InvalidProducerEpoch gets caught properly on stream side.

Reviewers: Guozhang Wang <wangguoz@gmail.com>, A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>

…o support IP connection rate quota (KIP-612) (#9628)

This PR adds support for IP entities to the `DescribeClientQuotas` and `AlterClientQuotas` APIs. This PR also adds support for describing/altering IP quotas via `kafka-configs` tooling.

Reviewers: Brian Byrne <bbyrne@confluent.io>, Anna Povzner <anna@confluent.io>, David Jacot <djacot@confluent.io>

This PR adds the IP throttling metric as described in KIP-612.

Reviewers: Anna Povzner <anna@confluent.io>, David Jacot <djacot@confluent.io>

#9639)

This patch adds logic to complete fetches immediately after resigning by returning the BROKER_NOT_AVAILABLE error. This ensures that the new election cannot be delayed by fetches which are stuck in purgatory. 

Reviewers: Jason Gustafson <jason@confluent.io>

Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Guozhang Wang <guozhang@confluent.io>

Also updated the jmh readme to make it easier for new people to know
what's possible and best practices.

There were some changes in the generated benchmarking code that
required adjusting `spotbugs-exclude.xml` and for a `javac` warning
to be suppressed for the benchmarking module. I took the chance
to make the spotbugs exclusion mode maintainable via a regex
pattern.

Tested the commands on Linux and macOS with zsh.

JMH highlights:

* async-profiler integration. Can be used with -prof async,
pass -prof async:help to look for the accepted options.
* perf c2c [2] integration. Can be used with -prof perfc2c,
if available.
* JFR profiler integration. Can be used with -prof jfr, pass
-prof jfr:help to look for the accepted options.

Full details:
* 1.24: https://mail.openjdk.java.net/pipermail/jmh-dev/2020-August/002982.html
* 1.25: https://mail.openjdk.java.net/pipermail/jmh-dev/2020-August/002987.html
* 1.26: https://mail.openjdk.java.net/pipermail/jmh-dev/2020-October/003024.html
* 1.27: https://mail.openjdk.java.net/pipermail/jmh-dev/2020-December/003096.html

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Chia-Ping Tsai <chia7712@gmail.com>, Bill Bejeck <bbejeck@gmail.com>, Lucas Bradstreet <lucasbradstreet@gmail.com>

StreamThreads can now be replaced in the streams uncaught exception handler

Reviewers: Bruno Cadonna <bruno@confluent.io>, John Roesler <vvcephei@apache.org>, Leah Thomas <lthomas@confluent.io>

… when updating producers (#9718)

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>

This was missed in 6f27bb0.

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

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

Reviewers: Chia-Ping Tsai <chia7712@gmail.com>

This reverts commit 8577cea.

Set it as a cluster action and update the handler in KafkaApis. We keep the `throttleTimeMs` field
since we intend to enable throttling in the future (especially relevant when we switch to the
built-in quorum mode).

Reviewers: David Arthur <mumrah@gmail.com>

This patch updates the request logger to output request and response payloads in JSON. Payloads are converted to JSON based on their auto-generated schema.

Reviewers:  Lucas Bradstreet <lucas@confluent.io>, David Mao <dmao@confluent.io>, David Jacot <djacot@confluent.io>

Reviewers: David Jacot <djacot@confluent.io>

…ation (#9661)

See https://cwiki.apache.org/confluence/display/KAFKA/KIP-272%3A+Add+API+version+tag+to+broker%27s+RequestsPerSec+metric

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