KAFKA-10124:Wrong rebalance.time.ms #8836
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
The consumer rebalance protocol is changed. Method onPartitionsRevoked is called after onPartitionsAssigned, so wrong joinTime is got. Change-Id: I561a48a13a870bd3cb03008825b69b804c6a94b4 Signed-off-by: Jiamei Xie <jiamei.xie@arm.com>
chia7712
reviewed
Jun 9, 2020
| var joinTimeMsInSingleRound = 0L | ||
|
|
||
| consumer.subscribe(topics.asJava, new ConsumerRebalanceListener { | ||
| def onPartitionsAssigned(partitions: util.Collection[TopicPartition]): Unit = { | ||
| joinTime.addAndGet(System.currentTimeMillis - joinStart) |
There was a problem hiding this comment.
It seems to me that the origin code want to count the elapsed time of joining group. Hence, the joinStart is required and the following joinStart = System.currentTimeMillis is required too.
For another, the initial value of joinStart is incorrect. It should be equal to testStartTime
There was a problem hiding this comment.
For onPartitionsRevoked. "In eager rebalancing, it will always be called at the start of a rebalance and after the consumer stops fetching data. In cooperative rebalancing, it will be called at the end of a rebalance on the set of partitions being revoked iff the set is non-empty" is said in https://kafka.apache.org/24/javadoc/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html#onPartitionsAssigned-java.util.Collection-.
onPartitionsRevoked is called after onPartitionsAssigned. joinStart is 0 when calling onPartitionsAssigned. joinTime.addAndGet(System.currentTimeMillis - joinStart) is equal to System.currentTimeMillis, which leads val fetchTimeInMs = (endMs - startMs) - joinGroupTimeInMs.get negative.
There was a problem hiding this comment.
Below is the output of kafka-consumer-perf-test.sh, where fetch.time.ms is negative.
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2020-06-03 02:50:50:008, 2020-06-03 02:52:41:376, 19073.5445, 171.2659, 20000061, 179585.3477, 1591123850918, -1591123739550, -0.0000, -0.0126
There was a problem hiding this comment.
I added logs in methods onPartitionsAssigned and onPartitionsRevoked and checked both of them are called just once in normal cases.
There was a problem hiding this comment.
@chia7712 So I think joinStart = System.currentTimeMillis in onPartitionsRevoked is not required, where joinStart is close to endMs.
There was a problem hiding this comment.
hmmm, is there a good place to update the start time of join if they are NOT called just once?
There was a problem hiding this comment.
@chia7712 I don't find a good place to update it. Perhaps we should remove this metric. The way it gets fetchTimeInMs value is also not a good way. val fetchTimeInMs = (endMs - startMs) - joinGroupTimeInMs.get . There might be some time waiting for connection.
There was a problem hiding this comment.
In the new code we may not trigger onPartitionsRevoked anymore although onPartitionsAssigned is always triggered, so we can no longer rely on them to measure the latency. I'm actually thinking we can just get the rebalance latency from metrics directly --- KafkaConsumer.metrics() exposed them --- you can find more details of the added metrics in KIP-429.
There was a problem hiding this comment.
@guozhangwang Sorry for my late reply. I am not working on this PR. Should I abandon this PR? So other people can work on this issue.
There was a problem hiding this comment.
This has been patched:
7ceb34b
|
@ijuma @huxihx @guozhangwang Call for a review. Do you think there is a good place to update |
Author: Xavier Léauté <xavier@confluent.io> Reviewers: Ismael Juma <ismael@juma.me.uk>, Chia-Ping Tsai <chia7712@gmail.com>, Manikumar Reddy <manikumar.reddy@gmail.com>
…gnment fails but Connect workers remain in the group (apache#8805) In the first version of the incremental cooperative protocol, in the presence of a failed sync request by the leader, the assignor was designed to treat the unapplied assignments as lost and trigger a rebalance delay. This commit applies optimizations in these cases to avoid the unnecessary activation of the rebalancing delay. First, if the worker that loses the sync group request or response is the leader, then it detects this failure by checking the what is the expected generation when it performs task assignments. If it's not the expected one, it resets its view of the previous assignment because it wasn't successfully applied and it doesn't represent a correct state. Furthermore, if the worker that has missed the assignment sync is an ordinary worker, then the leader is able to detect that there are lost assignments and instead of triggering a rebalance delay among the same members of the group, it treats the lost tasks as new tasks and reassigns them immediately. If the lost assignment included revocations that were not applied, the leader reapplies these revocations again. Existing unit tests and integration tests are adapted to test the proposed optimizations. Reviewers: Randall Hauch <rhauch@gmail.com>
There is some confusion over the compression rate metrics, as the meaning of the value isn't clearly stated in the metric description. In this case, it was assumed that a higher compression rate value meant better compression. This PR clarifies the meaning of the value, to prevent misunderstandings. Reviewers: Jason Gustafson <jason@confluent.io>
…orkers when incremental cooperative rebalancing is used (apache#8827) When Incremental Cooperative Rebalancing is enabled and a worker fails to read to the end of the config topic, it needs to voluntarily revoke its locally running tasks on time, before these tasks get assigned to another worker, creating a situation where redundant tasks are running in the Connect cluster. Additionally, instead of using the delay `worker.unsync.backoff.ms` that was defined for the eager rebalancing protocol and has a long default value (which coincidentally is equal to the default value of the rebalance delay of the incremental cooperative protocol), the worker should quickly attempt to re-read the config topic and backoff for a fraction of the rebalance delay. After this fix, the worker will retry for a maximum time of 5 times before it revokes its running assignment and for a cumulative delay less than the configured `scheduled.rebalance.max.delay.ms`. Unit tests are added to cover the backoff logic with incremental cooperative rebalancing. Reviewers: Randall Hauch <rhauch@gmail.com>
…hema (apache#7384) Struct value validation in Kafka Connect can be optimized to avoid creating an Iterator when the expectedClasses list is of size 1. This is a meaningful enhancement for high throughput connectors. Reviewers: Konstantine Karantasis <konstantine@confluent.io>
…#8816) For better visibility on the group rebalance, we are trying to print out the evicted members inside the group coordinator during rebalance complete. Reviewers: David Jacot <djacot@confluent.io>, Jason Gustafson <jason@confluent.io>, Guozhang Wang <wangguoz@gmail.com>
Uses a similar (but slightly different) algorithm as in KAFKA-9987 to produce a maximally sticky -- and perfectly balanced -- assignment of tasks to threads within a single client. This is important for in-memory stores which get wiped out when transferred between threads. Reviewers: John Roesler <vvcephei@apache.org>
|
test this please |
…he#8685) Ensure all channels get closed in `Selector.close`, even if some of them raise errors. Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>
The latest commit apache#8254 on this test deleted all topics after each test, but the topic was actually shared among tests before. And after that we are relying on the less-reliable auto-topic generation to get the topic which makes the test flaky. I'm now using different topics for different tests, also setting the app.id for tests differently. Reviewers: Boyang Chen <boyang@confluent.io>, A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>
…solation (apache#8630) This fix excludes `ConnectorClientConfigRequest` and its inner class from class loading isolation in a similar way that KAFKA-8415 excluded `ConnectorClientConfigOverridePolicy`. Reviewer: Konstantine Karantasis <konstantine@confluent.io>
Reviewers: Matthias J. Sax <matthias@confluent.io>
…#8833) Reviewers: Boyang Chen <boyang@confluent.io>, Guozhang Wang <guozhang@confluent.io>, A. Sophie Blee-Goldman <sophie@confluent.io>
…ache#8663) * Validate topic name against DQL topic in Sink connector config * Adding parseTopicsList method * Suppress warning * KAFKA-9985: Minor changes to improve readability of exception messages Co-authored-by: Randall Hauch <rhauch@gmail.com>
… return with an outdated assignment (apache#8453) With Incremental Cooperative Rebalancing, if a worker returns after it's been out of the group for sometime (essentially as a zombie worker) and hasn't voluntarily revoked its own connectors and tasks in the meantime, there's the possibility that these assignments have been distributed to other workers and redundant connectors and tasks might be running now in the Connect cluster. This PR complements previous fixes such as KAFKA-9184, KAFKA-9849 and KAFKA-9851 providing a last line of defense against zombie tasks: if at any rebalance round the leader worker detects that there are duplicate assignments in the group, it revokes them completely and resolves duplication with a correct assignment in the rebalancing round that will follow task revocation. Author: Wang <ywang50@ebay.com> Reviewer: Konstantine Karantasis <konstantine@confluent.io>
Author: Chris Egerton <chrise@confluent.io> Reviewers: Nigel Liang <nigel@nigelliang.com>, Randall Hauch <rhauch@gmail.com>
Reviewers: Guozhang Wang <guozhang@confluent.io>, John Roesler <john@confluent.io>
…property (apache#8455) * KAFKA-9845: Fix plugin.path when config provider is used * Revert "KAFKA-9845: Fix plugin.path when config provider is used" This reverts commit 96caaa9. * KAFKA-9845: Emit ERROR-level log message when config provider is used for plugin.path property * KAFKA-9845: Demote log message level from ERROR to WARN Co-Authored-By: Nigel Liang <nigel@nigelliang.com> * KAFKA-94845: Fix failing unit tests * KAFKA-9845: Add warning message to docstring for plugin.path config * KAFKA-9845: Apply suggestions from code review Co-authored-by: Randall Hauch <rhauch@gmail.com> Co-authored-by: Nigel Liang <nigel@nigelliang.com> Co-authored-by: Randall Hauch <rhauch@gmail.com>
…nup policy (apache#8828) This change adds a check to the KafkaConfigBackingStore, KafkaOffsetBackingStore, and KafkaStatusBackingStore to use the admin client to verify that the internal topics are compacted and do not use the `delete` cleanup policy. Connect already will create the internal topics with `cleanup.policy=compact` if the topics do not yet exist when the Connect workers are started; the new topics are created always as compacted, overwriting any user-specified `cleanup.policy`. However, if the topics already exist the worker did not previously verify the internal topics were compacted, such as when a user manually creates the internal topics before starting Connect or manually changes the topic settings after the fact. The current change helps guard against users running Connect with topics that have delete cleanup policy enabled, which will remove all connector configurations, source offsets, and connector & task statuses that are older than the retention time. This means that, for example, the configuration for a long-running connector could be deleted by the broker, and this will cause restart issues upon a subsequent rebalance or restarting of Connect worker(s). Connect behavior requires that its internal topics are compacted and not deleted after some retention time. Therefore, this additional check is simply enforcing the existing expectations, and therefore does not need a KIP. Author: Randall Hauch <rhauch@gmail.com> Reviewer: Konstantine Karantasis <konstantine@confluent.io>, Chris Egerton <chrise@confluent.io>
…rter (apache#8829) Make sure that the Errant Record Reporter recently added in KIP-610 adheres to the `errors.tolerance` policy. Author: Aakash Shah <ashah@confluent.io> Reviewers: Arjun Satish <arjunconfluent.io>, Randall Hauch <rhauch@gmail.com>
Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <guozhang@confluent.io>, John Roesler <john@confluent.io>
These changes allow herders to continue to function even when a connector they are running hangs in its start, stop, initialize, validate, and/or config methods. The main idea is to make these connector interactions asynchronous and accept a callback that can be invoked upon the completion (successful or otherwise) of these interactions. The distributed herder handles any follow-up logic by adding a new herder request to its queue in that callback, which helps preserve some synchronization and ordering guarantees provided by the current tick model. If any connector refuses to shut down within a graceful timeout period, the framework will abandon it and potentially start a new connector in its place (in cases such as connector restart or reconfiguration). Existing unit tests for the distributed herder and worker have been modified to reflect these changes, and a new integration test named `BlockingConnectorTest` has been added to ensure that they work in practice. Reviewers: Greg Harris <gregh@confluent.io>, Nigel Liang <nigel@nigelliang.com>, Randall Hauch <rhauch@gmail.com>, Konstantine Karantasis <konstantine@confluent.io>
…che#8818) Add an integration test for the task assignor. * ensure we see proper scale-out behavior with warmups * ensure in-memory stores are properly recycled and not restored through the scale-out process Fix two bugs revealed by the test: Bug 1: we can't remove active tasks in the cooperative algorithm, because this causes their state to get discarded (definitely for in-memory stores, and maybe for persistent ones, depending on the state cleaner). Instead, we convert them to standbys so they can keep warm. Bug 2: tasks with only in-memory stores weren't reporting their offset positions Reviewers: Matthias J. Sax <matthias@confluent.io>, A. Sophie Blee-Goldman <sophie@confluent.io>
…pache#8787) Split out the optimized source changelogs and fetch the committed offsets rather than the end offset for task lag computation Reviewers: John Roesler <vvcephei@apache.org>
…84) (apache#8680) In this PR, I have implemented various classes and integration for the read path of the feature versioning system (KIP-584). The ultimate plan is that the cluster-wide finalized features information is going to be stored in ZK under the node /feature. The read path implemented in this PR is centered around reading this finalized features information from ZK, and, processing it inside the Broker. Here is a summary of what's in this PR (a lot of it is new classes): A facility is provided in the broker to declare its supported features, and advertise its supported features via its own BrokerIdZNode under a features key. A facility is provided in the broker to listen to and propagate cluster-wide finalized feature changes from ZK. When new finalized features are read from ZK, feature incompatibilities are detected by comparing against the broker's own supported features. ApiVersionsResponse is now served containing supported and finalized feature information (using the newly added tagged fields). Reviewers: Boyang Chen <boyang@confluent.io>, Jun Rao <junrao@gmail.com>
…hen using default StreamsConfig serdes (apache#8764) Bug Details: Mistakenly setting the value serde to the key serde for an internal wrapped serde in the FKJ workflow. Testing: Modified the existing test to reproduce the issue, then verified that the test passes. Reviewers: Guozhang Wang <wangguoz@gmail.com>, John Roesler <vvcephei@apache.org>
Reviewers: Mickael Maison <mickael.maison@gmail.com>
Most of the values in the metadata upgrade test matrix are just testing the upgrade/downgrade path between two previous releases. This is unnecessary. We run the tests for all supported branches, so what we should test is the up-/down-gradability of released versions with respect to the current branch. Reviewers: Guozhang Wang <wangguoz@gmail.com>
…se directory (apache#8962) Two more edge cases I found producing extra TaskcorruptedException while playing around with the failing eos-beta upgrade test (sadly these are unrelated problems, as the test still fails with these fixes in place). * Need to write the checkpoint when recycling a standby: although we do preserve the changelog offsets when recycling a task, and should therefore write the offsets when the new task is itself closed, we do NOT write the checkpoint for uninitialized tasks. So if the new task is ultimately closed before it gets out of the CREATED state, the offsets will not be written and we can get a TaskCorruptedException * We do not write the checkpoint file if the current offset map is empty; however for eos the checkpoint file is not only used for restoration but also for clean shutdown. Although skipping a dummy checkpoint file does not actually violate any correctness since we are going to re-bootstrap from the log-start-offset anyways, it throws unnecessary TaskCorruptedException which has an overhead itself. Reviewers: John Roesler <vvcephei@apache.org>, Guozhang Wang <wangguoz@gmail.com>
Reviewers: Matthias J. Sax <matthias@confluent.io>
…WN_TOPIC_OR_PARTITION error (apache#8579) Log it as a warning and without a stacktrace (instead of error with stacktrace). This error can be seen in the following cases: * Topic creation, a follower broker of a new partition starts replica fetcher before the prospective leader broker of the new partition receives the leadership information from the controller (see KAFKA-6221). * Topic deletion, a follower broker of a to-be-deleted partition starts replica fetcher after the leader broker of the to-be-deleted partition processes the deletion information from the controller. As expected, clusters with frequent topic creation and deletion report UnknownTopicOrPartitionException with relatively higher frequency. Despite typically being a transient issue, UnknownTopicOrPartitionException may also indicate real issues if it doesn't fix itself after a short period of time. To ensure detection of such scenarios, we set the log level to warn instead of info. Reviewers: Jun Rao <junrao@gmail.com>, Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk>
…#8989) * make GroupInstanceId ignorable in DescribeGroup * tests and cleanups * add throttle test coverage
…ion setup timeouts (apache#8990) This PR fixes a bug introduced in apache#8683. While processing connection set up timeouts, we are iterating through the connecting nodes to process timeouts and we disconnect within the loop, removing the entry from the set in the loop that it iterating over the set. That raises a ConcurrentModificationException exception. The current unit test did not catch this because it was using only one node. Reviewers: Rajini Sivaram <rajinisivaram@googlemail.com>
Call KafkaStreams#cleanUp to reset local state before starting application up the second run. Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Boyang Chen <boyang@confluent.io>, John Roesler <john@confluent.io>
…ache#8934) The intention of using poll(0) is to not block on rebalance but still return some data; however, `updateAssignmentMetadataIfNeeded` have three different logic: 1) discover coordinator if necessary, 2) join-group if necessary, 3) refresh metadata and fetch position if necessary. We only want to make 2) to be non-blocking but not others, since e.g. when the coordinator is down, then heartbeat would expire and cause the consumer to fetch with timeout 0 as well, causing unnecessarily high CPU. Since splitting this function is a rather big change to make as a last minute blocker fix for 2.6, so I made a smaller change to make updateAssignmentMetadataIfNeeded has an optional boolean flag to indicate if 2) above should wait until either expired or complete, otherwise do not wait on the join-group future and just poll with zero timer. Reviewers: Jason Gustafson <jason@confluent.io>
…one call fails (apache#8985) Reviewers: Colin P. McCabe <cmccabe@apache.org>
Increase ZK connection and session timeout in system tests to match the defaults. Reviewers: Jun Rao <junrao@gmail.com>
…ribeConfigs() Add null check for configurationKey to avoid NPE, and add test for it. Author: Luke Chen <showuon@gmail.com> Reviewers: Tom Bentley <tbentley@redhat.com>, huxi <huxi_2b@hotmail.com>, Manikumar Reddy <manikumar.reddy@gmail.com> Closes apache#8966 from showuon/KAFKA-10220
Author: Tom Bentley <tbentley@redhat.com> Reviewers: David Jacot <djacot@confluent.io>, Manikumar Reddy <manikumar.reddy@gmail.com> Closes apache#8808 from tombentley/KAFKA-10109-AclComment-multiple-AdminClients
…cs (apache#3480) Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>
…idCredentialsTest (apache#8992) Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>
Reducing timeout of transaction to clean up the unstable offsets quicker. IN hard_bounce mode, transactional client is killed ungracefully. Hence, it produces unstable offsets which obstructs TransactionalMessageCopier from receiving position of group. Reviewers: Jun Rao <junrao@gmail.com>
Until now we always passed the default changelog topic name to the state serdes. However, for optimized source tables and global tables the changelog topic is the source topic. Most serdes do not use the topic name passed to them. However, if the serdes actually use the topic name for (de)serialization a org.apache.kafka.common.errors.SerializationException is thrown. This commits passed the correct changelog topic to the state serdes of the metered state stores. Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>, John Roesler <vvcephei@apache.org>
…orker workload - Currently we create single channel builder and reuse it in ConnectStressor workload. This will fail when testing with secure connections, as we close channel builder after first connection. This PR creates ChannelBuilder for each test connection. - Also increase to connect ready wait timeout to 500ms. Author: Manikumar Reddy <manikumar.reddy@gmail.com> Reviewers: Ismael Juma <ismael@juma.me.uk>, Rajini Sivaram <rajinisivaram@googlemail.com> Closes apache#8937 from omkreddy/Connect
Reviewers: Matthias J. Sax <matthias@confluent.io>
…tores (apache#8996) Fixes an asymmetry in which we avoid writing checkpoints for non-persistent stores, but still expected to read them, resulting in a spurious TaskCorruptedException. Reviewers: Matthias J. Sax <mjsax@apache.org>, John Roesler <vvcephei@apache.org>
…he#9005) Also piggy-back a small fix to use TreeMap other than HashMap to preserve iteration ordering. Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, John Roesler <vvcephei@apache.org>
…he#9010) Reviewers: A. Sophie Blee-Goldman <sohpie@confluent.io>, John Roesler <john@confluent.io>
Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Matthias J. Sax <matthias@confluent.io>
The consumer rebalance protocol is changed. Method onPartitionsRevoked is called after onPartitionsAssigned, so wrong joinTime is got. Change-Id: I561a48a13a870bd3cb03008825b69b804c6a94b4 Signed-off-by: Jiamei Xie <jiamei.xie@arm.com>
Signed-off-by: Jiamei Xie <jiamei.xie@arm.com>
Signed-off-by: Jiamei Xie <jiamei.xie@arm.com>
…into wrongFetchTimeMs Signed-off-by: Jiamei Xie <jiamei.xie@arm.com>
|
@jiameixie Are you still on this PR? If yes could you try to address my previous comment and rebase it? |
|
That's fine. Maybe you can close this PR, and update the JIRA ticket as well so others can pick up? |
Ok, I have closed the PR and unsigned the JIRA ticket. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
The consumer rebalance protocol is changed. Method onPartitionsRevoked
is called after onPartitionsAssigned, so wrong joinTime is got.
Change-Id: I561a48a13a870bd3cb03008825b69b804c6a94b4
Signed-off-by: Jiamei Xie jiamei.xie@arm.com
More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.
Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.
Committer Checklist (excluded from commit message)