Reviewers: Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>

There is a regression in 4.6 that causes
`testAll` to fail:

gradle/gradle#4680

Reviewers: Jason Gustafson <jason@confluent.io>

… merged (#4766)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Manikumar Reddy, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io>

…4762)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

…seReceived events on the controller (#4668)

Reviewed by Jun Rao <junrao@gmail.com>

…sisting state in ZooKeeper (#2708)

Prior to this, there have been instances where invalid data was allowed to be persisted in
ZooKeeper, which causes ClassCastExceptions when a broker is restarted and reads this
type-unsafe data.

Adds basic structural and type validation for the reassignment JSON via
introduction of Scala case classes that map to the expected JSON
structure. Also use the Scala case classes to deserialize the JSON
to avoid duplication.

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Viktor Somogyi <viktor.somogyi@cloudera.com>, Ismael Juma <ismael@juma.me.uk>

In the stream thread capture of the TaskMigration exception, print the task full information in WARN. In other places only log as INFO, plus additional context information.

Reviewers: Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>

Reviewers: Guozhang Wang <wangguoz@gmail.com>

#4808)

KafkaStreams.waitOnState() should check the state to be the given one instead of the hard-coded `NOT_RUNNING`.

Reviewers: Guozhang Wang <wangguoz@gmail.com>

In the group coordinator, we currently check whether the partition is owned before checking whether it is loading. Since loading is a prerequisite for partition ownership, it means that it is not actually possible to see the COORDINATOR_LOAD_IN_PROGRESS error. The impact is mostly harmless: while loading the group, the client may send unnecessary FindCoordinator requests to rediscover the coordinator. I've fixed the bug and restructured the code to enable testing.

In the process of fixing this bug, the following improvements have been made:

1. We now verify valid groupId in all request handlers.
2. Currently if the coordinator is loading when a SyncGroup is received, we'll return NOT_COORDINATOR. I've changed this to return REBALANCE_IN_PROGRESS since the rebalance state will have been lost on coordinator failover. This effectively forces the consumer to rejoin the group, which seems preferable over unnecessarily rediscovering the coordinator. 
3. I added a check for the COORDINATOR_LOAD_IN_PROGRESS handler in SyncGroup. Although we do not currently return this error, it seems reasonable that we might want to some day, so it seems better to get the check in now.

Reviewers: Guozhang Wang <wangguoz@gmail.com>

By allowing `max.connections.per.ip` to be 0, Kafka can support IP-based filtering using `max.connections.per.ip.overrides`.

…ecorded (#4811)

Avoid test hanging when there is a failure by limiting wait time.

…tarballs

#4760 unintentionally included extra raw class files in the release tarballs by making the .class file output (instead of the jar) for a streams a dependency of the streams-test-utils. This fixes that issue by instead breaking the circular dependency by using a `compileOnly`/`provided` dependency on those sources and also including the dependency as a test dependency.

I verified by using `gradlew clean installAll releaseTarGzAll`, then checking that the release tarball doesn't have the extraneous files and the installed pom file has the expected dependencies. The dependency on kafka-streams is now in the `test` scope, but that should be fine since a streams application would only use this dependency if it already had a dependency on streams in `compile` (or in weird edge cases the user could handle specifying the right dependencies). This actually seems to even be an improvement over the previous situation where the actual dependency was not expressed in the pom at all (since the dependency was on the sourceSet output rather than the actual project).

Author: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Matthias J. Sax <mjsax@apache.org>, Guozhang Wang <wangguoz@gmail.com>, John Roesler <vvcephei@users.noreply.github.com>

Closes #4816 from ewencp/fix-streams-dependencies

## Summary of the problem
When the `header.converter` is not specified in the worker config or the connector config, a bug in the `Plugins` test causes it to never instantiate the `HeaderConverter` instance, even though there is a default value.

This is a problem as soon as the connector deals with headers, either in records created by a source connector or in messages on the Kafka topics consumed by a sink connector. As soon as that happens, a NPE occurs.

A workaround is to explicitly set the `header.converter` configuration property, but this was added in AK 1.1 and thus means that upgrading to AK 1.1 will not be backward compatible and will require this configuration change.

## The Changes

The `Plugins.newHeaderConverter` methods were always returning null if the `header.converter` configuration value was not specified in the supplied connector or worker configuration. Thus, even though the `header.converter` property has a default, it was never being used.

The fix was to only check whether a `header.converter` property was specified when the connector configuration was being used, and if no such property exists in the connector configuration to return null. Then, when the worker configuration is being used, the method simply gets the `header.converter` value (or the default if no value was explicitly set).

Also, the ConnectorConfig had the same default value for the `header.converter` property as the WorkerConfig, but this resulted in very confusing log messages that implied the default header converter should be used even when the worker config specified the `header.converter` value. By removing the default, the log messages now make sense, and the Worker still properly instantiates the correct header converter.

Finally, updated comments and added log messages to make it more clear which converters are being used and how they are being converted.

## Testing

Several new unit tests for `Plugins.newHeaderConverter` were added to check the various behavior. Additionally, the runtime JAR with these changes was built and inserted into an AK 1.1 installation, and a source connector was manually tested with 8 different combinations of settings for the `header.converter` configuration:

1. default value
1. worker configuration has `header.converter` explicitly set to the default
1. worker configuration has `header.converter` set to a custom `HeaderConverter` implementation in the same plugin
1. worker configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin
1. connector configuration has `header.converter` explicitly set to the default
1. connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in the same plugin
1. connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin
1. worker configuration has `header.converter` explicitly set to the default, and the connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin

The worker created the correct `HeaderConverter` implementation with the correct configuration in all of these tests.

Finally, the default configuration was used with the aforementioned custom source connector that generated records with headers, and an S3 connector that consumes the records with headers (but didn't do anything with them). This test also passed.

Author: Randall Hauch <rhauch@gmail.com>

Reviewers: Ewen Cheslack-Postava <ewen@confluent.io>

Closes #4815 from rhauch/kafka-6728

Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Ignore headers when down-converting to V0/V1 since they are not supported. Added a test-case to verify down-conversion sanity in presence of headers.

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

Reviewers: Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>, John Roesler <john@confluent.io>

Reviewers: Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

)

The invalid topic name is already handled locally so it is unnecessary to send the DeleteTopicsRequest. This PR adds a count to MockClient for testing.

Reviewers: Colin Patrick McCabe <colin@cmccabe.xyz>, Jason Gustafson <jason@confluent.io>

Implementation of KIP-86. Client, server and login callback handlers have been made configurable for both brokers and clients.

Reviewers: Jun Rao <junrao@gmail.com>, Ron Dagostino <rndgstn@gmail.com>, Manikumar Reddy <manikumar.reddy@gmail.com>

…ponses (#4741)

…ThreadTimeRecorded() test (#4824)

Reviewers: Jason Gustafson <jason@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>

Add a validation check to make sure max.connections.per.ip.overrides is configured when max.connections.per.ip is set zero. Also clean up the config description.

…uate call (#4827)

After the punctuate() call, we would like to double check on the scheduled flag since the call itself may cancel it.

Reviewers: Guozhang Wang <wangguoz@gmail.com>, John Roesler <john@confluent.io>

…ng transaction (#4826)

As Frederic reported on mailing list under the subject "kafka-streams Invalid transition attempted from state READY to state ABORTING_TRANSACTION", producer#abortTransaction should only be called when transactionInFlight is true.

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Matthias J. Sax <matthias@confluent.io>

…#4757)

Currently, WorkerUtils will be able to create topics when there is no security. To be able to work with secure kafka, WorkerUtils.createTopic() needs to be able to take security configs. This PR adds commonClientConf field to both producer bench and roundtrip workload specs so that users can specify security and other common configs once for producer/consumer and adminClient. Also added adminClientConf field to workload specs so that users can specify adminClient specific configs if they want to. For completeness, added consumerConf and producerConf to roundtrip workload spec.

Reviewers: Ismael Juma <ismael@juma.me.uk>, Colin P. Mccabe <cmccabe@confluent.io>, Rajini Sivaram <rajinisivaram@googlemail.com>

Enable quota calculation to be customized using a configurable callback. See KIP-257 for details.

Reviewers: Jun Rao <junrao@gmail.com>

Reviewers: Guozhang Wang <guozhang@confluent.io>, Bill Bejeck <bill@confluent.io>, John Roesler <john@confluent.io>, Damian Guy <damian@confluent.io>

This commit allows secrets in Connect configs to be externalized and replaced with variable references of the form `${provider:[path:]key}`, where the "path" is optional.

There are 2 main additions to `org.apache.kafka.common.config`: a `ConfigProvider` and a `ConfigTransformer`.  The `ConfigProvider` is an interface that allows key-value pairs to be provided by an external source for a given "path".  An a TTL can be associated with the key-value pairs returned from the path.  The `ConfigTransformer` will use instances of `ConfigProvider` to replace variable references in a set of configuration values.

In the Connect framework, `ConfigProvider` classes can be specified in the worker config, and then variable references can be used in the connector config.  In addition, the herder can be configured to restart connectors (or not) based on the TTL returned from a `ConfigProvider`.  The main class that performs restarts and transformations is `WorkerConfigTransformer`.

Finally, a `configs()` method has been added to both `SourceTaskContext` and `SinkTaskContext`.  This allows connectors to get configs with variables replaced by the latest values from instances of `ConfigProvider`.

Most of the other changes in the Connect framework are threading various objects through classes to enable the above functionality.

Author: Robert Yokota <rayokota@gmail.com>
Author: Ewen Cheslack-Postava <me@ewencp.org>

Reviewers: Randall Hauch <rhauch@gmail.com>, Ewen Cheslack-Postava <ewen@confluent.io>

Closes #5068 from rayokota/KAFKA-6886-connect-secrets

This patch contains a few follow-up improvements/cleanup for KIP-266:

- Add upgrade notes
- Add missing `commitSync(Duration)` API
- Improve timeout messages and fix some naming inconsistencies
- Various small cleanups

Reviewers: John Roesler <john@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

implements KIP-268

Reviewers: Bill Bejeck <bill@confluent.io>, John Roesler <john@confluent.io>, Guozhang Wang <guozhang@confluent.io>

…n broker [KIP-283] (#4871)

Implementation for lazy down-conversion in a chunked manner for efficient memory usage during down-conversion. This pull request is mainly to get initial feedback on the direction of the patch. The patch includes all the main components from KIP-283.

Reviewers: Jason Gustafson <jason@confluent.io>

)

In #4919 we propagate the SerDes for each of these aggregation operators.

As @guozhangwang mentioned in that PR:

```
reduce: inherit the key and value serdes from the parent XXImpl class.
count: inherit the key serdes, enforce setting the Serdes.Long() for value serdes.
aggregate: inherit the key serdes, do not set for value serdes internally.
```

Although it's all good for reduce and count, it is quiet unsafe to have aggregate without Materialized given. In fact I don't see why we would not give a Materialized for the aggregate since the result type will always be different (otherwise use reduce) and also the value Serde is simply not propagated.

This has been discussed previously in a broader PR before but I believe for aggregate we could pass implicitly a Materialized the same way we pass a Joined, just to avoid the stupid case. Then if the user wants to specialize, he can give his own Materialized.

Reviewers: Debasish Ghosh <dghosh@acm.org>, Guozhang Wang <guozhang@confluent.io>

Specifying an invalid config (i.e. something other than `CreateTime` or
`LogAppendTime`) via `TopicCommand` would previously cause the
broker to fail on start-up.

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Ismael Juma <ismael@juma.me.uk>

Changes to keep the operation name as is and make the sensor name unique.

Reviewers: John Roesler <john@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

- Override toString in LeaderAndIsrResponse and StopReplicaResponse
- Add unit tests

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

- Removed internal kafka.admin.AdminClient.deleteRecordsBefore since it's
no longer used.
- Removed redundant tests and rewrote non redundant ones to use the Java
AdminClient.

Reviewers: Viktor Somogyi <viktor.somogyi@cloudera.com>, Manikumar Reddy <manikumar.reddy@gmail.com>, Ismael Juma <ismael@juma.me.uk>

…ributes (#5114)

This reverts commit c9ec292 (#5011). That
commit introduces an anonymous inner class which retains a
reference to the non-serializable outer class `KafkaMbean`
breaking Serialization. This means that reading JMX metrics
via JConsole or JmxTool no longer works since RMI relies
on Java Serialization.

Reviewers: Jason Gustafson <jason@confluent.io>, Dong Lin <lindong28@gmail.com>, Ismael Juma <ismael@juma.me.uk>

…1] (#4818)

This patch implements KIP-281, which adds a configurable timeout to the consumer performance tool with a default value of 10 seconds. The old timeout was hard-coded as 1 second. Additionally, this patch adds a warning message when the tool exits after a timeout rather than returning silently.

Reviewers: Dhruvil Shah <dhruvil@confluent.io>, Jason Gustafson <jason@confluent.io>

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

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Dong Lin <lindong28@gmail.com>

Closes #5030 from rajinisivaram/MINOR-group-test

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

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

…failure (#5112)

We added logic to reassign nodes in callToSend after a connection failure, but we do not handle the case when there is no node currently available to reassign the request to. This can happen when using MetadataUpdateNodeIdProvider if all of the known nodes are blacked out awaiting the retry backoff. To fix this, we need to ensure that the call is added to pendingCalls if a new node cannot be found.

Reviewers: Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>

Adding checks on "version" field for tools using it.
This is a new version of the closed PR #3887 (to see for more comments and related discussion).

Author: Paolo Patierno <ppatierno@live.com>

Reviewers: Dong Lin <lindong28@gmail.com>

Closes #5126 from ppatierno/kafka-5919-update

#4919 unintentionally changed the topology naming scheme. This change returns to the prior scheme.

Reviewers: Bill Bejeck <bill@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

…Processor API (#5128)

Reviewers: Guozhang Wang <wangguoz@gmail.com>

)

Make HTTPS the default ssl.endpoint.identification.algorithm.

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

)

PrincipalBuilder implementations can now take the listener into account
when creating the Principal. This is especially interesting in deployments
where inter-broker traffic is on a different listener than client traffic or
when the same protocol is used by multiple listeners.

The change in itself is mostly "plumbing" as the listener name needs to be
passed from ChannelBuilders all the way down to all classes implementing
AuthenticationContext.

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>, Ismael Juma <ismael@juma.me.uk>

Co-authored-by: Edoardo Comar <ecomar@uk.ibm.com>
Co-authored-by: Mickael Maison <mickael.maison@gmail.com>

Reviewers: Bill Bejeck <bill@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>

…es store (#4801)

While using an iterator from IQ, it's possible to get an InvalidStateStoreException if the StreamThread closes the store during a range query.

Added a unit test to SegmentIteratorTest for this condition.

Reviewers: John Roesler <john@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>