This small PR pulls ThrottledReplicationRateLimit out of KafkaConfig and puts it in a class that defines Dynamic Configs. Client configs are also placed in this class and validation added.
Author: Ben Stopford <benstopford@gmail.com>
Reviewers: Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1864 from benstopford/KAFKA-4177
Simple jira which alters two things:
1. kafka-reassign-partitions --verify prints Throttle was removed regardless of whether a throttle was applied. It should only print this if the value was actually changed.
2. --verify should exception if the —throttle argument. (check generate too)
To test this I extracted all validation logic into a separate method and added a test which covers the majority of combinations. The validation logic was retained as is, other than implementing (2) and adding validation to the --broker-list option which you can currently apply to any of hte main actions (where it is ignored). Requirement 1 was tested manually (as it's just println).
Testing:
- Build passes locally.
- System test reassign_partitions_test.py also passes.
Author: Ben Stopford <benstopford@gmail.com>
Reviewers: Jun Rao <junrao@gmail.com>
Closes#1896 from benstopford/KAFKA-4200
This patch adds proper warning message and necessary doc updates for updating the default partition assignment strategy of Mirror Maker from range to round robin. The actual switch would occur as part of a major release cycle (to be scheduled).
Author: Vahid Hashemian <vahidhashemian@us.ibm.com>
Reviewers: Jason Gustafson <jason@confluent.io>
Closes#1499 from vahidhashemian/KAFKA-3831
There is a corner case bug, where during partition reassignment, if the
controller and a broker receiving a new replica are bounced at the same
time, the partition reassignment is failed.
The cause of this bug is a block of code in the KafkaController which
fails the reassignment if the aliveNewReplicas != newReplicas, ie. if
some of the new replicas are offline at the time a controller fails
over.
The fix is to have the controller listen for ISR change events even for
new replicas which are not alive when the controller boots up. Once the
said replicas come online, they will be in the ISR set, and the new
controller will detect this, and then mark the reassignment as
successful.
Interestingly, the block of code in question was introduced in
KAFKA-990, where a concern about this exact scenario was raised :)
This bug was revealed in the system tests in https://github.com/apache/kafka/pull/1904.
The relevant tests will be enabled in either this or a followup PR when PR-1904 is merged.
Thanks to junrao identifying the issue and providing the patch.
Author: Apurva Mehta <apurva.1618@gmail.com>
Reviewers: Jun Rao <junrao@gmail.com>
Closes#1910 from apurvam/KAFKA-4214
Author: Arun Mahadevan <aiyer@hortonworks.com>
Reviewers: Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk>
Closes#1376 from arunmahadevan/cons-consumer-fix
Technically this does not strictly adhere to RFC-952 however it is valid for domain names, urls and uris so we should loosen the requirements a tad.
Author: Ryan Pridgeon <ryan.n.pridgeon@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>
Closes#1856 from rnpridgeon/KAFKA-3719
Author: Jason Gustafson <jason@confluent.io>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Guozhang Wang <wangguoz@gmail.com>
Closes#1859 from hachikuji/KAFKA-3590
We had a number of failures recently due to these timeouts being too low. It's a particular problem if multiple forks are used while running the tests.
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Jason Gustafson <jason@confluent.io>
Closes#1889 from ijuma/increase-zk-timeout-in-tests
Build is unstable, so it's hard to validate this change. Of the various builds up until 11am BST the test ran twice and passed twice.
Author: Ben Stopford <benstopford@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>
Closes#1873 from benstopford/KAFKA-4184
Author: Jiangjie Qin <becket.qin@gmail.com>
Reviewers: Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>
Closes#1852 from becketqin/KAFKA-4148
We added the `reporters` parameter as part of KIP-74: Cluster Id.
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Jason Gustafson <jason@confluent.io>
Closes#1878 from ijuma/add-kafka-startable-overload-for-compat
Changed the lowerBound argument reference in the summary comment of the translateOffset method to match the actual argument name: startingFilePosition.
Author: Luke Zaparaniuk <luke.zaparaniuk@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>
Closes#1876 from lukezaparaniuk/patch-1
Implementation and tests for secure quotas at <user> and <user, client-id> levels as described in KIP-55. Also adds dynamic default quotas for <client-id>, <user> and <user-client-id>. For each client connection, the most specific quota matching the connection is used, with user quota taking precedence over client-id quota.
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
Reviewers: Jun Rao <junrao@gmail.com>
Closes#1753 from rajinisivaram/KAFKA-3492
This PR implements KIP-78:Cluster Identifiers [(link)](https://cwiki.apache.org/confluence/display/KAFKA/KIP-78%3A+Cluster+Id#KIP-78:ClusterId-Overview) and includes the following changes:
1. Changes to broker code
- generate cluster id and store it in Zookeeper
- update protocol to add cluster id to metadata request and response
- add ClusterResourceListener interface, ClusterResource class and ClusterMetadataListeners utility class
- send ClusterResource events to the metric reporters
2. Changes to client code
- update Cluster and Metadata code to support cluster id
- update clients for sending ClusterResource events to interceptors, (de)serializers and metric reporters
3. Integration tests for interceptors, (de)serializers and metric reporters for clients and for protocol changes and metric reporters for broker.
4. System tests for upgrading from previous versions.
Author: Sumit Arrawatia <sumit.arrawatia@gmail.com>
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Jun Rao <junrao@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1830 from arrawatia/kip-78
This applies to Replication Quotas
based on KIP-73 [(link)](https://cwiki.apache.org/confluence/display/KAFKA/KIP-73+Replication+Quotas) originally motivated by KAFKA-1464.
System Tests Run: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/544/
**This first PR demonstrates the approach**.
**_Overview of Change_**
The guts of this change are relatively small. Throttling occurs on both leader and follower sides. A single class tracks the throttled throughput in and out of each broker (**_ReplicationQuotaManager_**).
On the follower side, the Follower Throttled Rate is calculated as fetch responses arrive. Then, before the next fetch request is sent, we check to see if the quota is violated, removing throttled partitions from the request if it is. This is all encapsulated in a few lines of code in the **_ReplicaFetcherThread_**. There is existing code to handle temporal back off, if the request ends up being empty.
On the leader side it's a little more complex. When a fetch request arrives in the leader, it is built, partition by partition, in **_ReplicaManager.readFromLocalLog_**. As we put each partition into the fetch response, we check if the total size fits in the current quota. If the quota is exceeded, the partition will not be added to the fetch response. Importantly, we don't increase the quota at this point, we just check to see if the bytes will fit.
Now, if there aren't enough bytes to send the response immediately, which is common if we're catching up and throttled, then the request will be put in purgatory. I've added some simple code to **_DelayedFetch_** to handle throttled partitions (throttled partitions are checked against the quota, rather than the messages available in the log).
When the delayed fetch completes, and exits purgatory, _**ReplicaManager.readFromLocalLog**_ will be called again. This is why _**ReplicaManager.readFromLocalLog**_ does not actually increase the quota, it just checks whether enough bytes are available for a partition.
Finally, when there are enough bytes to be sent, or the delayed fetch times out, the response will be sent. Before it is sent the throttled-outbound-rate is increased, based on the size of throttled partitions being sent. This is at the end of _**KafkaApis.handleFetchRequest**_, exactly where client quotas are recorded.
There is an acceptance test which asserts the whole throttling process stabilises on the desired value. This covers a number of use cases including many-to-many replication. See **_ReplicationQuotaTest_**.
Note:
It should be noted that this protocol can over-request. The request is built, based on the quota at time t1 (_ReplicaManager.readFromLocalLog_). The bytes in the response are recorded at time t2 (end of _KafkaApis.handleFetchRequest_), where t2 > t1. For this reason I originally included an OverRequestedRate as a JMX metric, but testing has not seen revealed any obvious issue. Over-requesting is quickly compensated by subsequent requests, stabilising close to the quota value.
_**Main stuff left to do:**_
- The fetch size is currently unbounded. This will be addressed in KIP-74, but we need to ensure this ensures requests don’t go beyond the throttle window.
- There are two failures showing up in the system tests on this branch: StreamsSmokeTest.test_streams (which looks like it fails regularly) and OffsetValidationTest.test_broker_rolling_bounce (which I need to look into)
_**Stuff left to do that could be deferred:**_
- Add the extra metrics specified in the KIP.
- There are no system tests.
- There is no validation for the cluster size / throttle combination that could lead to ISR dropouts
Author: Ben Stopford <benstopford@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Apurva Mehta <apurva@confluent.io>, Jun Rao <junrao@gmail.com>
Closes#1776 from benstopford/rep-quotas-v2
Author: Dong Lin <lindong28@gmail.com>
Reviewers: Joel Koshy <jjkoshy.w@gmail.com>, Jiangjie Qin <becket.qin@gmail.com>
Closes#1851 from lindong28/KAFKA-4158
Now uses LogSegment.largestTimestamp to determine age of segment's messages.
Author: Eric Wasserman <eric.wasserman@gmail.com>
Reviewers: Jun Rao <junrao@gmail.com>
Closes#1794 from ewasserman/feat-1981
change console producer default acks to 1, update acks docs. Also added the -1 config to the acks docs since that question comes up often. ijuma and vahidhashemian, does this look reasonable to you?
Author: Dustin Cote <dustin@confluent.io>
Reviewers: Vahid Hashemian <vahidhashemian@us.ibm.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1795 from cotedm/KAFKA-3129
- use AdminTool to check for active consumer group
Author: Matthias J. Sax <matthias@confluent.io>
Reviewers: Guozhang Wang <wangguoz@gmail.com>
Closes#1767 from mjsax/kafka-4058-trunk
Get channel remote address before calling ```channel.close```
Author: Tao Xiao <xiaotao183@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>
Closes#1826 from xiaotao183/KAFKA-4129
Typically this error condition is caused by topic-level configuration issues, so it is useful to include which topic partition was reset for operator use when debugging the root cause.
Author: Dana Powers <dana.powers@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>
Closes#1801 from dpkp/log_topic_partition_reset_dirty_offset
Author: Jiangjie Qin <becket.qin@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>
Closes#1809 from becketqin/KAFKA-4099
Author: Jason Gustafson <jason@confluent.io>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io>
Closes#1807 from hachikuji/KAFKA-4103
set print-data-log option when offset-decoder is set. hachikuji we had talked about this one before, does this change look ok to you?
Author: Dustin Cote <dustin@confluent.io>
Reviewers: Jason Gustafson <jason@confluent.io>, Ewen Cheslack-Postava <ewen@confluent.io>
Closes#1797 from cotedm/KAFKA-4062
They don't require access to `ZkClient`.
Also include a few obvious clean-ups in `ZKUtils`:
* Remove redundant rethrows and braces
* Use named arguments for booleans
Author: Ismael Juma <ismael@juma.me.uk>
Reviewers: Gwen Shapira <cshapi@gmail.com>
Closes#1775 from ijuma/move-some-zk-utils-methods-to-companion-object
Author: Jason Gustafson <jason@confluent.io>
Reviewers: Ismael Juma <ismael@juma.me.uk>
Closes#1786 from hachikuji/hotfix-ctrlchannelmgr-verbose-logging
Change cleanup.policy to accept a comma separated list of valid policies.
Updated LogCleaner.CleanerThread to also run deletion for any topics configured with compact,delete.
Ensure Log.deleteSegments only runs when delete is enabled.
Additional Integration and unit tests to cover new option
Author: Damian Guy <damian.guy@gmail.com>
Reviewers: Grant Henke <granthenke@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>
Closes#1742 from dguy/kafka-4015
handled by adding a catch all for any unhandled exception. Because the jira specifically mentions the InvalidReplicationFactor exception, a test was added for that specific case.
Author: Grant Henke <granthenke@gmail.com>
Reviewers: Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk>
Closes#1739 from granthenke/create-errors
Use System.nanoseconds instead of System.currentTimeMillis in broker timer tasks to cope with changes to wall-clock time.
Author: Rajini Sivaram <rajinisivaram@googlemail.com>
Reviewers: Gwen Shapira
Closes#1768 from rajinisivaram/KAFKA-4051
As discussed in https://issues.apache.org/jira/browse/KAFKA-3894, this PR makes the log cleaner do a "partial" clean on a segment, whereby it builds a partial offset map up to a particular offset in a segment. Once cleaning resumes again, we will continue from the next dirty offset, which can now be located in the middle of a segment.
Prior to this PR, segments with overly numerous keys could crash the log cleaner thread, as it was required that the log cleaner had to fit at least a single segment in the offset map.
Author: Tom Crayford <tcrayford@googlemail.com>
Reviewers: Jun Rao <junrao@gmail.com>
Closes#1725 from tcrayford/dont_crash_log_cleaner_thread_if_segment_overflows_buffer
junrao Could you take a look when get a chance? Thanks.
Author: Jiangjie Qin <becket.qin@gmail.com>
Reviewers: Jun Rao <junrao@gmail.com>
Closes#1769 from becketqin/KAFKA-3163-follow-up
Add an optional configuration for the SecureRandom PRNG implementation, with the default behavior being the same (use the default implementation in the JDK/JRE).
Author: Todd Palino <Todd Palino>
Reviewers: Grant Henke <granthenke@gmail.com>, Ismael Juma <ismael@juma.me.uk>, Joel Koshy <jjkoshy@gmail.com>, Jiangjie Qin <becket.qin@gmail.com>, Rajini Sivaram <rajinisivaram@googlemail.com>
Closes#1747 from toddpalino/trunk
Author: Jason Gustafson <jason@confluent.io>
Reviewers: Ewen Cheslack-Postava <ewen@confluent.io>, Ismael Juma <ismael@juma.me.uk>, Guozhang Wang <wangguoz@gmail.com>
Closes#1627 from hachikuji/KAFKA-3888
Author: Manikumar Reddy O <manikumar.reddy@gmail.com>
Reviewers: Sriharsha Chintalapani <harsha@hortonworks.com>, Jun Rao <junrao@gmail.com>
Closes#1723 from omkreddy/KAFKA-4035
Author: Grant Henke <granthenke@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Jun Rao <junrao@gmail.com>
Closes#1730 from granthenke/test-delete
Author: Jason Gustafson <jason@confluent.io>
Reviewers: Guozhang Wang <wangguoz@gmail.com>, Ismael Juma <ismael@juma.me.uk>
Closes#1720 from hachikuji/KAFKA-4034
Author: Grant Henke <granthenke@gmail.com>
Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <me@ewencp.org>, Jun Rao <junrao@gmail.com>
Closes#1616 from granthenke/delete-wire-new
Jason Gustafson
Ben Stopford
Vahid Hashemian
Apurva Mehta
Arun Mahadevan
Ryan Pridgeon
Ismael Juma
Jiangjie Qin
Luke Zaparaniuk
Andrey Neporada
Rajini Sivaram
Sumit Arrawatia
Dong Lin
Eric Wasserman
Dustin Cote
Matthias J. Sax
Tao Xiao
Dana Powers
Damian Guy
Grant Henke
Tom Crayford
Shuai Zhang
Todd Palino
Manikumar Reddy O
Alexey Ozeritsky