KAFKA-10500: Add thread option

[wcarlson5]

Can add stream threads now

replace #9581 after rebase

Committer Checklist (excluded from commit message)

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

[wcarlson5]

@cadonna @mjsax @vvcephei Part 3 of KIP-663 add threads is ready for review

[cadonna]

Thank you for the PR, @wcarlson5 !

Here is my feedback

[cadonna]

Why do we need to synchronize the whole method on stateLock?

[wcarlson5]

Well we don't want it changing state while adding a thread

[cadonna]

Wouldn't it be sufficent to check in the beginning if the Streams client is in RUNNING or REBALANCING, then optimistically create the stream thread, and before we start the stream thread check if the client is not in PENDING_SHUTDOWN and not in ERROR and not in NOT_RUNNING? State changes between RUNNING and REBALANCING should not affect adding a stream thread, right?

[wcarlson5]

Okay we can check then only synchronize around the start of the thread to make sure it doesn't shutdown between the check and the starting

[cadonna]

Wouldn't it be sufficent to check in the beginning if the Streams client is in RUNNING or REBALANCING, then optimistically create the stream thread, and before we start the stream thread check if the client is not in PENDING_SHUTDOWN and not in ERROR and not in NOT_RUNNING? State changes between RUNNING and REBALANCING should not affect adding a stream thread, right?

[cadonna]

@wcarlson5 Thank you for the updates!

I have just some minor comments.

[cadonna]

LGTM!

I have just a nit-comment.

Call for committer review and merge: @mjsax @vvcephei @ableegoldman @guozhangwang @abbccdda

[mjsax]

Couple of comments/questions.

[wcarlson5]

@mjsax I think I got all your comments

[mjsax]

i + 1 -> i

Wondering why this does not result in a test failure? (Or does it; Jenkins is still running.) -- Maybe we want to add a small test that verifies that we name threads correctly.

[wcarlson5]

good catch. I don't think we make sure the thread index starts at 1. But let me fix that

[wcarlson5]

@mjsax god catch on the +1. Hopefully I cleared up the remaining concerns, I think I took care of what you meant by "We should ensure that we do proper cleanup for all cases." for threads

[wcarlson5]

good catch. I don't think we make sure the thread index starts at 1. But let me fix that

[ableegoldman]

We should also shutdown the thread if it doesn't get started, otherwise me may leak (consumer or producer) clients. But I'm actually not sure why we don't just do everything (resize cache, create thread) inside the synchronized block? I'm guessing it would deadlock due to locking on the statelock but can't we just synchronize on something else that wouldn't interfere with the StreamThread creation?

[cadonna]

Good point about the shutdown of the stream thread!

Actually, I did not want to have everything in the synchronized block because I thought blocking the client state more than needed was not a good idea. I thought decreasing the size of the cache might be costly if the evicted records are forwarded downstream.
Now that you mention to synchronize on a separate lock, I noticed that we probably need to put resize, start, and cleanup in the same synchronized block. The reason is that if two threads call addStreamThread() one after the other and the later thread passes

final long cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);

before the earlier thread adds the new stream thread to threads in createStreamThread(), the later thread would compute the wrong cache size.

So, I am in favor of having a separate lock that just synchronizes the threads calling addStreamThread(). Maybe we can simply synchronize the whole method (which means to synchronize with start() and close()).

Still a minor issue seems to be the synchronization betweenisRunningOrRebalancing() and streamThread.start(). If between these two calls the Streams client transits to ERROR (the global stream thread died) an IllegalStateException would be thrown from the StreamStateListener because the Streams client would try to transit from ERROR to REBALANCING. But I guess that would also happen if the Streams client transits to ERROR before the new stream thread transits to PARTITION_ASSIGNED and calls the StreamStateListener that would transit the Streams client to REBALANCING. So it needs to be fixed somewhere else.

Did I miss something?

[wcarlson5]

Unfortunately I don't think we can shutdown a thread until we have started it.

I don't think there should be a dead lock by just using the state lock around most of the method. It as mostly about cost. However I think that its is probably the safest way as it solves our problem about needing to remove a thread we have just created and we won't potentially waste time resizing the cache as such.

If we synchronize on a new lock we end up with all the problems we were trying to solve earlier anyways with the possible state changes and having to clean up un-started threads

[cadonna]

Unfortunately I don't think we can shutdown a thread until we have started it.

Have a look at

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

Lines 976 to 983 in aeeb7b2

	public void shutdown() {
	log.info("Informed to shut down");
	final State oldState = setState(State.PENDING_SHUTDOWN);
	if (oldState == State.CREATED) {
	// The thread may not have been started. Take responsibility for shutting down
	completeShutdown(true);
	}
	}

[wcarlson5]

Oh okay, when I shutdown unstarted threads in a test I got a java.lang.IllegalStateException Unexpected state transition. But it looks like that is the client.

I added a new lock for the add threads, and shutdown the thread. I think this address the problem you found with concurrent resizes. As well as @ableegoldman 's concerns

[cadonna]

nit: If it happens that you need to push another commit, could you fix the indentation here? Sorry that I haven't noticed this before.

[wcarlson5]

ah good catch. the diff makes that hard to see as it was actually moved to a new method.

[cadonna]

Sorry to bother you again with the synchronization on the stateLock, but could you explain why we still need it after we synchronize on newThread?

[wcarlson5]

Well newThread only syncs the addThread method. There is still the race condition between the second check of is running and starting the thread. It seems like a bad idea to leave that open as it could cause thread state changes when there shouldn't be. Starting the thread is relatively low cost so this shouldn't have much impact perf wise.

[wcarlson5]

Expanding on this, the problem in the shutdown thread. When the join only waits for alive threads, and to be alive the thread needs to be started.

So if in between the check and the start thread another thread transitions the state to NOT_RUNNING the thread will not join in the shutdown thread. Then when it continues it will start as it passed the check and we will have a thread running after the client is shutdown.

This would be extremely though race condition to find or reproduce so best to just avoid it.

[cadonna]

Shouldn't that be int i = 1; i <= threads.size(); i++? Otherwise, we would look up *-StreamThread-0" and we would not look up "*-StreamThread-" + threads.size().

Could you add some tests that check the correct naming as @mjsax suggested?

[wcarlson5]

Sure, I missed that suggestion

[mjsax]

Just one final question...