C0urante commented on code in PR #13375:
URL: https://github.com/apache/kafka/pull/13375#discussion_r1143840978
##########
connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java:
##########
@@ -152,11 +154,6 @@ public void stop() {
connectCluster.forEach(this::stopWorker);
try {
kafkaCluster.stop();
- } catch (UngracefulShutdownException e) {
- log.warn("Kafka did not shutdown gracefully");
Review Comment:
Is this part no longer necessary?
##########
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java:
##########
@@ -241,7 +245,7 @@ public KafkaClusterTestKit build() throws Exception {
bootstrapMetadata);
} catch (Throwable e) {
log.error("Error creating controller {}", node.id(),
e);
- Utils.swallow(log, Level.WARN,
"sharedServer.stopForController error", () -> sharedServer.stopForController());
+ Utils.swallow(log, Level.WARN,
"sharedServer.stopForController error", sharedServer::stopForController);
Review Comment:
I <3 method references
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java:
##########
@@ -195,7 +197,22 @@ public void testRestartFailedTask() throws Exception {
public void testBrokerCoordinator() throws Exception {
ConnectorHandle connectorHandle =
RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);
workerProps.put(DistributedConfig.SCHEDULED_REBALANCE_MAX_DELAY_MS_CONFIG,
String.valueOf(5000));
- connect = connectBuilder.workerProps(workerProps).build();
+ Properties brokerProps = new Properties();
+
+ // Find a free port and use it in the Kafka broker's listeners config.
We can't use port 0 in the listeners
+ // config to get a random free port because in this test we want to
stop the Kafka broker and then bring it
+ // back up and listening on the same port in order to verify that the
Connect cluster can re-connect to Kafka
+ // and continue functioning normally. If we were to use port 0 here,
the Kafka broker would most likely listen
+ // on a different random free port the second time it is started.
Review Comment:
Any reason we're not preserving the `startOnlyKafkaOnSamePorts` method for
this type of use case?
I see that this change is mentioned in the description, but don't see a
rationale for it.
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/DedicatedMirrorIntegrationTest.java:
##########
@@ -106,9 +106,6 @@ public void testSingleNodeCluster() throws Exception {
EmbeddedKafkaCluster clusterA = startKafkaCluster("A", 1, brokerProps);
EmbeddedKafkaCluster clusterB = startKafkaCluster("B", 1, brokerProps);
- clusterA.start();
- clusterB.start();
Review Comment:
🤦
##########
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java:
##########
@@ -174,13 +174,17 @@ private KafkaConfig createNodeConfig(TestKitNode node) {
props.put(KafkaConfig$.MODULE$.LogDirsProp(),
String.join(",", brokerNode.logDataDirectories()));
}
- props.put(KafkaConfig$.MODULE$.ListenerSecurityProtocolMapProp(),
- "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT");
- props.put(KafkaConfig$.MODULE$.ListenersProp(),
listeners(node.id()));
- props.put(KafkaConfig$.MODULE$.InterBrokerListenerNameProp(),
- nodes.interBrokerListenerName().value());
- props.put(KafkaConfig$.MODULE$.ControllerListenerNamesProp(),
- "CONTROLLER");
+
+ // listeners could be defined via Builder::setConfigProp which
shouldn't be overridden
+ if (!props.containsKey(KafkaConfig$.MODULE$.ListenersProp())) {
Review Comment:
I wonder if this has the potential to be a footgun. We're gating the default
values of several properties on the presence of a single one; could it confuse
people to set just the `listeners` property and then see failures because the
previously-used defaults for the protocol map, controller listener, etc.
weren't set?
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
