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

Show file tree
Hide file tree
Changes from 4 commits
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
1 change: 1 addition & 0 deletions streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -980,6 +980,7 @@ public Map<String, Object> getMainConsumerConfigs(final String groupId,
// add admin retries configs for creating topics
final AdminClientConfig adminClientDefaultConfig = new AdminClientConfig(getClientPropsWithPrefix(ADMIN_CLIENT_PREFIX, AdminClientConfig.configNames()));
consumerProps.put(adminClientPrefix(AdminClientConfig.RETRIES_CONFIG), adminClientDefaultConfig.getInt(AdminClientConfig.RETRIES_CONFIG));
consumerProps.put(adminClientPrefix(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG), adminClientDefaultConfig.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG));
Comment thread
Pasvaz marked this conversation as resolved.

// verify that producer batch config is no larger than segment size, then add topic configs required for creating topics
final Map<String, Object> topicProps = originalsWithPrefix(TOPIC_PREFIX, false);
Expand Down
31 changes: 26 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);

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;
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,23 @@ 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("Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n" +
"Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n" +
"Error message was: {}", topicName, retryBackOffMs, couldNotCreateTopic.toString());
}
} else {
throw new StreamsException(String.format("Could not create topic %s.", topicName),
couldNotCreateTopic);
Expand Down