Skip to content

KAFKA-10500: Add thread option #9615

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 20 commits into from
Dec 4, 2020
Merged

KAFKA-10500: Add thread option #9615

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
dfbd9f4
ThreadCache Resizes
wcarlson5 Nov 6, 2020
dc4fcb8
comments
wcarlson5 Nov 11, 2020
5b2df7e
improved names and checks
wcarlson5 Nov 17, 2020
82f307c
make helper for cache size
wcarlson5 Nov 17, 2020
65f237d
remove redundant check
wcarlson5 Nov 17, 2020
c0a94e3
add thread
wcarlson5 Nov 18, 2020
ad38800
add thread tests
wcarlson5 Nov 19, 2020
d43bb5e
remove extra methods
wcarlson5 Nov 19, 2020
d8793d6
add line
wcarlson5 Nov 19, 2020
9f77626
fix tests
wcarlson5 Nov 19, 2020
c42551a
add start and gets name properly
wcarlson5 Nov 19, 2020
acd65f2
need to add an int test
wcarlson5 Nov 19, 2020
8fdaca1
added int test
wcarlson5 Nov 20, 2020
a698a68
address comments
wcarlson5 Nov 24, 2020
df25f11
reduce line size
wcarlson5 Nov 24, 2020
6d394a3
address comments
wcarlson5 Dec 1, 2020
d818144
address comments pt 2
wcarlson5 Dec 1, 2020
99649b9
wait for running in test
wcarlson5 Dec 2, 2020
17d8ca4
shutdown thread
wcarlson5 Dec 2, 2020
af3e567
Add tests for thread names
wcarlson5 Dec 3, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 104 additions & 42 deletions streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.TreeMap;
Expand Down Expand Up @@ -161,11 +162,20 @@ public class KafkaStreams implements AutoCloseable {
private final ProcessorTopology taskTopology;
private final ProcessorTopology globalTaskTopology;
private final long totalCacheSize;
private final StreamStateListener streamStateListener;
private final StateRestoreListener delegatingStateRestoreListener;
private final Map<Long, StreamThread.State> threadState;
private final ArrayList<StreamThreadStateStoreProvider> storeProviders;
private final UUID processId;
private final KafkaClientSupplier clientSupplier;
private final InternalTopologyBuilder internalTopologyBuilder;

GlobalStreamThread globalStreamThread;
private KafkaStreams.StateListener stateListener;
private StateRestoreListener globalStateRestoreListener;
private boolean oldHandler;
private java.util.function.Consumer<Throwable> streamsUncaughtExceptionHandler;
private final Object changeThreadCount = new Object();

// container states
/**
Expand Down Expand Up @@ -397,6 +407,7 @@ public void setUncaughtExceptionHandler(final StreamsUncaughtExceptionHandler st
final Consumer<Throwable> handler = exception -> handleStreamsUncaughtException(exception, streamsUncaughtExceptionHandler);
synchronized (stateLock) {
if (state == State.CREATED) {
this.streamsUncaughtExceptionHandler = handler;
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
Objects.requireNonNull(streamsUncaughtExceptionHandler);
for (final StreamThread thread : threads) {
thread.setStreamsUncaughtExceptionHandler(handler);
Expand Down Expand Up @@ -755,7 +766,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
this.config = config;
this.time = time;
// The application ID is a required config and hence should always have value
final UUID processId = UUID.randomUUID();
processId = UUID.randomUUID();
final String userClientId = config.getString(StreamsConfig.CLIENT_ID_CONFIG);
final String applicationId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG);
if (userClientId.length() <= 0) {
Expand All @@ -765,7 +776,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
}
final LogContext logContext = new LogContext(String.format("stream-client [%s] ", clientId));
this.log = logContext.logger(getClass());

this.clientSupplier = clientSupplier;
final MetricConfig metricConfig = new MetricConfig()
.samples(config.getInt(StreamsConfig.METRICS_NUM_SAMPLES_CONFIG))
.recordLevel(Sensor.RecordingLevel.forName(config.getString(StreamsConfig.METRICS_RECORDING_LEVEL_CONFIG)))
Expand All @@ -792,7 +803,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
ClientMetrics.addStateMetric(streamsMetrics, (metricsConfig, now) -> state);
log.info("Kafka Streams version: {}", ClientMetrics.version());
log.info("Kafka Streams commit ID: {}", ClientMetrics.commitId());

this.internalTopologyBuilder = internalTopologyBuilder;
// re-write the physical topology according to the config
internalTopologyBuilder.rewriteTopology(config);

Expand Down Expand Up @@ -820,18 +831,18 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
throw new TopologyException("Topology has no stream threads and no global threads, " +
"must subscribe to at least one source topic or global table.");
}

oldHandler = false;
totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG);
final long cacheSizePerThread = getCacheSizePerThread(numStreamThreads);
final boolean hasPersistentStores = taskTopology.hasPersistentLocalStore() ||
(hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore());

streamsUncaughtExceptionHandler = this::defaultStreamsUncaughtExceptionHandler;
try {
stateDirectory = new StateDirectory(config, time, hasPersistentStores);
} catch (final ProcessorStateException fatal) {
throw new StreamsException(fatal);
}
final StateRestoreListener delegatingStateRestoreListener = new DelegatingStateRestoreListener();
delegatingStateRestoreListener = new DelegatingStateRestoreListener();
GlobalStreamThread.State globalThreadState = null;
if (hasGlobalTopology) {
final String globalThreadId = clientId + "-GlobalStreamThread";
Expand All @@ -845,67 +856,118 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
time,
globalThreadId,
delegatingStateRestoreListener,
this::defaultStreamsUncaughtExceptionHandler
streamsUncaughtExceptionHandler
);
globalThreadState = globalStreamThread.state();
}

// use client id instead of thread client id since this admin client may be shared among threads
adminClient = clientSupplier.getAdmin(config.getAdminConfigs(ClientUtils.getSharedAdminClientId(clientId)));

final Map<Long, StreamThread.State> threadState = new HashMap<>(numStreamThreads);
final ArrayList<StreamThreadStateStoreProvider> storeProviders = new ArrayList<>();
for (int i = 0; i < numStreamThreads; i++) {
final StreamThread streamThread = StreamThread.create(
internalTopologyBuilder,
config,
clientSupplier,
adminClient,
processId,
clientId,
streamsMetrics,
time,
streamsMetadataState,
cacheSizePerThread,
stateDirectory,
delegatingStateRestoreListener,
i + 1,
KafkaStreams.this::closeToError,
this::defaultStreamsUncaughtExceptionHandler
);
threads.add(streamThread);
threadState.put(streamThread.getId(), streamThread.state());
storeProviders.add(new StreamThreadStateStoreProvider(streamThread));
}

ClientMetrics.addNumAliveStreamThreadMetric(streamsMetrics, (metricsConfig, now) ->
Math.toIntExact(threads.stream().filter(thread -> thread.state().isAlive()).count()));

final StreamStateListener streamStateListener = new StreamStateListener(threadState, globalThreadState);
threadState = new HashMap<>(numStreamThreads);
storeProviders = new ArrayList<>();
streamStateListener = new StreamStateListener(threadState, globalThreadState);
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
if (hasGlobalTopology) {
globalStreamThread.setStateListener(streamStateListener);
}
for (final StreamThread thread : threads) {
thread.setStateListener(streamStateListener);
for (int i = 1; i <= numStreamThreads; i++) {
createStreamThread(cacheSizePerThread, i);
}

ClientMetrics.addNumAliveStreamThreadMetric(streamsMetrics, (metricsConfig, now) ->
Math.toIntExact(threads.stream().filter(thread -> thread.state().isAlive()).count()));

final GlobalStateStoreProvider globalStateStoreProvider = new GlobalStateStoreProvider(internalTopologyBuilder.globalStateStores());
queryableStoreProvider = new QueryableStoreProvider(storeProviders, globalStateStoreProvider);

stateDirCleaner = setupStateDirCleaner();
oldHandler = false;
maybeWarnAboutCodeInRocksDBConfigSetter(log, config);
rocksDBMetricsRecordingService = maybeCreateRocksDBMetricsRecordingService(clientId, config);
}

private StreamThread createStreamThread(final long cacheSizePerThread, final int threadIdx) {
final StreamThread streamThread = StreamThread.create(
internalTopologyBuilder,
config,
clientSupplier,
adminClient,
processId,
clientId,
streamsMetrics,
time,
streamsMetadataState,
cacheSizePerThread,
stateDirectory,
delegatingStateRestoreListener,
threadIdx,
KafkaStreams.this::closeToError,
streamsUncaughtExceptionHandler
);
streamThread.setStateListener(streamStateListener);
threads.add(streamThread);
threadState.put(streamThread.getId(), streamThread.state());
storeProviders.add(new StreamThreadStateStoreProvider(streamThread));
return streamThread;
}

/**
* Adds and starts a stream thread in addition to the stream threads that are already running in this
* Kafka Streams client.
* <p>
* 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
* {@link StreamsConfig#CACHE_MAX_BYTES_BUFFERING_CONFIG}.
* <p>
* 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() {
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
synchronized (changeThreadCount) {
Comment thread
wcarlson5 marked this conversation as resolved.
if (isRunningOrRebalancing()) {
final int threadIdx = getNextThreadIndex();
final long cacheSizePerThread = getCacheSizePerThread(threads.size() + 1);
resizeThreadCache(cacheSizePerThread);
final StreamThread streamThread = createStreamThread(cacheSizePerThread, threadIdx);
synchronized (stateLock) {

@cadonna cadonna Dec 3, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 wcarlson5 Dec 3, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 wcarlson5 Dec 3, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

if (isRunningOrRebalancing()) {
streamThread.start();
return Optional.of(streamThread.getName());
} else {
streamThread.shutdown();
threads.remove(streamThread);
resizeThreadCache(getCacheSizePerThread(threads.size()));
return Optional.empty();
}
}
}
}
return Optional.empty();
}

private int getNextThreadIndex() {
final HashSet<String> names = new HashSet<>();
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
for (final StreamThread streamThread: threads) {
names.add(streamThread.getName());
}
final String baseName = clientId + "-StreamThread-";
for (int i = 1; i <= threads.size(); i++) {
final String name = baseName + i;
if (!names.contains(name)) {
return i;
}
}
return threads.size() + 1;
}
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated

private long getCacheSizePerThread(final int numStreamThreads) {
return totalCacheSize / (numStreamThreads + ((globalTaskTopology != null) ? 1 : 0));
}

private void resizeThreadCache(final int numStreamThreads) {
final long cacheSizePreThread = getCacheSizePerThread(numStreamThreads);
private void resizeThreadCache(final long cacheSizePerThread) {
for (final StreamThread streamThread: threads) {
streamThread.resizeCache(cacheSizePreThread);
streamThread.resizeCache(cacheSizePerThread);
}
}

Expand Down
3 changes: 3 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ public void resize(final long newCacheSizeBytes) {
final boolean shrink = newCacheSizeBytes < maxCacheSizeBytes;
maxCacheSizeBytes = newCacheSizeBytes;
if (shrink) {
if (caches.values().isEmpty()) {
Comment thread
wcarlson5 marked this conversation as resolved.
Outdated
return;
}
final CircularIterator<NamedCache> circularIterator = new CircularIterator<>(caches.values());
while (sizeBytes() > maxCacheSizeBytes) {
final NamedCache cache = circularIterator.next();
Expand Down
44 changes: 44 additions & 0 deletions streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.Executors;
Expand Down Expand Up @@ -312,6 +313,9 @@ private void prepareStreamThread(final StreamThread thread, final boolean termin
StreamThread.State.PARTITIONS_ASSIGNED);
return null;
}).anyTimes();
thread.resizeCache(EasyMock.anyLong());
EasyMock.expectLastCall().anyTimes();
EasyMock.expect(thread.getName()).andStubReturn("newThread");
thread.shutdown();
EasyMock.expectLastCall().andAnswer(() -> {
supplier.consumer.close();
Expand Down Expand Up @@ -588,6 +592,46 @@ public void testCloseIsIdempotent() {
closeCount, MockMetricsReporter.CLOSE_COUNT.get());
}

@Test
public void shouldAddThreadWhenRunning() throws InterruptedException {
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
streams.start();
final int oldSize = streams.threads.size();
TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING, 15L, "wait until running");
assertThat(streams.addStreamThread(), equalTo(Optional.of("newThread")));
assertThat(streams.threads.size(), equalTo(oldSize + 1));
}

@Test
public void shouldNotAddThreadWhenCreated() {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
final int oldSize = streams.threads.size();
assertThat(streams.addStreamThread(), equalTo(Optional.empty()));
assertThat(streams.threads.size(), equalTo(oldSize));
}

@Test
public void shouldNotAddThreadWhenClosed() {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
final int oldSize = streams.threads.size();
streams.close();
assertThat(streams.addStreamThread(), equalTo(Optional.empty()));
assertThat(streams.threads.size(), equalTo(oldSize));
}

@Test
public void shouldNotAddThreadWhenError() {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
final int oldSize = streams.threads.size();
streams.start();
streamThreadOne.shutdown();
streamThreadTwo.shutdown();
assertThat(streams.addStreamThread(), equalTo(Optional.empty()));
assertThat(streams.threads.size(), equalTo(oldSize));
}


@Test
public void testCannotStartOnceClosed() {
final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time);
Expand Down
Loading