[KAFKA-7132] [WIP] Consider adding a faster form of rebalance by ConcurrencyPractitioner · Pull Request #5340 · apache/kafka

ConcurrencyPractitioner · 2018-07-06T06:04:09Z

Currently, when a consumer falls out of a consumer group, it will restart processing from the last checkpointed offset. However, this design could result in a lag which some users could not afford to let happen. For example, lets say a consumer crashed at offset 100, with the last checkpointed offset being at 70. When it recovers at a later offset (say, 120), it will be behind by an offset range of 50 (120 - 70). This is because the consumer restarted at 70, forcing it to reprocess old data. To avoid this from happening, one option would be to allow the current consumer to start processing not from the last checkpointed offset (which is 70 in the example), but from 120 where it recovers. Meanwhile, a new KafkaConsumer will be instantiated and start reading from offset 70 in concurrency with the old process, and will be terminated once it reaches 120. In this manner, a considerable amount of lag can be avoided, particularly since the old consumer could proceed as if nothing had happened.

Here is the design doc for the pull request:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-333%3A+Add+faster+mode+of+rebalancing

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

ConcurrencyPractitioner · 2018-07-09T00:35:52Z

cc @becketqin @hachikuji

tedyu · 2018-07-09T08:24:23Z

    private final ConsumerInterceptors<?, ?> interceptors;
    private final boolean excludeInternalTopics;
    private final AtomicInteger pendingAsyncCommits;
+    private List<RebalanceConsumerCoordinator> rebalanceCoordinators = null;


RebalanceConsumerCoordinator -> ConsumerRebalancingCoordinator

tedyu · 2018-07-09T08:25:31Z

+            //fetch start offset
+            //fetch end offset (from which the consumer will start polling)
+            //create new RebalanceConsumer
+            rebalanceInProgress.compareAndSet(true, false);


Should the return value be checked ?

I am not sure about the current design right now ... I will need to check.

tedyu · 2018-07-09T08:26:39Z

    }

+    public void setValue(final boolean useMultithreadRebalancing) {
+        if (this.useMultithreadRebalancing == null) {


What if this.useMultithreadRebalancing and useMultithreadRebalancing carry different values (one being true and the other being false) ?

The purpose of this setup is to make this.useMultithreadRebalancing effectively final. Once set to a non-null value, it can no longer be changed.

tedyu · 2018-07-09T08:37:01Z

                        } else if (heartbeat.pollTimeoutExpired(now)) {
                            // the poll timeout has expired, which means that the foreground thread has stalled
                            // in between calls to poll(), so we explicitly leave the group.
+                            rebalanceInProgress.compareAndSet(false, true);


This flag is better set in LeaveGroupResponseHandler where we know whether the LeaveGroup request succeeded or not

You have a point, I will make the change. :)

tedyu · 2018-07-24T20:22:40Z

-                    true,
-                    new ApiVersions(),
-                    throttleTimeSensor,
+                    time, true, new ApiVersions(), throttleTimeSensor,


It's better to keep the parameters aligned (having same indentation)

NPath Complexity is a boon. Checkstyle believes KafkaConsumer's constructor contains too many lines. I did this to make the method shorter.

tedyu · 2018-07-24T20:31:27Z

+                consumerThread = new Thread(rebalanceConsumer);
+            }
+            if (coordinator.isRebalancing(false)) {
+                System.out.println("Starting thread");


Use log object if this is to be kept

tedyu · 2018-07-24T20:32:59Z

+                                                               null,
+                                                               new HashMap<>(),
+                                                               new HashMap<>());
+                consumerThread = new Thread(rebalanceConsumer);


consumerThread -> rebalancingConsumerThread

tedyu · 2018-07-24T20:35:53Z

+    private RebalanceKafkaConsumer.RequestResult pollForResults(final long timeoutMs, final long now) {
+        long elapsed = time.milliseconds() - now;
+        boolean condition = remainingTimeAtLeastZero(timeoutMs, elapsed) != 0;
+        while (result == null && condition) {


Since condition is just a comparison, you can put the comparison here directly

Oh, I did this to make it shorter. (Wanted to make it more readable)

tedyu · 2018-07-24T20:37:17Z

+        while (result == null && condition) {
+            try {
+                Thread.sleep(retryBackoffMs);
+            } catch (InterruptedException exc) { }


Restore interrupt status by calling Thread.currentThread().interrupt();

Oh, sure I will do that.

tedyu · 2018-07-24T20:54:58Z

+        return result;
+    }
+
+    private ConsumerRecords<K, V>  mergeRecords(final ConsumerRecords<K, V> records1, final ConsumerRecords<K, V> records2) {


This can be moved to ConsumerRecords class

tedyu · 2018-07-24T20:55:35Z

+    }
+
+    private ConsumerRecords<K, V>  mergeRecords(final ConsumerRecords<K, V> records1, final ConsumerRecords<K, V> records2) {
+        final HashMap<TopicPartition, List<ConsumerRecord<K, V>>> map = new HashMap<>();


nit: when records2 is empty, you can return immediately.

tedyu · 2018-07-24T20:56:09Z

+        return new ConsumerRecords<>(map);
+    }
+
+    private ConsumerRecords<K, V> processRecords(final long timeoutMs,


Add javadoc for the parameters

tedyu · 2018-07-24T20:58:24Z

+                    result == null ? (ConsumerRecords<K, V>) pollForResults(timeoutMs, checkRebalanceStart).value
+                            : (ConsumerRecords<K, V>) result.value;
+            if (offsetLagRecords == null) {
+                return this.interceptors.onConsume(new ConsumerRecords<>(records));


nit: extract the call to this.interceptors.onConsume(new ConsumerRecords<>(records)) above line 1258 - its result would always be used

tedyu · 2018-07-24T21:02:56Z

+     * @return true if the secondary consumer thread created is alive.
+     *         false if not alive or has null value
+     */
+    public boolean childConsumerIsAlive() {


rebalanceConsumerIsAlive

tedyu · 2018-07-24T21:15:14Z

+                final long hashCode2 = childConsumerMetadata.hashCode();
+
+                rebalanceConsumer.setOptionalInputArgument(childConsumerMetadata, hashCode1, hashCode2);
+                rebalanceConsumer.sendRequest(null,


What if only one of this call and the commitOffsetsAsync call on line 1642 succeeds ?

tedyu · 2018-07-24T21:20:29Z

+                final Map<TopicPartition, Long> offsets1 = fetcher.endOffsets(partitions, timeout.toMillis());
+                final Map<TopicPartition, Long> offsets2 = (Map<TopicPartition, Long>) pollForResults(timeout.toMillis(), time.milliseconds()).value;
+                final Map<TopicPartition, Long> result = new HashMap<>();
+                for (final TopicPartition partition : partitions) {


See if you can refactor this code which is similar to what beginningOffsets has (apart from the condition between pos1 and pos2)

tedyu · 2018-07-24T21:22:02Z

 import java.util.Objects;
 import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.PriorityBlockingQueue;


Why switch data structure ?

Due to the current policy I have set up in the KIP, this part is adapted to accomadate for OffsetCommitCallback. Currently, we split offsets being committed into two, so now we have two CommitCallbacks. If both of them are called separately, then we have two calls to the callback. We cannot let that happen to ensure that Kafka's behavior remains the same. Therefore, we need both OffsetCommitCompletions in PriorityBlockingQueue before calling OffsetCommitCallback as one call instead of two. The Hashcodes are used to compare two OffsetCommitCompletions.

tedyu · 2018-07-24T21:23:02Z

        addMetadataListener();
    }

+    // method will automatically set to false upon retrieving value


value is assigned instead of false, right ?

Yeah, need to update.

tedyu · 2018-07-24T21:25:24Z

+                    completions.get(completions.size() - 1);
+            boolean containsMatch =
+                    !completions.isEmpty() && previous.hashCode == completion.hashCode;
+            if (completion.hashCode == 0) {


When would hashCode be 0 ?

By default. When we are not splitting the offsets between the two consumers for committing, then we are setting the hashCode to zero.

tedyu · 2018-07-24T21:28:29Z

+                completions.add(completion);
+            }
+        }
+        for (OffsetCommitCompletion completion : completions) {


Consider using java.util.Collections.addAll()

addAll() is not thread safe for PriorityBlockingQueue. Unlike add(). This way we could make it more thread safe.

Richard Yu and others added 23 commits May 12, 2018 11:37

[KAFKA-4696] Streams standby task assignment should be state-store aware

cccc551

Merge branch 'trunk' into KAFKA-4696

5ba559a

Fixing compilation error

588eb8c

Fixing bytebuffer length

7fbc3c7

Adding correct upgrade path

6c66356

Adding StreamsPartitionAssignor upgrade path

c862c45

Implementing better approach of weighting tasks

97a25e8

Reverting previous attempt to weight tasks

b535e73

Fixing bug

d1e482a

Adding some format corrections

0e98ae8

Adding partial test

1805240

Fixing indentation

9df26a8

Changes

67f6584

negating previous change

6f85d0a

Adding set up for separation computation

c55e408

Renaming methods

dae3397

Adding separation and method compaction

d5ddd81

Adding test to verify better balancing

0c845a1

Reverting trunk

a9781bd

[KAFKA-7132] [WIP] Consider adding a faster form of rebalance

db428b8

Merge branch 'trunk' into KIP-333

facd567

Adding comment

b287533

Adding coordinator for possible future use

49fe340

ConcurrencyPractitioner added 2 commits July 9, 2018 10:36

Adding setup for mode

d9ff4de

Adding some changes

78acb2b

tedyu reviewed Jul 9, 2018

View reviewed changes

ConcurrencyPractitioner added 2 commits July 10, 2018 11:29

Chaning inheritance hiearchy

4d4a173

Fixing some tests

b69e1e9

ConcurrencyPractitioner added 10 commits July 19, 2018 16:23

Adding some minor changes

5a80d96

Fixing error

9ed1755

Undoing test change

b747170

Fixing minor bug (again)

0b8e1c6

Changing callbacks for commitAsync

7f13ba6

Adding some conditionals

c804abf

Removing unnecessary method

f1623c6

Improving tests to accomadate for poll

1e6625c

Setting up working version of commitAsync

69c91a6

Adding some changes

f862158

tedyu reviewed Jul 24, 2018

View reviewed changes

ConcurrencyPractitioner added 16 commits July 25, 2018 10:52

Adding finishing touches

f283f1e

Updating kip PR

20ddd33

Adding some refinements

20fa468

Fixing commitAsync policy

56c7222

Removing errors

52e5d4d

Setting up more thread-safe setup

dc99929

Removing print statemnts

f0da700

Removing extra /

0b9940e

Setting up tests to better suit current setup

0be21b6

Setting up better method for commit methods

bad4efc

Removing comment citations

06c75a8

Resolving some comments

79cc952

Adding a comment

6dc9a37

Making less flaky version of sending requests

e6e4d2a

Preventing flaky test

eed7001

Merge branch 'trunk' into KIP-333

e57dc42

ConcurrencyPractitioner closed this Jan 2, 2019

Conversation

ConcurrencyPractitioner commented Jul 6, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

ConcurrencyPractitioner commented Jul 9, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

ConcurrencyPractitioner commented Jul 6, 2018 •

edited

Loading