Skip to content

MINOR: add lower case lister name integration test #19932

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
merged 7 commits into from
Jun 17, 2025
Merged

MINOR: add lower case lister name integration test #19932

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a7133c0
MINOR: add lower case lister name integration test
FrankYang0529 Jun 9, 2025
de10c94
add more assertion
FrankYang0529 Jun 12, 2025
0dea88b
Merge branch 'trunk' into MINOR-add-endpoint-test
FrankYang0529 Jun 12, 2025
081064f
add controller.listener.names lowercase unit test
FrankYang0529 Jun 16, 2025
f406914
Merge branch 'trunk' into MINOR-add-endpoint-test
FrankYang0529 Jun 16, 2025
b68eb29
Merge branch 'trunk' into MINOR-add-endpoint-test
FrankYang0529 Jun 16, 2025
f8c3dca
add comment
FrankYang0529 Jun 16, 2025
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
28 changes: 27 additions & 1 deletion core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
import org.apache.kafka.network.SocketServerConfigs
import org.apache.kafka.server.authorizer._
import org.apache.kafka.server.common.{ApiMessageAndVersion, KRaftVersion, MetadataVersion}
import org.apache.kafka.server.config.{KRaftConfigs, ServerConfigs}
import org.apache.kafka.server.config.{KRaftConfigs, ReplicationConfigs, ServerConfigs}
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig
import org.apache.kafka.server.quota
import org.apache.kafka.server.quota.{ClientQuotaCallback, ClientQuotaType}
Expand Down Expand Up @@ -130,6 +130,32 @@ class KRaftClusterTest {
}
}

@Test
def testClusterWithLowerCaseListeners(): Unit = {
Using.resource(new KafkaClusterTestKit.Builder(
new TestKitNodes.Builder().
setNumBrokerNodes(1).
setBrokerListenerName(new ListenerName("external")).
setNumControllerNodes(3).
build()).build()
) { cluster =>
cluster.format()
cluster.startup()
cluster.brokers().forEach((_, broker) => {
assertEquals("external://localhost:0", broker.config.get(SocketServerConfigs.LISTENERS_CONFIG))
assertEquals("external", broker.config.get(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG))
assertEquals("external:PLAINTEXT,CONTROLLER:PLAINTEXT", broker.config.get(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG))
})
TestUtils.waitUntilTrue(() => cluster.brokers().get(0).brokerState == BrokerState.RUNNING,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What we want to test is to add a lower-case listener name to the broker configuration and then ensure it works. Therefore, could you please ensure the "external" is NOT converted to upper case when creating the embedded kafka? For example, you could check the value of listeners or inter.broker.listener.name within the KafkaConfig.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the suggestion. I add some new assertions for listeners, inter.broker.listener.name, and listener.security.protocol.map.

"Broker never made it to RUNNING state.")
TestUtils.waitUntilTrue(() => cluster.raftManagers().get(0).client.leaderAndEpoch().leaderId.isPresent,
"RaftManager was not initialized.")
Using.resource(Admin.create(cluster.clientProperties())) { admin =>
assertEquals(cluster.nodes().clusterId(), admin.describeCluster().clusterId().get())
}
}
}

@Test
def testCreateClusterAndWaitForBrokerInRunningState(): Unit = {
val cluster = new KafkaClusterTestKit.Builder(
Expand Down
8 changes: 8 additions & 0 deletions core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1869,4 +1869,12 @@ class KafkaConfigTest {
props.put(ShareGroupConfig.SHARE_GROUP_RECORD_LOCK_DURATION_MS_CONFIG, "30000")
assertDoesNotThrow(() => KafkaConfig.fromProps(props))
}

@Test
def testLowercaseControllerListenerNames(): Unit = {
val props = createDefaultConfig()
props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "controller")
val message = assertThrows(classOf[IllegalArgumentException], () => KafkaConfig.fromProps(props)).getMessage
assertEquals("requirement failed: controller.listener.names must contain at least one value appearing in the 'listeners' configuration when running the KRaft controller role", message)
}
}
2 changes: 1 addition & 1 deletion ...n/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -556,7 +556,7 @@ public String bootstrapServers() {
int brokerId = entry.getKey();
BrokerServer broker = entry.getValue();
ListenerName listenerName = nodes.brokerListenerName();
int port = broker.boundPort(listenerName);
int port = broker.boundPort(ListenerName.normalised(listenerName.value()));
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you please add comments to explain the behavior? Also, the controller listener name needs comment too.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the controller listener name needs comment too.

line#578

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Updated it. Thanks.

if (port <= 0) {
throw new RuntimeException("Broker " + brokerId + " does not yet " +
"have a bound port for " + listenerName + ". Did you start " +
Expand Down