You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by rh...@apache.org on 2020/06/11 02:46:04 UTC
[kafka] branch 2.4 updated: KAFKA-9985: Sink connector may exhaust
broker when writing in DLQ (#8663)
This is an automated email from the ASF dual-hosted git repository.
rhauch pushed a commit to branch 2.4
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.4 by this push:
new 30264c2 KAFKA-9985: Sink connector may exhaust broker when writing in DLQ (#8663)
30264c2 is described below
commit 30264c2709a645f5725e662160c1d072cc9245a9
Author: Mario Molina <mm...@gmail.com>
AuthorDate: Wed Jun 10 19:58:00 2020 -0500
KAFKA-9985: Sink connector may exhaust broker when writing in DLQ (#8663)
* Validate topic name against DQL topic in Sink connector config
* Adding parseTopicsList method
* Suppress warning
* KAFKA-9985: Minor changes to improve readability of exception messages
Co-authored-by: Randall Hauch <rh...@gmail.com>
---
.../kafka/connect/runtime/SinkConnectorConfig.java | 43 ++++++++++++++++++++++
.../kafka/connect/runtime/WorkerSinkTask.java | 8 ++--
.../kafka/connect/runtime/AbstractHerderTest.java | 30 +++++++++++++++
3 files changed, 76 insertions(+), 5 deletions(-)
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java
index 0672f4e..0ed1b2b 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SinkConnectorConfig.java
@@ -18,12 +18,17 @@ package org.apache.kafka.connect.runtime;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigDef.Importance;
+import org.apache.kafka.common.config.ConfigDef.Type;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.connect.runtime.isolation.Plugins;
import org.apache.kafka.connect.sink.SinkTask;
import org.apache.kafka.connect.transforms.util.RegexValidator;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
/**
* Configuration needed for all sink connectors
@@ -86,6 +91,7 @@ public class SinkConnectorConfig extends ConnectorConfig {
public static void validate(Map<String, String> props) {
final boolean hasTopicsConfig = hasTopicsConfig(props);
final boolean hasTopicsRegexConfig = hasTopicsRegexConfig(props);
+ final boolean hasDlqTopicConfig = hasDlqTopicConfig(props);
if (hasTopicsConfig && hasTopicsRegexConfig) {
throw new ConfigException(SinkTask.TOPICS_CONFIG + " and " + SinkTask.TOPICS_REGEX_CONFIG +
@@ -96,6 +102,25 @@ public class SinkConnectorConfig extends ConnectorConfig {
throw new ConfigException("Must configure one of " +
SinkTask.TOPICS_CONFIG + " or " + SinkTask.TOPICS_REGEX_CONFIG);
}
+
+ if (hasDlqTopicConfig) {
+ String dlqTopic = props.get(DLQ_TOPIC_NAME_CONFIG).trim();
+ if (hasTopicsConfig) {
+ List<String> topics = parseTopicsList(props);
+ if (topics.contains(dlqTopic)) {
+ throw new ConfigException(String.format("The DLQ topic '%s' may not be included in the list of "
+ + "topics ('%s=%s') consumed by the connector", dlqTopic, SinkTask.TOPICS_REGEX_CONFIG, topics));
+ }
+ }
+ if (hasTopicsRegexConfig) {
+ String topicsRegexStr = props.get(SinkTask.TOPICS_REGEX_CONFIG);
+ Pattern pattern = Pattern.compile(topicsRegexStr);
+ if (pattern.matcher(dlqTopic).matches()) {
+ throw new ConfigException(String.format("The DLQ topic '%s' may not be included in the regex matching the "
+ + "topics ('%s=%s') consumed by the connector", dlqTopic, SinkTask.TOPICS_REGEX_CONFIG, topicsRegexStr));
+ }
+ }
+ }
}
public static boolean hasTopicsConfig(Map<String, String> props) {
@@ -108,6 +133,24 @@ public class SinkConnectorConfig extends ConnectorConfig {
return topicsRegexStr != null && !topicsRegexStr.trim().isEmpty();
}
+ public static boolean hasDlqTopicConfig(Map<String, String> props) {
+ String dqlTopicStr = props.get(DLQ_TOPIC_NAME_CONFIG);
+ return dqlTopicStr != null && !dqlTopicStr.trim().isEmpty();
+ }
+
+ @SuppressWarnings("unchecked")
+ public static List<String> parseTopicsList(Map<String, String> props) {
+ List<String> topics = (List<String>) ConfigDef.parseType(TOPICS_CONFIG, props.get(TOPICS_CONFIG), Type.LIST);
+ if (topics == null) {
+ return Collections.emptyList();
+ }
+ return topics
+ .stream()
+ .filter(topic -> !topic.isEmpty())
+ .distinct()
+ .collect(Collectors.toList());
+ }
+
public String dlqTopicName() {
return getString(DLQ_TOPIC_NAME_CONFIG);
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
index 227f2f6..0fa28e6 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
@@ -54,7 +54,6 @@ import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
@@ -287,10 +286,9 @@ class WorkerSinkTask extends WorkerTask {
SinkConnectorConfig.validate(taskConfig);
if (SinkConnectorConfig.hasTopicsConfig(taskConfig)) {
- String[] topics = taskConfig.get(SinkTask.TOPICS_CONFIG).split(",");
- Arrays.setAll(topics, i -> topics[i].trim());
- consumer.subscribe(Arrays.asList(topics), new HandleRebalance());
- log.debug("{} Initializing and starting task for topics {}", this, topics);
+ List<String> topics = SinkConnectorConfig.parseTopicsList(taskConfig);
+ consumer.subscribe(topics, new HandleRebalance());
+ log.debug("{} Initializing and starting task for topics {}", this, Utils.join(topics, ", "));
} else {
String topicsRegexStr = taskConfig.get(SinkTask.TOPICS_REGEX_CONFIG);
Pattern pattern = Pattern.compile(topicsRegexStr);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
index 48a79b1..fe57d57 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/AbstractHerderTest.java
@@ -318,6 +318,36 @@ public class AbstractHerderTest {
verifyAll();
}
+ @Test(expected = ConfigException.class)
+ public void testConfigValidationTopicsWithDlq() {
+ AbstractHerder herder = createConfigValidationHerder(TestSinkConnector.class, noneConnectorClientConfigOverridePolicy);
+ replayAll();
+
+ Map<String, String> config = new HashMap<>();
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSinkConnector.class.getName());
+ config.put(SinkConnectorConfig.TOPICS_CONFIG, "topic1");
+ config.put(SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG, "topic1");
+
+ herder.validateConnectorConfig(config);
+
+ verifyAll();
+ }
+
+ @Test(expected = ConfigException.class)
+ public void testConfigValidationTopicsRegexWithDlq() {
+ AbstractHerder herder = createConfigValidationHerder(TestSinkConnector.class, noneConnectorClientConfigOverridePolicy);
+ replayAll();
+
+ Map<String, String> config = new HashMap<>();
+ config.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestSinkConnector.class.getName());
+ config.put(SinkConnectorConfig.TOPICS_REGEX_CONFIG, "topic.*");
+ config.put(SinkConnectorConfig.DLQ_TOPIC_NAME_CONFIG, "topic1");
+
+ herder.validateConnectorConfig(config);
+
+ verifyAll();
+ }
+
@Test()
public void testConfigValidationTransformsExtendResults() {
AbstractHerder herder = createConfigValidationHerder(TestSourceConnector.class, noneConnectorClientConfigOverridePolicy);