KAFKA-4175: Can't have StandbyTasks in KafkaStreams where NUM_STREAM_THREADS_CONFIG > 1

[dguy]

standby tasks should be assigned per consumer not per process

[dguy]

@guozhangwang @enothereska @mjsax @hjafarpour - this fixes the lock error worked around in: #1861
Also re-instantes NUM_THREADS_CONFIG to 3 in the SmokeTestClient

[guozhangwang]

Thanks @dguy the patch looks good to me.

But one question I have just thought about is, with num.threads = 1 and num.standby >0, the TaskAssignor will assign both the active and the standby of the same task to the same thread as well, which can also trigger locking issues, right? In that case why changing num.threads from 2 to 1 can temporarily fix the system test?

[dguy]

@guozhangwang - The TaskAssignor doesn't assign the standby tasks to the same process. So if num.threads = 0 and num.standby > 0 you only get the active task assigned. Which seems correct to me as there isn't anywhere for the standby to run.
The change of threads to 1 fixed the system test as all the threads in the same KafkaStreams instance were being assigned the same set of standby tasks, hence they were competing for the same locks.

[enothereska]

LGTM

[guozhangwang]

@dguy In that case, won't a TaskAssignmentException("failed to find an assignable client"); gets thrown since the state == null.

[dguy]

@guozhangwang In TaskAssignor

public void assignStandbyTasks(int numStandbyReplicas) {
        int numReplicas = Math.min(numStandbyReplicas, states.size() - 1);
        for (int i = 0; i < numReplicas; i++) {
            assignTasks(false);
        }
    }

So, that exception will not be thrown for the standby tasks.

[guozhangwang]

You are right. Thanks for the patience!

Merged to trunk.