Skip to content

KAFKA-7655 Metadata spamming requests from Kafka Streams under some circumstances, potential DOS #5929

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 5 commits into from
Dec 13, 2018
Merged

KAFKA-7655 Metadata spamming requests from Kafka Streams under some circumstances, potential DOS #5929

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0b3ae81
Re-validate and make sure the topic either exists or it's gone by usi…
pqlmwam Nov 19, 2018
c33757f
Add RETRY_BACKOFF_MS_CONFIG to the admin prefixed properties
pqlmwam Nov 20, 2018
21b3def
Removed tab
pqlmwam Nov 20, 2018
e341d4c
Better error message
pqlmwam Nov 23, 2018
23dfca5
Updated tests and renamed createTopicNames to createdTopicNames
pqlmwam Nov 26, 2018
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
30 changes: 25 additions & 5 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ private InternalAdminClientConfig(final Map<?, ?> props) {
private final AdminClient adminClient;

private final int retries;
private final long retryBackOffMs;

public InternalTopicManager(final AdminClient adminClient,
final StreamsConfig streamsConfig) {
Expand All @@ -67,7 +68,9 @@ public InternalTopicManager(final AdminClient adminClient,

replicationFactor = streamsConfig.getInt(StreamsConfig.REPLICATION_FACTOR_CONFIG).shortValue();
windowChangeLogAdditionalRetention = streamsConfig.getLong(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG);
retries = new InternalAdminClientConfig(streamsConfig.getAdminConfigs("dummy")).getInt(AdminClientConfig.RETRIES_CONFIG);
final InternalAdminClientConfig dummyAdmin = new InternalAdminClientConfig(streamsConfig.getAdminConfigs("dummy"));
retries = dummyAdmin.getInt(AdminClientConfig.RETRIES_CONFIG);
retryBackOffMs = dummyAdmin.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG);
Comment thread
Pasvaz marked this conversation as resolved.
Outdated

log.debug("Configs:" + Utils.NL,
"\t{} = {}" + Utils.NL,
Expand Down Expand Up @@ -115,6 +118,7 @@ public void makeReady(final Map<String, InternalTopicConfig> topics) {

// TODO: KAFKA-6928. should not need retries in the outer caller as it will be retried internally in admin client
int remainingRetries = retries;
boolean retryBackOff = false;
Comment thread
Pasvaz marked this conversation as resolved.
Outdated
boolean retry;
do {
retry = false;
Expand All @@ -124,6 +128,10 @@ public void makeReady(final Map<String, InternalTopicConfig> topics) {
final Set<String> createTopicNames = new HashSet<>();
for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
try {
if (retryBackOff) {
retryBackOff = false;
Thread.sleep(retryBackOffMs);
}
createTopicResult.getValue().get();
createTopicNames.add(createTopicResult.getKey());
} catch (final ExecutionException couldNotCreateTopic) {
Expand All @@ -135,10 +143,22 @@ public void makeReady(final Map<String, InternalTopicConfig> topics) {
log.debug("Could not get number of partitions for topic {} due to timeout. " +
"Will try again (remaining retries {}).", topicName, remainingRetries - 1);
} else if (cause instanceof TopicExistsException) {
createTopicNames.add(createTopicResult.getKey());
log.info("Topic {} exist already: {}",
topicName,
couldNotCreateTopic.toString());
// This topic didn't exist earlier, it might be marked for deletion or it might differ
// from the desired setup. It needs re-validation.
final Map<String, Integer> existingTopicPartition = getNumPartitions(Collections.singleton(topicName));

if (existingTopicPartition.containsKey(topicName)
&& validateTopicPartitions(Collections.singleton(topics.get(topicName)), existingTopicPartition).isEmpty()) {
createTopicNames.add(createTopicResult.getKey());
log.info("Topic {} exists already and has the right number of partitions: {}",
topicName,
couldNotCreateTopic.toString());
Comment thread
Pasvaz marked this conversation as resolved.
} else {
retry = true;
retryBackOff = true;
log.info("Topic {} seems existing already but it doesn't, it is probably marked for deletion: {} " +
Comment thread
Pasvaz marked this conversation as resolved.
Outdated
"will try again in {} ms", topicName, couldNotCreateTopic.toString(), retryBackOffMs);
}
} else {
throw new StreamsException(String.format("Could not create topic %s.", topicName),
couldNotCreateTopic);
Expand Down