Page History

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}.
     *
     * Stream threads can only be added if this Kafka Streams client is in state RUNNING or REBALANCING.
     *
     * @return name of the added stream thread or empty if a new stream thread could not be added 
	 */
	public Optional<String> 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}. 
     *
     * @return name of the removed stream thread or empty if a stream thread could not be removed because 
     *         no stream threads are alive
	 */
	public Optional<String> removeStreamThread();

	/**
	 * 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}. 
     *
     * If the given timeout is exceeded, the method will throw a {@code TimeoutException}. 
     *
     * @param  timeout
     * @return name of the removed stream thread or empty if a stream thread could not be removed because 
     *         no stream threads are alive
	 */
    public Optional<String> removeStreamThread(final Duration timeout);

}