You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "peterxcli (via GitHub)" <gi...@apache.org> on 2024/12/13 19:31:20 UTC
[PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
peterxcli opened a new pull request, #18175:
URL: https://github.com/apache/kafka/pull/18175
ProducerRebootstrapTest` has many issues in using kraft.
1. it passes `unclean.leader.election.enable` as server config but it is not configured on controller node. We should append the config to topic in creating topic.
2. it assumes `topic` is distributed on [0, 1]. If the order is not correct, the test can't pass. We should create the topic manually to ensure the distribution.
3. in kraft mode, the time to trigger unclean election is 5 mins. it is too long to wait. We should configure controller to reduce the number.
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #18175:
URL: https://github.com/apache/kafka/pull/18175#issuecomment-2578417190
Failed tests are tracked by:
- https://issues.apache.org/jira/browse/KAFKA-15474
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #18175:
URL: https://github.com/apache/kafka/pull/18175#issuecomment-2567096696
I think my original thought was wrong
`unclean.leader.election.enable` has two scopes: topic-level and cluster-wide. If the cluster is configured with `unclean.leader.election.enable=true` at startup, the interval setting (`unclean.leader.election.interval.ms`) becomes irrelevant. This is because whenever the leader of a topic partition’s ISR is fenced, an unclean election will occur immediately. IMO, `unclean.leader.election.interval.ms` only become necessary when the cluster disable unclean election and enable it dynamically.
> `unclean.leader.election.interval.ms` is merely a periodic task that scans for topic partitions without a leader and, based on the value of `unclean.leader.election.enable`(topic's || cluster's), determines whether to perform unclean leader elections for those partitions.
---
Regarding enabling unclean elections at the cluster or topic level: Currently, in `RebootstrapTest`, only the relevant topics have unclean election enabled. I believe it is unnecessary to enable cluster-wide unclean election because those occurrence of unclean election for some topics are expected.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1885728413
##########
core/src/test/java/kafka/server/ProducerRebootstrapIntegrationTest.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 kafka.server;
+
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.ClusterTestExtensions;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Timeout(240)
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ }
+
+)
+public class ProducerRebootstrapIntegrationTest {
+ final String topicName = "topic";
+ final Map<Integer, List<Integer>> expectedReplicaAssignment = Map.of(0, List.of(0, 1), 1, List.of(0, 1));
+ private static final int[] PORTS = choosePorts(2);
+
+ static {
+ // Set system properties that will be used to replace ${port0} placeholders
+ System.setProperty("port0", String.valueOf(PORTS[0]));
+ System.setProperty("port1", String.valueOf(PORTS[1]));
+ }
+
+ @Timeout(240)
+ @ClusterTest(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
+ }
+ )
+ public void testRebootstrap(ClusterInstance cluster) throws Exception {
+ boolean useRebootstrapTriggerMs = true;
+ int part = 0;
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.putAll(clientConfig(useRebootstrapTriggerMs));
+
+
+ try (Admin admin = cluster.admin();
+ Producer<String, String> producer = cluster.producer(Utils.propsToMap(producerProps))) {
+
+ Properties topicProps = new Properties();
+ topicProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "true");
+
+ admin.createTopics(List.of(new NewTopic(topicName, 2, (short) 2).configs(Utils.propsToStringMap(topicProps)))).all().get();
Review Comment:
Should we need `cluster.waitForTopic(topicName, 2);` here?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1900441105
##########
core/src/test/scala/integration/kafka/api/ProducerRebootstrapTest.scala:
##########
@@ -18,15 +18,14 @@ package kafka.api
import org.apache.kafka.clients.producer.ProducerRecord
import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.Disabled
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
class ProducerRebootstrapTest extends RebootstrapTest {
- @Disabled("KAFKA-17986")
@ParameterizedTest(name = "{displayName}.quorum=kraft.useRebootstrapTriggerMs={0}")
@ValueSource(booleans = Array(false, true))
def testRebootstrap(useRebootstrapTriggerMs: Boolean): Unit = {
+ // It's ok to shut the leader down, cause the reelection is small enough to the producer timeout.
Review Comment:
helper function I added to test above stuff.
```scala
def createTopicWithAssignmentAndConfig(
topic: String,
partitionReplicaAssignment: collection.Map[Int, Seq[Int]],
topicConfig: Properties = new Properties,
listenerName: ListenerName = listenerName,
adminClientConfig: Properties = new Properties
): scala.collection.immutable.Map[Int, Int] = {
if (isKRaftTest()) {
Using.resource(createAdminClient(brokers, listenerName, adminClientConfig)) { admin =>
TestUtils.createTopicWithAdmin(
admin = admin,
topic = topic,
brokers = brokers,
controllers = controllerServers,
replicaAssignment = partitionReplicaAssignment,
topicConfig = topicConfig
)
}
} else {
TestUtils.createTopic(
zkClient = zkClient,
topic = topic,
partitionReplicaAssignment = partitionReplicaAssignment,
servers = servers,
topicConfig = topicConfig
)
}
}
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1886475196
##########
core/src/test/java/kafka/server/ProducerRebootstrapIntegrationTest.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 kafka.server;
+
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.ClusterTestExtensions;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Timeout(240)
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ }
+
+)
+public class ProducerRebootstrapIntegrationTest {
+ final String topicName = "topic";
+ final Map<Integer, List<Integer>> expectedReplicaAssignment = Map.of(0, List.of(0, 1), 1, List.of(0, 1));
+ private static final int[] PORTS = choosePorts(2);
+
+ static {
+ // Set system properties that will be used to replace ${port0} placeholders
+ System.setProperty("port0", String.valueOf(PORTS[0]));
+ System.setProperty("port1", String.valueOf(PORTS[1]));
+ }
+
+ @Timeout(240)
+ @ClusterTest(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
+ }
+ )
+ public void testRebootstrap(ClusterInstance cluster) throws Exception {
+ boolean useRebootstrapTriggerMs = true;
+ int part = 0;
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.putAll(clientConfig(useRebootstrapTriggerMs));
+
+
+ try (Admin admin = cluster.admin();
+ Producer<String, String> producer = cluster.producer(Utils.propsToMap(producerProps))) {
+
+ Properties topicProps = new Properties();
+ topicProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "true");
+
+ admin.createTopics(List.of(new NewTopic(topicName, 2, (short) 2).configs(Utils.propsToStringMap(topicProps)))).all().get();
Review Comment:
Yes, but create topic will change metadata, right?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1886659453
##########
core/src/test/java/kafka/server/ProducerRebootstrapIntegrationTest.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 kafka.server;
+
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.ClusterTestExtensions;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Timeout(240)
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ }
+
+)
+public class ProducerRebootstrapIntegrationTest {
+ final String topicName = "topic";
+ final Map<Integer, List<Integer>> expectedReplicaAssignment = Map.of(0, List.of(0, 1), 1, List.of(0, 1));
+ private static final int[] PORTS = choosePorts(2);
+
+ static {
+ // Set system properties that will be used to replace ${port0} placeholders
+ System.setProperty("port0", String.valueOf(PORTS[0]));
+ System.setProperty("port1", String.valueOf(PORTS[1]));
+ }
+
+ @Timeout(240)
+ @ClusterTest(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
+ }
+ )
+ public void testRebootstrap(ClusterInstance cluster) throws Exception {
+ boolean useRebootstrapTriggerMs = true;
+ int part = 0;
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.putAll(clientConfig(useRebootstrapTriggerMs));
+
+
+ try (Admin admin = cluster.admin();
+ Producer<String, String> producer = cluster.producer(Utils.propsToMap(producerProps))) {
+
+ Properties topicProps = new Properties();
+ topicProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "true");
+
+ admin.createTopics(List.of(new NewTopic(topicName, 2, (short) 2).configs(Utils.propsToStringMap(topicProps)))).all().get();
Review Comment:
Oh thanks, you're right.
But seems the old test infra doesn't wait for the metadata propagation explicitly.
So I think it's ok the skip that.
https://github.com/apache/kafka/blob/772aa241b2f1c3e8c8b7ad73ddc910c8272f2d1d/core/src/test/scala/unit/kafka/utils/TestUtils.scala#L380-L406
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1900440899
##########
core/src/test/scala/integration/kafka/api/ProducerRebootstrapTest.scala:
##########
@@ -18,15 +18,14 @@ package kafka.api
import org.apache.kafka.clients.producer.ProducerRecord
import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.Disabled
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
class ProducerRebootstrapTest extends RebootstrapTest {
- @Disabled("KAFKA-17986")
@ParameterizedTest(name = "{displayName}.quorum=kraft.useRebootstrapTriggerMs={0}")
@ValueSource(booleans = Array(false, true))
def testRebootstrap(useRebootstrapTriggerMs: Boolean): Unit = {
+ // It's ok to shut the leader down, cause the reelection is small enough to the producer timeout.
Review Comment:
tested
```scala
val partitionAssigment = Map(0 -> Seq(1, 0), 1 -> Seq(1, 0))
createTopicWithAssignmentAndConfig(topic, partitionReplicaAssignment = partitionAssigment, topicConfig = topicProps)
```
and
```scala
val partitionAssigment = Map(0 -> Seq(0 1), 1 -> Seq(0, 1))
createTopicWithAssignmentAndConfig(topic, partitionReplicaAssignment = partitionAssigment, topicConfig = topicProps)
```
as the topic's assignment, both of them worked well.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1899164224
##########
core/src/test/scala/integration/kafka/api/RebootstrapTest.scala:
##########
@@ -20,18 +20,36 @@ import kafka.server.{KafkaBroker, KafkaConfig}
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
+import org.apache.kafka.server.config.ReplicationConfigs
+import org.junit.jupiter.api.{BeforeEach, TestInfo}
import java.util.Properties
+import java.util.concurrent.TimeUnit
abstract class RebootstrapTest extends AbstractConsumerTest {
override def brokerCount: Int = 2
def server0: KafkaBroker = serverForId(0).get
def server1: KafkaBroker = serverForId(1).get
+ @BeforeEach
+ override def setUp(testInfo: TestInfo): Unit = {
+ super.doSetup(testInfo, createOffsetsTopic = true)
+
+ // Enable unclean leader election on the controller node as well
+ val topicProps = new Properties
+ topicProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "true")
+
+ // create the test topic with all the brokers as replicas
+ createTopic(topic, 2, brokerCount, adminClientConfig = this.adminClientConfig, topicConfig = topicProps)
+ }
+
override def generateConfigs: Seq[KafkaConfig] = {
val overridingProps = new Properties()
overridingProps.put(GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, brokerCount.toString)
+ // Since DELIVERY_TIMEOUT_MS_CONFIG defaults to 120000ms,
+ // we need to set a smaller value for unclean.leader.election.interval.ms
+ overridingProps.put(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, TimeUnit.MILLISECONDS.toMillis(1))
Review Comment:
we should set this config for controller, right?
##########
core/src/test/java/kafka/server/ProducerRebootstrapIntegrationTest.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 kafka.server;
+
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.ClusterTestExtensions;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Timeout(240)
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
Review Comment:
this value is too small ... `3000` is good enough
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #18175:
URL: https://github.com/apache/kafka/pull/18175#issuecomment-2548242457
Hi @chia7712, this PR is ready for review, PTAL. Thanks!
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1885730279
##########
core/src/test/java/kafka/server/ProducerRebootstrapIntegrationTest.java:
##########
@@ -0,0 +1,267 @@
+/*
+ * 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 kafka.server;
+
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.config.TopicConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.ClusterTestExtensions;
+import org.apache.kafka.common.test.api.Type;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.coordinator.group.GroupCoordinatorConfig;
+import org.apache.kafka.server.config.ReplicationConfigs;
+
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Timeout(240)
+@ExtendWith(ClusterTestExtensions.class)
+@ClusterTestDefaults(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ }
+
+)
+public class ProducerRebootstrapIntegrationTest {
+ final String topicName = "topic";
+ final Map<Integer, List<Integer>> expectedReplicaAssignment = Map.of(0, List.of(0, 1), 1, List.of(0, 1));
+ private static final int[] PORTS = choosePorts(2);
+
+ static {
+ // Set system properties that will be used to replace ${port0} placeholders
+ System.setProperty("port0", String.valueOf(PORTS[0]));
+ System.setProperty("port1", String.valueOf(PORTS[1]));
+ }
+
+ @Timeout(240)
+ @ClusterTest(
+ types = {Type.KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, value = "true"),
+ @ClusterConfigProperty(key = ReplicationConfigs.UNCLEAN_LEADER_ELECTION_INTERVAL_MS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "2")
+ }
+ )
+ public void testRebootstrap(ClusterInstance cluster) throws Exception {
+ boolean useRebootstrapTriggerMs = true;
+ int part = 0;
+ Properties producerProps = new Properties();
+ producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ producerProps.putAll(clientConfig(useRebootstrapTriggerMs));
+
+
+ try (Admin admin = cluster.admin();
+ Producer<String, String> producer = cluster.producer(Utils.propsToMap(producerProps))) {
+
+ Properties topicProps = new Properties();
+ topicProps.put(TopicConfig.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, "true");
+
+ admin.createTopics(List.of(new NewTopic(topicName, 2, (short) 2).configs(Utils.propsToStringMap(topicProps)))).all().get();
Review Comment:
I think that's only necessary when the topic metadata has changed?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on code in PR #18175:
URL: https://github.com/apache/kafka/pull/18175#discussion_r1900440899
##########
core/src/test/scala/integration/kafka/api/ProducerRebootstrapTest.scala:
##########
@@ -18,15 +18,14 @@ package kafka.api
import org.apache.kafka.clients.producer.ProducerRecord
import org.junit.jupiter.api.Assertions.assertEquals
-import org.junit.jupiter.api.Disabled
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.ValueSource
class ProducerRebootstrapTest extends RebootstrapTest {
- @Disabled("KAFKA-17986")
@ParameterizedTest(name = "{displayName}.quorum=kraft.useRebootstrapTriggerMs={0}")
@ValueSource(booleans = Array(false, true))
def testRebootstrap(useRebootstrapTriggerMs: Boolean): Unit = {
+ // It's ok to shut the leader down, cause the reelection is small enough to the producer timeout.
Review Comment:
tested
```scala
val partitionAssigment = Map(0 -> Seq(1, 0), 1 -> Seq(1, 0))
```
and
```scala
val partitionAssigment = Map(0 -> Seq(0 1), 1 -> Seq(0, 1))
```
as the topic's assignment, both of them worked well.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #18175:
URL: https://github.com/apache/kafka/pull/18175
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #18175:
URL: https://github.com/apache/kafka/pull/18175#issuecomment-2542800283
@peterxcli I update the topic to match the changes. it is ok to file two PRs :)
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ConsumerRebootstrapTest and ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #18175:
URL: https://github.com/apache/kafka/pull/18175#issuecomment-2548245560
> @peterxcli I update the topic to match the changes. it is ok to file two PRs :)
As we discussed https://github.com/apache/kafka/pull/18175#issuecomment-2548241448, change this PR title back to "Fix ConsumerRebootstrapTest and ProducerRebootstrapTest".
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-17986: Fix ProducerRebootstrapTest [kafka]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #18175:
URL: https://github.com/apache/kafka/pull/18175#issuecomment-2548241448
This PR only fix these two tests, will migrate them to `ClusteTest` after I add the ability to use fixed broker port with it.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org