Adds a new listener config `max.connections` to limit the number of active connections on each listener. The config may be prefixed with listener prefix. This limit may be dynamically reconfigured without restarting the broker.

This is one of the PRs for KIP-402 (https://cwiki.apache.org/confluence/display/KAFKA/KIP-402%3A+Improve+fairness+in+SocketServer+processors). Note that this is currently built on top of PR #6022

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Gwen Shapira <cshapi@gmail.com>

Closes #6034 from rajinisivaram/KAFKA-7730-max-connections

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

Just a minor cleanup to use Java 8 lambdas vs anonymous classes in this test.

I ran all tests in the streams test suite

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

When unclean leader election is enabled dynamically on brokers, we notify controller of the update before updating KafkaConfig. When processing this event, controller's decision to elect unclean leaders is based on the current KafkaConfig, so there is a small timing window when the controller may not elect unclean leader because KafkaConfig of the server was not yet updated. The commit fixes this timing window by using the existing BrokerReconfigurable interface used by other classes which rely on the current value of KafkaConfig.

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

To keep align with the way it handles the offset topic, TopicCommand should not be able to alter transaction topic partition count.

Author: huxihx <huxi_2b@hotmail.com>

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

Closes #6109 from huxihx/KAFKA-7801

DynamicBrokerReconfigurationTest.testAddRemoveSaslListeners removes a listener, waits for the config to be propagated to all brokers and then validates that connections to the removed listener fail. But there is a small timing window between config update and Processor shutdown. Before validating that connections to a removed listener fail, this commit waits for all metrics of the removed listener to be deleted, ensuring that the Processors of the listener have been shutdown.

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

- Update delete topics zk path  from /admin/delete_topic to /admin/delete_topics in assertion error
messages

Author: huxihx <huxi_2b@hotmail.com>

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

Closes #6422 from huxihx/delete-topics

-  Use automatic RPC generation in Metadata Request/Response classes
-  https://cwiki.apache.org/confluence/display/KAFKA/KIP-430+-+Return+Authorized+Operations+in+Describe+Responses

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

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #6352 from omkreddy/KIP-430-METADATA

*Handle OptionException while parsing options when using console producer and print usage before die.*

Author: Suman BN <sumannewton@gmail.com>

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

Closes #6386 from sumannewton/console-producer-parse-printusage

After the 2.1 release, if the broker hasn't been upgrade to the latest inter-broker protocol version, the committed offsets stored in the __consumer_offset topic will get cleaned up way earlier than it should be when the offsets are loaded back from the __consumer_offset topic in GroupCoordinator, which will happen during leadership transition or after broker bounce. This patch fixes the bug by setting expireTimestamp to None if it is the default value after loading v1 offset records from __consumer_offsets.

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

Test currently checks that there were at least 5 polls when 5 connections are established with connectionQueueSize=1. But we could be doing the check just after the 5th connection before the 5th poll, so updated the check to verify that there were at least 4 polls.

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

Metadata may be updated from the background thread, so we need to protect access to SubscriptionState. This patch restructures the metadata handling so that we only check pattern subscriptions in the foreground. Additionally, it improves the following:

1. SubscriptionState is now the source of truth for the topics that will be fetched. We had a lot of messy logic previously to try and keep the the topic set in Metadata consistent with the subscription, so this simplifies the logic.
2. The metadata needs for the producer and consumer are quite different, so it made sense to separate the custom logic into separate extensions of Metadata. For example, only the producer requires topic expiration.
3. We've always had an edge case in which a metadata change with an inflight request may cause us to effectively miss an expected update. This patch implements a separate version inside Metadata which is bumped when the needed topics changes.
4. This patch removes the MetadataListener, which was the cause of https://issues.apache.org/jira/browse/KAFKA-7764. 

Reviewers: David Arthur <mumrah@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>

Added additional description for the "time" parameter for GetOffsetShell which adds " No offset is returned if timestamp provided is greater than recently committed record timestamp." in the description.

Author: KartikVK <karthikkalaghatgi123@gmail.com>

Reviewers: huxi <huxi_2b@hotmail.com>, Manikumar Reddy <manikumar.reddy@gmail.com>

Closes #6357 from Kartikvk1996/kartik-branch

KAFKA-8018: Flaky Test SaslSslAdminClientIntegrationTest#testLegacyAclOpsNeverAffectOrReturnPrefixed

Disable forceSync in EmbeddedZookeeper.
Increase ZK tick to allow longer maxSessionTimeout in tests.
Increase ZK client session timeout in tests.
Handle transient ZK session expiration exception in test utils for createTopic.

Author: Jun Rao <junrao@gmail.com>

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

Closes #6354 from junrao/KAFKA-8018

Stack trace generated from the test failure shows that test failed even though threads were runnable and making progress, indicating that the timeout may be too small when test machine is slow. Increasing timeout from 10 to 15 seconds, consistent with the default wait in other tests. Thread dump also showed a lot of left over threads from other tests, so added clean up of those as well.

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

Ensure that controller is not shutdown in the test.

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

Port 22 is used by ssh, which causes the AdminClient to throw an OOM:

> java.lang.OutOfMemoryError: Java heap space
> 	at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
> 	at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
> 	at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30)
> 	at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:112)
> 	at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:424)
> 	at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:385)
> 	at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:640)
> 	at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:561)
> 	at org.apache.kafka.common.network.Selector.poll(Selector.java:472)
> 	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:535)
> 	at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1140)
> 	at java.lang.Thread.run(Thread.java:748)
>
>

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

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

Closes #6360 from omkreddy/KAFKA-7312

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

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

Closes #6324 from mimaison/sasl-authenticate

This patch fixes a bug in log dir reassignment where Partition.maybeReplaceCurrentWithFutureReplica would compare the entire LogEndOffsetMetadata of each replica to determine whether the reassignment has completed. If the active segments of both replicas have different base segments (for example, if the current replica had previously been cleaned and the future replica rolled segments at different points), the reassignment will never complete. The fix is to compare only the LogEndOffsetMetadata.messageOffset for each replica. Tested with a unit test that simulates the compacted current replica case.

Reviewers: Anna Povzner <anna@confluent.io>, Jason Gustafson <jason@confluent.io>

-  Use automatic RPC generation in DescribeGroups Request/Response classes

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

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

Closes #6322 from omkreddy/KIP-430-Return-Ops

This patch fixes a regression in the replica fetcher which occurs when the replica fetcher manager simultaneously calls `removeFetcherForPartitions`, removing the corresponding partitionStates, while a replica fetcher thread attempts to truncate the same partition(s) in `truncateToHighWatermark`. This causes an NPE which causes the fetcher to crash.

This change simply checks that the `partitionState` is not null first. Note that a similar guard exists in `truncateToEpochEndOffsets`.

Reviewers: Stanislav Kozlovski <stanislav_kozlovski@outlook.com>, Jason Gustafson <jason@confluent.io>

Reviewers: Jason Gustafson <jason@confluent.io>

This patch adds several new log messages to provide more information about errors during log dir movement and to make it clear when each partition movement is finished. 

Reviewers: Jason Gustafson <jason@confluent.io>

KAFKA-7956 In ShutdownableThread, immediately complete the shutdown if the thread has not been started (#6218)

In some test cases it's desirable to instantiate a subclass of `ShutdownableThread` without starting it. Since most subclasses of `ShutdownableThread` put cleanup logic in `ShutdownableThread.shutdown()`, being able to call `shutdown()` on the non-running thread would be useful.

This change allows us to avoid blocking in `ShutdownableThread.shutdown()` if the thread's `run()` method has not been called. We also add a check that `initiateShutdown()` was called before `awaitShutdown()`, to protect against the case where a user calls `awaitShutdown()` before the thread has been started, and unexpectedly is not blocked on the thread shutting down.

Reviewers : Dhruvil Shah <dhruvil@confluent.io>, Jun Rao <junrao@gmail.com>

In order to debug problems with log directory reassignments, it is helpful to know when the fetcher thread begins moving a particular partition. This patch refactors the fetch logic so that we stick to a selected partition as long as it is available and log a message when a different partition is selected.

Reviewers: Viktor Somogyi-Vass <viktorsomogyi@gmail.com>, Dong Lin <lindong28@gmail.com>, Jun Rao <junrao@gmail.com>

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

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

Closes #6301 from mimaison/sasl-handshake

Reviewers: Sriharsha Chintalapani <sriharsha@apache.org>, Jun Rao <junrao@gmail.com>

Currently, commitTransaction and abortTransaction wait indefinitely for the respective operation to be completed. This patch uses the producer's max block time to limit the time that we will wait. If the timeout elapses, we raise a TimeoutException, which allows the user to either close the producer or retry the operation.

Reviewers: Guozhang Wang <wangguoz@gmail.com>, Jason Gustafson <jason@confluent.io>

Use of `MetadataRequest.isAllTopics` is not consistently defined for all versions of the api. For v0, it evaluates to false. This patch makes the behavior consistent for all versions.

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>

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

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

KAFKA-7283: Enable lazy mmap on index files and skip sanity check for segments below recovery point (#5498)

Per the KIP-263 discussion, we think we can improve broker restart time by avoiding performing costly disk operations when sanity checking index files for segments below recovery point on broker startup.

This PR includes the following changes:

1. Mmap the index file and populate fields of the index file on-demand rather than performing costly disk operations when creating the index object on broker startup.
2. Skip sanity checks on the time index and offset index of segments.
   
   1. For segment with offset below the flushed point (recovery point), these segments are safely flushed so we don't need to sanity check the index files. if there are indeed data corruption on disk, given that we don't sanity check the segment file, sanity checking only the indexes adds little benefit.
   2. For segment with offset above the flushed point (recovery point), we will recover these segments in `recoveryLog()` (Log.scala) in any case so sanity checking the index files for these segments is redundant.

We did experiments on a cluster with 15 brokers, each of which has ~3k segments (and there are 31.8k partitions with RF=3 which are evenly distributed across brokers; total bytes-in-rate is around 400 MBps). The results show that rolling bounce time reduces from 135 minutes to 55 minutes.

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

Fail produce requests using zstd until the inter.broker.protocol.version is large enough that replicas are ensured to support it. Otherwise, followers receive the `UNSUPPORTED_COMPRESSION_TYPE` when fetching zstd data and ISRs shrink.

Reviewers: Jason Gustafson <jason@confluent.io>

Fix the following situations, where pending members (one that has a member-id, but hasn't joined the group) can cause rebalance operations to fail: 

- In AbstractCoordinator, a pending consumer should be allowed to leave.
- A rebalance operation must successfully complete if a pending member either joins or times out.
- During a rebalance operation, a pending member must be able to leave a group.

Reviewers: Boyang Chen <bchen11@outlook.com>, Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

- Compare last offset of first batch (instead of first offset) with index offset
- Early exit from loop due to zero entries must happen before checking for mismatch
- {TimeIndex,OffsetIndex}.entry should return absolute offset like other methods.
These methods are only used by DumpLogSegments.
- DumpLogSegments now calls `closeHandlers` on OffsetIndex, TimeIndex
and FileRecords.
- Add OffsetIndex, TimeIndex and DumpLogSegments tests
- Remove unnecessary casts by using covariant returns in OffsetIndex and TimeIndex
- Minor clean-ups
- Fix `checkArgs` so that it does what it says only.

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>, Sriharsha Chintalapani <sriharsha@apache.org>

Replaced `forall` with `exists`. Added a unit test to `KafkaApisTest` that failed before the change.

Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>