THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
Status
Current state: Under Discussion
Discussion thread:
JIRA:
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
Currently, there is no possibility in Kafka Streams to increase or decrease the number of stream threads after the Kafka Streams client has been started. There are at least two situations where such functionality would be useful:
- Reacting on an error that killed a stream thread.
- Adapt the number of stream threads to the current workload without the need to stop and start the Kafka Streams client.
Uncaught exceptions thrown in a stream thread kill the stream thread leaving the Kafka Streams client with less stream threads for processing than when the client was started. The only way to replace the killed stream thread is to restart the whole Kafka Streams client. For transient errors, it might make sense to replace a killed stream thread with a new one while users try to find the root cause of the error. That could be accomplished by starting a new stream thread in the uncaught exception handler of the killed stream thread.
When the workload of a Kafka Streams client increases, it might be beneficial to scale up the Kafka Streams client by increasing the number of stream threads that the client is currently running. On the other hand, too many stream threads might also negatively impact performance, so that decreasing the number of running stream threads could also be beneficial to a Kafka Streams application. Having the possibility to increase and decrease the number of stream threads without restarting a Kafka Streams client would allow to adapt a client to its environment on the fly.
Another functionality of a Kafka Streams client that would improve the handling of error conditions, is closing a Kafka Streams client in the uncaught exception handler. For example, when an error occurs that depends on the machine on which the Kafka Streams client is running, e.g., the node is running out of disk space or the disk is corrupted. Currently, it is not possible to close a Kafka Streams client in the uncaught exception handler because of a deadlock. The deadlock occurs because the Kafka Streams client waits for the termination of the stream thread that initiated the closing of the client and the stream thread that initiated the closing of the client waits for the completion of the closing.
In this KIP, we propose to extend the API of the Kafka Streams client to start and shutdown stream threads as well as reliably shutdown the Kafka Streams client, also from the uncaught exception handler.
Public Interfaces
package org.apache.kafka.streams;
public class KafkaStreams implements AutoCloseable {
/**
* Adds and starts a stream thread in addition to the stream threads that are already running in this
* Kafka Streams client.
*
* Since the number of stream threads increases, the sizes of the caches in the new stream thread
* and the existing stream threads are adapted so that the sum of the cache sizes over all stream
* threads does not exceed the total cache size specified in configuration
* {@code cache.max.bytes.buffering}.
*/
public void addStreamThread();
/**
* Removes one stream thread out of the running stream threads from this Kafka Streams client.
*
* The removed stream thread is gracefully shut down. This method does not specify which stream
* thread is shut down.
*
* Since the number of stream threads decreases, the sizes of the caches in the remaining stream
* threads are adapted so that the sum of the cache sizes over all stream threads equals the total
* cache size specified in configuration {@code cache.max.bytes.buffering}.
*/
public void removeStreamThread();
/**
* Request the shutdown of this Kafka Streams client.
*
* This method requests a shutdown, requests a rebalance, and returns.
* That means, that the method does not block until all stream threads are shut down and the Kafka Streams
* client is closed.
*/
public void requestClose();
}
Proposed Changes
We propose to add the above methods to the KafkaStreams class. The behavior of those methods is described in this section alongside other behavioral changes we propose.
Currently, when a Kafka Streams client is started via KafkaStreams#start(), it starts as many stream threads as specified in configuration num.stream.threads.
When KafkaStreams#startStreamThread() is called, a new full-fledged stream thread will be started in addition to the stream threads started by KafkaStreams#start(). The new stream thread will use the same configuration as the existing stream threads. The sum of the cache sizes over the stream thread of a Kafka Streams client specified in configuration cache.max.bytes.buffering will be redistributed over the new stream thread and the existing ones, i.e., the cache of each stream thread will be resized after the next rebalance. Starting a new stream thread will trigger a rebalance. Once the new stream thread has been assigned tasks, it will start to execute them as any other pre-existing stream thread.
When KafkaStreams#shutDownStreamThread() is called, a running stream thread in the Kafka Streams client is shut down. It is not specified which stream thread is shut down. The chosen stream thread will stop executing tasks and close all its resources. The sum of the cache sizes over the stream thread of a Kafka Streams client specified in configuration cache.max.bytes.buffering will be redistributed over the remaining stream threads, i.e., the cache of each remaining stream thread will be resized after the next rebalance. Shutting down a stream thread will trigger a rebalance (also if static membership is configured).
If the last running stream thread is shut down with KafkaStreams#shutDownStreamThread(), the Kafka Streams client will transit to state ERROR. If the Kafka Streams client is in state ERROR and a new stream thread is added via KafkaStreams#startStreamThread(), the client will transit to state REBALANCING. That means, a new transition will be added from state ERROR to state REBALANCING. In such a way, a Kafka Streams client is able to recover from the ERROR state without the need to restart the client.
Stream threads that are in state DEAD will be removed from the stream threads of a Kafka Streams client. This implies that a call to KafkaStreams#localThreadsMetadata() that currently returns the streams thread in state DEAD might not return them anymore depending on the relative timing of the removal of the stream thread and the call to KafkaStreams#localThreadsMetadata().
When KafkaStreams#requestClose() is called, the Kafka Streams client will not close immediately but will record the request to close, request a rebalance, and return. Before joining the group the Kafka Streams client will follow the request to close and shut down itself instead of joining the group. During the shut down the Kafka Streams client will stop all of its threads, i.e., stream threads and other threads, as well as clean up resources. After the shut down the Kafka Streams client will be in state NOT_RUNNING. With such an asynchronous close deadlocks can be avoided when this method is called from an uncaught exception handler of a stream thread.
Compatibility, Deprecation, and Migration Plan
The proposal is backward-compatible because it only adds new methods and does not change any existing methods. The behavioral changes regarding the transitions between states of a Kafka Streams client are backward compatible, because the new transition between state ERROR and state REBALANCING can only be triggered by one of the methods proposed in this KIP. The only proposed change that slightly changes the current behavior is to remove stream threads in state DEAD from the Kafka Streams client, hence they might not be contained in the results of KafkaStreams#localThreadsMetadata(). We regard this change as minor and not relevant to operational continuity.
No methods need to be deprecated and no migration plan is required.
Rejected Alternatives
none