Overview

When application using parallel consumer experiencing frequent rebalancing the restart, there is possibility that very few messages will be lost
This article is to explain the root cause of this kind of message loss.

Work Flow Chart

Explanation：

• In the current work flow, the commit offset per partition is control by offsetHighestSeen, while the increment of this var is changed as soon as the record is polled from kafka topic, even before processing it.
• While the committing offset, it is periodical. Therefore it is potential that the offset is committed even before the message is actually processed. In normal case they are fine. But as the flow chart showed above, when the parallel consumer is beginning to close, and when the message is polled but yet to reach mailbox queue and working threadpool, even when in closing phase, para-consumer will try to drain the mailbox queue and waiting for the task to finish in threadpool. The message cannot be processed.
• the fix is to use offsetHighestSucceeded instead of offsetHighestSeen to make sure the offset will be increased only when it is processed successfully.

PR

Problems Summary:

Sliding Window

• Count Complete Subarrays in an Array

  ans += (nums.size() - front); // with number of array possible ended with at pointer value of “front”, since every end between nums size and front are all valid

Overview

Parallel Consumer is an open source software to enable fast processing message with less instances
This article is to introduce the performance improvement for parallel consumer.

Work Flow Chart

Explanation：

• Previously to get the available workers, it has to go through all shards and get all workers and fill them out based on certain conditions
• Now, we just to get from an atomic variable and it will be updated when the worker is newly-added / failed / finished
• time improvement : O(n) -> O(1)

PR

https://github.com/confluentinc/parallel-consumer/pull/644

Overview

Parallel Consumer is an open source software to enable fast processing message with less instances
This article is to introduce the internal mechanism to share.

Work Flow Chart

Terminology：

AbstractParallelEoSStreamProcessor

• This is the main entry of parallel consumer
• two main components : controlLoop & BrokerPollSystem
• will select the available work container to be applied on the user function

ControlLoop

• check if the new records, if yes, then will create a new work container
• if it is already processed, updated the offset and commit if needed

BrokerPollSystem

• to poll consumer records from broker
• check if throughput threshold has been reached, if reached, it will pause the consumer
• will add to mailbox queue

WorkManager

• Sharded, prioritised, offset managed, order controlled, delayed work queue
• PartitionStateManager && ShardManager

ShardManager

• Shards are local queues of work to be processed.
• maintain map of shards Map<ShardKey, ProcessingShard<K, V>> processingShards

PartitionStateManager

• maintain map of PartitionState Map<TopicPartition, PartitionState<K, V>> partitionStates
• Record the generations of partition assignment, for fencing off invalid work Map<TopicPartition, Long> partitionsAssignmentEpochs

PartitionState

• incompleteOffsets: Offsets beyond the highest committable offset (see getOffsetHighestSequentialSucceeded()) which haven't totally succeeded. Based on decoded metadata and polled records (not offset ranges)
• dirty: if there are offsets to be committed
• partitionsAssignmentEpoch: from PartitionStateManager

WorkContainer

• consumer record
• epoch : if stale worker

Analysis:

define a trie with two elements to show the bit status. And walk through the trie to find the sum of i^1

Time Complexity:

• O(n)

Space Complexity：

• O(n)

Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class Trie(object):
    def __init__(self):
        self.vec = [None for i in xrange(2)]

class Solution(object):
    def findMaximumXOR(self, nums):
        """
        :type nums: List[int]
        :rtype: int
        """
        maxsum = 0

        root = Trie()
        for num in nums:
            curNode = root
            for i in xrange(31, -1, -1):
                curIdx = (num >> i) & 1
                if curNode.vec[curIdx] == None:
                    curNode.vec[curIdx] = Trie()
                curNode = curNode.vec[curIdx]

        for num in nums:
            cursum = 0
            curNode = root
            for i in xrange(31, -1, -1):
                idx = (num >> i) & 1
                if curNode.vec[idx^1] != None:
                    cursum += (1 << i)
                    curNode = curNode.vec[idx^1]
                else:
                    curNode = curNode.vec[idx]
            maxsum = max(maxsum, cursum)

        return maxsum

此方法请测有效，重度BC依赖者福音，再不怕重启MAC了

1

rm -rf "/Users/$(whoami)/Library/Application Support/Beyond Compare/registry.dat"

Analysis:

simple question, should pay attention to the constraints to make sure, the same numbers will
not fall to the next level

Time Complexity:

• O(n)

Space Complexity：

• O(1)

Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

class Solution(object):
    def thirdMax(self, nums):
        """
        :type nums: List[int]
        :rtype: int
        """
        firstmax, secondmax, thirdmax = -sys.maxint, -sys.maxint, -sys.maxint

        for n in nums:
            if n > firstmax:
                thirdmax = secondmax
                secondmax = firstmax
                firstmax = n
            elif n > secondmax and n < firstmax:
                thirdmax = secondmax
                secondmax = n
            elif n > thirdmax and n < secondmax:
                thirdmax = n

        return thirdmax if thirdmax != -sys.maxint else firstmax

Analysis:

Pay attention that list may contain null, so the length should be calculated beforehand and skip the null value

Time Complexity:

• O(n)

Space Complexity：

• O(n)

Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class Vector2D(object):
    def __init__(self, vec2d):
        """
        Initialize your data structure here.
        :type vec2d: List[List[int]]
        """
        self.curx, self.cury = 0, 0
        self.vec2d = vec2d
        self.cnt = 0
        for i in xrange(len(vec2d)):
            for j in xrange(len(vec2d[i])):
                self.cnt += 1

    def next(self):
        """
        :rtype: int
        """
        while self.cury < len(self.vec2d) and len(self.vec2d[self.cury]) == 0:
            self.cury += 1
        retval = self.vec2d[self.cury][self.curx]
        if self.curx+1 < len(self.vec2d[self.cury]):
            self.curx += 1
        else:
            self.curx = 0
            self.cury += 1

        self.cnt -= 1
        return retval

    def hasNext(self):
        """
        :rtype: bool
        """
        return  self.cnt > 0

Analysis:

use count and low and high to mark the boundaries of every left-subtree and right-subtree.
use global_max to record the maximum value

Time Complexity:

• O(n)

Space Complexity：

• O(n)

Code

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

class Solution(object):
    def findLargestHelper(self, root):
        if root is None:
            return 0, 0, 0, 0

        if root.left is None and root.right is None:
            return 1, 1, root.val, root.val

        lcnt, lmax, llow, lhigh = 0, 0, root.val, root.val
        rcnt, rmax, rlow, rhigh = 0, 0, root.val, root.val
        if root.left != None:
            lcnt, lmax, llow, lhigh = self.findLargestHelper(root.left)
        if root.right != None:
            rcnt, rmax, rlow, rhigh = self.findLargestHelper(root.right)

        curcnt, curmax = -1, max(lmax, rmax)
        if root.val >= lhigh and root.val <= rlow:
            if lcnt > -1 and rcnt > -1:
                curcnt = lcnt + rcnt + 1
                curmax = max(curmax, curcnt)

        return curcnt, curmax, llow, rhigh

    def largestBSTSubtree(self, root):
        """
        :type root: TreeNode
        :rtype: int
        """
        curcnt, curmax = -1, 0
        curcnt, curmax, _, _ = self.findLargestHelper(root)
        return curmax

Analysis:

treat it as a tree, find the path to the leaf with minimum hops

Time Complexity:

• O(n)

Space Complexity：

• O(n)

Code

1
2
3
4
5
6
7
8
9
10
11
12
13

class Solution(object):
    def integerReplacement(self, n):
        """
        :type n: int
        :rtype: int
        """
        if n == 1:
            return 0

        if n%2 == 0:
            return self.integerReplacement(n/2)+1
        else:
            return min(self.integerReplacement(n-1), self.integerReplacement(n+1)) + 1