Skip to content

KAFKA-9127: don't create StreamThreads for global-only topology #8540

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
vvcephei merged 11 commits into from
Apr 29, 2020
Merged

KAFKA-9127: don't create StreamThreads for global-only topology #8540

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
d5eecbb
dont create threads, fix up and add tests
ableegoldman Apr 23, 2020
f08afa2
remove unnecessary sync
ableegoldman Apr 23, 2020
ec5dee4
github suggestion
ableegoldman Apr 23, 2020
365c987
fix tests, should stop using find and replace to makes changes alread…
ableegoldman Apr 23, 2020
494818e
review props
ableegoldman Apr 24, 2020
f542c7a
change exception to TopologyException
ableegoldman Apr 27, 2020
f1702f1
'add' test
ableegoldman Apr 27, 2020
459e38c
add separate integration test
ableegoldman Apr 27, 2020
0d07c56
remove added excpt
ableegoldman Apr 27, 2020
85ec07e
remove extra lines
ableegoldman Apr 27, 2020
049e352
fix test exception message
ableegoldman Apr 28, 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
2 changes: 1 addition & 1 deletion checkstyle/suppressions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@
files="StreamsPartitionAssignor.java"/>

<suppress checks="NPathComplexity"
files="(ProcessorStateManager|InternalTopologyBuilder|StreamsPartitionAssignor|StreamThread|TaskManager|AssignorConfiguration).java"/>
files="(AssignorConfiguration|InternalTopologyBuilder|KafkaStreams|ProcessorStateManager|StreamsPartitionAssignor|StreamThread|TaskManager).java"/>

@mjsax mjsax Apr 24, 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.

Do we really need to add this exception? How much work would it be to reduce the complexity of KafkaStreams?

@vvcephei vvcephei Apr 24, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

NPathComplexity is a tough one to work around. We'd wind up having to move some blocks of logic to separate helper classes.


<suppress checks="(FinalLocalVariable|UnnecessaryParentheses|BooleanExpressionComplexity|CyclomaticComplexity|WhitespaceAfter|LocalVariableName)"
files="Murmur3.java"/>
Expand Down
35 changes: 27 additions & 8 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 @@ -38,6 +38,7 @@
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.internals.ApiUtils;
import org.apache.kafka.streams.internals.metrics.ClientMetrics;
import org.apache.kafka.streams.kstream.KStream;
Expand All @@ -51,6 +52,7 @@
import org.apache.kafka.streams.processor.internals.ClientUtils;
import org.apache.kafka.streams.processor.internals.DefaultKafkaClientSupplier;
import org.apache.kafka.streams.processor.internals.GlobalStreamThread;
import org.apache.kafka.streams.processor.internals.GlobalStreamThread.State;
import org.apache.kafka.streams.processor.internals.InternalTopologyBuilder;
import org.apache.kafka.streams.processor.internals.ProcessorTopology;
import org.apache.kafka.streams.processor.internals.StateDirectory;
Expand Down Expand Up @@ -482,8 +484,9 @@ public synchronized void onChange(final Thread thread,
final GlobalStreamThread.State newState = (GlobalStreamThread.State) abstractNewState;
globalThreadState = newState;

// special case when global thread is dead
if (newState == GlobalStreamThread.State.DEAD) {
if (newState == GlobalStreamThread.State.RUNNING) {
maybeSetRunning();
} else if (newState == GlobalStreamThread.State.DEAD) {
if (setState(State.ERROR)) {
log.error("Global thread has died. The instance will be in error state and should be closed.");
}
Expand Down Expand Up @@ -701,18 +704,34 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
internalTopologyBuilder,
parseHostInfo(config.getString(StreamsConfig.APPLICATION_SERVER_CONFIG)));

final int numStreamThreads;
if (internalTopologyBuilder.hasNoNonGlobalTopology()) {
log.info("Overriding number of StreamThreads to zero for global-only topology");
numStreamThreads = 0;
} else {
numStreamThreads = config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG);
}

// create the stream thread, global update thread, and cleanup thread
threads = new StreamThread[config.getInt(StreamsConfig.NUM_STREAM_THREADS_CONFIG)];
threads = new StreamThread[numStreamThreads];

final ProcessorTopology globalTaskTopology = internalTopologyBuilder.buildGlobalStateTopology();
final boolean hasGlobalTopology = globalTaskTopology != null;

if (numStreamThreads == 0 && !hasGlobalTopology) {
log.error("Topology with no input topics will create no stream threads and no global thread.");
throw new TopologyException("Topology has no stream threads and no global threads, " +
"must subscribe to at least one source topic or global table.");
}

long totalCacheSize = config.getLong(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG);
if (totalCacheSize < 0) {
totalCacheSize = 0;
log.warn("Negative cache size passed in. Reverting to cache size of 0 bytes.");
}
final ProcessorTopology globalTaskTopology = internalTopologyBuilder.buildGlobalStateTopology();
final long cacheSizePerThread = totalCacheSize / (threads.length + (globalTaskTopology == null ? 0 : 1));
final long cacheSizePerThread = totalCacheSize / (threads.length + (hasGlobalTopology ? 1 : 0));
final boolean hasPersistentStores = taskTopology.hasPersistentLocalStore() ||
(globalTaskTopology != null && globalTaskTopology.hasPersistentGlobalStore());
(hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore());

try {
stateDirectory = new StateDirectory(config, time, hasPersistentStores);
Expand All @@ -722,7 +741,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,

final StateRestoreListener delegatingStateRestoreListener = new DelegatingStateRestoreListener();
GlobalStreamThread.State globalThreadState = null;
if (globalTaskTopology != null) {
if (hasGlobalTopology) {
final String globalThreadId = clientId + "-GlobalStreamThread";
globalStreamThread = new GlobalStreamThread(
globalTaskTopology,
Expand Down Expand Up @@ -766,7 +785,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder,
Math.toIntExact(Arrays.stream(threads).filter(thread -> thread.state().isAlive()).count()));

final StreamStateListener streamStateListener = new StreamStateListener(threadState, globalThreadState);
if (globalTaskTopology != null) {
if (hasGlobalTopology) {
globalStreamThread.setStateListener(streamStateListener);
}
for (final StreamThread thread : threads) {
Expand Down
4 changes: 4 additions & 0 deletions ...s/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1259,6 +1259,10 @@ synchronized Pattern sourceTopicPattern() {
return sourceTopicPattern;
}

public boolean hasNoNonGlobalTopology() {
return !usesPatternSubscription() && sourceTopicCollection().isEmpty();
}

private boolean isGlobalSource(final String nodeName) {
final NodeFactory<?, ?> nodeFactory = nodeFactories.get(nodeName);

Expand Down
Loading