kkonstantine commented on a change in pull request #8654:
URL: https://github.com/apache/kafka/pull/8654#discussion_r425899194
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java
##########
@@ -66,25 +71,55 @@
/**
* Specify the desired number of partitions for the topic.
*
- * @param numPartitions the desired number of partitions; must be
positive
+ * @param numPartitions the desired number of partitions; must be
positive, or -1 to
+ * signify using the broker's default
* @return this builder to allow methods to be chained; never null
*/
public NewTopicBuilder partitions(int numPartitions) {
+ if (numPartitions == NO_PARTITIONS) {
Review comment:
this check seems to replicate a bit what's happening in `NewTopic`.
Should we just pass the value, given that we have validations too already?
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/util/TopicAdmin.java
##########
@@ -66,25 +71,55 @@
/**
* Specify the desired number of partitions for the topic.
*
- * @param numPartitions the desired number of partitions; must be
positive
+ * @param numPartitions the desired number of partitions; must be
positive, or -1 to
+ * signify using the broker's default
* @return this builder to allow methods to be chained; never null
*/
public NewTopicBuilder partitions(int numPartitions) {
+ if (numPartitions == NO_PARTITIONS) {
+ return defaultPartitions();
+ }
this.numPartitions = numPartitions;
return this;
}
+ /**
+ * Specify the topic's number of partition should be the broker
configuration for
+ * {@code num.partitions}.
+ *
+ * @return this builder to allow methods to be chained; never null
+ */
+ public NewTopicBuilder defaultPartitions() {
+ this.numPartitions = null;
+ return this;
+ }
+
/**
* Specify the desired replication factor for the topic.
*
- * @param replicationFactor the desired replication factor; must be
positive
+ * @param replicationFactor the desired replication factor; must be
positive, or -1 to
+ * signify using the broker's default
* @return this builder to allow methods to be chained; never null
*/
public NewTopicBuilder replicationFactor(short replicationFactor) {
+ if (replicationFactor == NO_REPLICATION_FACTOR) {
Review comment:
same question as above
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
+import org.apache.kafka.test.IntegrationTest;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Integration test for the creation of internal topics.
+ */
+@Category(IntegrationTest.class)
+public class InternalTopicsIntegrationTest {
+
+ private static final Logger log =
LoggerFactory.getLogger(InternalTopicsIntegrationTest.class);
+
+ private EmbeddedConnectCluster.Builder connectBuilder;
+ private EmbeddedConnectCluster connect;
+ Map<String, String> workerProps = new HashMap<>();
+ Properties brokerProps = new Properties();
+
+ @Before
+ public void setup() {
+ // setup Kafka broker properties
+ brokerProps.put("auto.create.topics.enable", String.valueOf(false));
+
+ // build a Connect cluster backed by Kafka and Zk
+ connectBuilder = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(1)
+ .numBrokers(1)
+ .brokerProps(brokerProps);
+ }
+
+ @After
+ public void close() {
+ // stop all Connect, Kafka and Zk threads.
+ connect.stop();
+ }
+
+ @Test
+ public void testCreateInternalTopicsWithDefaultSettings() throws
InterruptedException {
+ int numWorkers = 1;
+ int numBrokers = 3;
+ connect = new
EmbeddedConnectCluster.Builder().name("connect-cluster-1")
Review comment:
Is naming here to distinguish between runs of integration tests?
I have a commit that will help us know the boundaries between ITs. I'll
submit separately.
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/distributed/DistributedConfigTest.java
##########
@@ -105,4 +106,204 @@ public void shouldValidateAllVerificationAlgorithms() {
algorithms.add(algorithms.remove(0));
}
}
+
+ @Test
+ public void shouldAllowNegativeOneAndPositiveForPartitions() {
+ Map<String, String> settings = configs();
+ settings.put(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG, "-1");
+ settings.put(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG, "-1");
+ new DistributedConfig(configs());
+ settings.remove(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG);
+ settings.remove(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG);
+
+ for (int i = 1; i != 100; ++i) {
+ settings.put(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG,
Integer.toString(i));
+ new DistributedConfig(settings);
+
settings.remove(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG);
+
+ settings.put(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG,
Integer.toString(i));
+ new DistributedConfig(settings);
+ }
+ }
+
+ @Test
+ public void shouldNotAllowZeroPartitions() {
+ Map<String, String> settings = configs();
+ settings.put(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG, "0");
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+ settings.remove(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG);
+
+ settings.put(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG, "0");
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+ }
+
+ @Test
+ public void shouldNotAllowNegativePartitionsLessThanNegativeOne() {
+ Map<String, String> settings = configs();
+ for (int i = -2; i > -100; --i) {
+ settings.put(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG,
Integer.toString(i));
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+
settings.remove(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG);
+
+ settings.put(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG,
Integer.toString(i));
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+ }
+ }
+
+ @Test
+ public void shouldAllowNegativeOneAndPositiveForReplicationFactor() {
+ Map<String, String> settings = configs();
+
settings.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG, "-1");
+
settings.put(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG, "-1");
+
settings.put(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG, "-1");
+ new DistributedConfig(configs());
+
settings.remove(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG);
+ settings.remove(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG);
+ settings.remove(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG);
+
+ for (int i = 1; i != 100; ++i) {
+
settings.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG,
Integer.toString(i));
+ new DistributedConfig(settings);
+
settings.remove(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG);
+
+ settings.put(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG,
Integer.toString(i));
+ new DistributedConfig(settings);
+
settings.remove(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG);
+
+ settings.put(DistributedConfig.STATUS_STORAGE_PARTITIONS_CONFIG,
Integer.toString(i));
+ new DistributedConfig(settings);
+ }
+ }
+
+ @Test
+ public void shouldNotAllowZeroReplicationFactor() {
+ Map<String, String> settings = configs();
+
settings.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG, "0");
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+
settings.remove(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG);
+
+
settings.put(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG, "0");
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+
settings.remove(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG);
+
+
settings.put(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG, "0");
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+ }
+
+ @Test
+ public void shouldNotAllowNegativeReplicationFactorLessThanNegativeOne() {
+ Map<String, String> settings = configs();
+ for (int i = -2; i > -100; --i) {
+
settings.put(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG,
Integer.toString(i));
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+
settings.remove(DistributedConfig.CONFIG_STORAGE_REPLICATION_FACTOR_CONFIG);
+
+
settings.put(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG,
Integer.toString(i));
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+
settings.remove(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG);
+
+
settings.put(DistributedConfig.STATUS_STORAGE_REPLICATION_FACTOR_CONFIG,
Integer.toString(i));
+ assertThrows(ConfigException.class, () -> new
DistributedConfig(settings));
+ }
+ }
+
+ @Test
+ public void shouldAllowSettingConfigTopicSettings() {
+ Map<String, String> topicSettings = new HashMap<>();
+ topicSettings.put("foo", "foo value");
+ topicSettings.put("bar", "bar value");
+ topicSettings.put("baz.bim", "100");
+ Map<String, String> settings = configs();
+ topicSettings.entrySet().forEach(e -> {
+ settings.put(DistributedConfig.CONFIG_STORAGE_PREFIX + e.getKey(),
e.getValue());
+ });
+ DistributedConfig config = new DistributedConfig(settings);
+ assertEquals(topicSettings, config.configStorageTopicSettings());
+ }
+
+ @Test
+ public void shouldAllowSettingOffsetTopicSettings() {
+ Map<String, String> topicSettings = new HashMap<>();
+ topicSettings.put("foo", "foo value");
+ topicSettings.put("bar", "bar value");
+ topicSettings.put("baz.bim", "100");
+ Map<String, String> settings = configs();
+ topicSettings.entrySet().forEach(e -> {
+ settings.put(DistributedConfig.OFFSET_STORAGE_PREFIX + e.getKey(),
e.getValue());
+ });
+ DistributedConfig config = new DistributedConfig(settings);
+ assertEquals(topicSettings, config.offsetStorageTopicSettings());
+ }
+
+ @Test
+ public void shouldAllowSettingStatusTopicSettings() {
+ Map<String, String> topicSettings = new HashMap<>();
+ topicSettings.put("foo", "foo value");
+ topicSettings.put("bar", "bar value");
+ topicSettings.put("baz.bim", "100");
+ Map<String, String> settings = configs();
+ topicSettings.entrySet().forEach(e -> {
+ settings.put(DistributedConfig.STATUS_STORAGE_PREFIX + e.getKey(),
e.getValue());
+ });
+ DistributedConfig config = new DistributedConfig(settings);
+ assertEquals(topicSettings, config.statusStorageTopicSettings());
+ }
+
+ @Test
+ public void shouldRemoveCompactionFromConfigTopicSettings() {
+ Map<String, String> expectedTopicSettings = new HashMap<>();
+ expectedTopicSettings.put("foo", "foo value");
+ expectedTopicSettings.put("bar", "bar value");
+ expectedTopicSettings.put("baz.bim", "100");
+ Map<String, String> topicSettings = new
HashMap<>(expectedTopicSettings);
+ topicSettings.put("cleanup.policy", "something-else");
+ topicSettings.put("partitions", "3");
+
+ Map<String, String> settings = configs();
+ topicSettings.entrySet().forEach(e -> {
Review comment:
`topicSettings.forEach((k, v) -> {...});` is a better shorthand in cases
like these here where you don't need to pass the entry between stream stages.
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/integration/InternalTopicsIntegrationTest.java
##########
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
+import org.apache.kafka.test.IntegrationTest;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Integration test for the creation of internal topics.
+ */
+@Category(IntegrationTest.class)
+public class InternalTopicsIntegrationTest {
+
+ private static final Logger log =
LoggerFactory.getLogger(InternalTopicsIntegrationTest.class);
+
+ private EmbeddedConnectCluster.Builder connectBuilder;
+ private EmbeddedConnectCluster connect;
+ Map<String, String> workerProps = new HashMap<>();
+ Properties brokerProps = new Properties();
+
+ @Before
+ public void setup() {
+ // setup Kafka broker properties
+ brokerProps.put("auto.create.topics.enable", String.valueOf(false));
+
+ // build a Connect cluster backed by Kafka and Zk
+ connectBuilder = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(1)
+ .numBrokers(1)
+ .brokerProps(brokerProps);
+ }
+
+ @After
+ public void close() {
+ // stop all Connect, Kafka and Zk threads.
+ connect.stop();
+ }
+
+ @Test
+ public void testCreateInternalTopicsWithDefaultSettings() throws
InterruptedException {
+ int numWorkers = 1;
+ int numBrokers = 3;
+ connect = new
EmbeddedConnectCluster.Builder().name("connect-cluster-1")
+ .workerProps(workerProps)
+ .numWorkers(numWorkers)
+ .numBrokers(numBrokers)
+ .brokerProps(brokerProps)
+ .build();
+
+ // Start the Connect cluster
+ connect.start();
+ connect.assertions().assertExactlyNumBrokersAreUp(numBrokers, "Brokers
did not start in time.");
+ connect.assertions().assertExactlyNumWorkersAreUp(numWorkers, "Worker
did not start in time.");
+ log.info("Completed startup of {} Kafka brokers and {} Connect
workers", numBrokers, numWorkers);
+
+ // Check the topics
+ log.info("Verifying the internal topics for Connect");
+ connect.assertions().assertTopicsExist(configTopic(), offsetTopic(),
statusTopic());
+ assertInternalTopicSettings();
+
+ // Remove the Connect worker
+ log.info("Stopping the Connect worker");
+ connect.removeWorker();
+
+ // Sleep for a bit
+ Thread.sleep(3000);
Review comment:
did you notice that this matters? Not suggesting to remove necessarily,
but I wonder whether you noticed it didn't work otherwise.
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/util/TopicAdminTest.java
##########
@@ -97,12 +103,65 @@ public void shouldNotCreateTopicWhenItAlreadyExists() {
}
@Test
- public void shouldCreateTopicWhenItDoesNotExist() {
- NewTopic newTopic =
TopicAdmin.defineTopic("myTopic").partitions(1).compacted().build();
- Cluster cluster = createCluster(1);
- try (MockAdminClient mockAdminClient = new
MockAdminClient(cluster.nodes(), cluster.nodeById(0))) {
- TopicAdmin admin = new TopicAdmin(null, mockAdminClient);
- assertTrue(admin.createTopic(newTopic));
+ public void shouldCreateTopicWithPartitionsWhenItDoesNotExist() {
+ for (int numBrokers = 1; numBrokers < 10; ++numBrokers) {
+ int expectedReplicas = Math.min(3, numBrokers);
+ int maxDefaultRf = Math.min(numBrokers, 5);
+ for (int numPartitions = 1; numPartitions != 30; ++numPartitions) {
+ NewTopic newTopic =
TopicAdmin.defineTopic("myTopic").partitions(numPartitions).compacted().build();
+
+ // Try clusters with no default replication factor or default
partitions
+ assertTopicCreation(numBrokers, newTopic, null, null,
expectedReplicas, numPartitions);
+
+ // Try clusters with different default partitions
+ for (int defaultPartitions = 1; defaultPartitions != 20;
++defaultPartitions) {
+ assertTopicCreation(numBrokers, newTopic,
defaultPartitions, null, expectedReplicas, numPartitions);
+ }
+
+ // Try clusters with different default replication factors
+ for (int defaultRF = 1; defaultRF != maxDefaultRf;
++defaultRF) {
Review comment:
I didn't comment elsewhere, but that's where `!=` has the potential of
stackoverflow vs `<=`. I understand we trust ourselves to set `maxDefaultRf` to
be `>0` but that's a couple indirections away.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
