Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala (Scala)

Code before:
  }

  private def assertEndpointsEqual(expected: Endpoint, actual: Endpoint): Unit = {
    assertEquals(expected.host(), actual.host(), "Host mismatch")
    assertEquals(expected.port(), actual.port(), "Port mismatch")
    assertEquals(expected.listener(), actual.listener(), "Listener mismatch")
    assertEquals(expected.securityProtocol(), actual.securityProtocol(), "Security protocol mismatch")
  }

  private def listenerListToEndPoints(listenerList: java.util.List[String],
                              securityProtocolMap: util.Map[ListenerName, SecurityProtocol] = SocketServerConfigs.DEFAULT_NAME_TO_SECURITY_PROTO) =
    AbstractKafkaConfig.listenerListToEndPoints(listenerList, securityProtocolMap)

  @Test
  def testListenerDefaults(): Unit = {
    val props = new Properties()
    props.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker")
    props.setProperty(ServerConfigs.BROKER_ID_CONFIG, "1")
    props.setProperty(QuorumConfig.QUORUM_BOOTSTRAP_SERVERS_CONFIG, "localhost:9093")
    props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")

    // configuration with no listeners
    val conf = KafkaConfig.fromProps(props)
    assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.listeners)
    assertNull(conf.listeners.find(_.securityProtocol == SecurityProtocol.PLAINTEXT).get.host)
    assertEquals(conf.effectiveAdvertisedBrokerListeners, listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala)
  }

  private def isValidKafkaConfig(props: Properties): Boolean = {
    try {
      KafkaConfig.fromProps(props)
      true
    } catch {
      case _: IllegalArgumentException | _: ConfigException => false
    }
  }

  @Test
  def testUncleanLeaderElectionDefault(): Unit = {
    val props = TestUtils.createBrokerConfig(0, port = 8181)
    val serverConfig = KafkaConfig.fromProps(props)

    assertEquals(serverConfig.uncleanLeaderElectionEnable, false)
  }

  @Test
  def testUncleanElectionDisabled(): Unit = {
    val props = TestUtils.createBrokerConfig(0, port = 8181)
    props.setProperty(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, String.valueOf(false))
    val serverConfig = KafkaConfig.fromProps(props)


Review comment (suggestion):
It should be in reverse order.
```suggestion
    assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.effectiveAdvertisedBrokerListeners))
```

Code after:
    assertEquals(expected.host(), actual.host(), "Host mismatch")
    assertEquals(expected.port(), actual.port(), "Port mismatch")
    assertEquals(expected.listener(), actual.listener(), "Listener mismatch")
    assertEquals(expected.securityProtocol(), actual.securityProtocol(), "Security protocol mismatch")
  }

  private def listenerListToEndPoints(listenerList: java.util.List[String],
                              securityProtocolMap: util.Map[ListenerName, SecurityProtocol] = SocketServerConfigs.DEFAULT_NAME_TO_SECURITY_PROTO) =
    AbstractKafkaConfig.listenerListToEndPoints(listenerList, securityProtocolMap)

  @Test
  def testListenerDefaults(): Unit = {
    val props = new Properties()
    props.setProperty(KRaftConfigs.PROCESS_ROLES_CONFIG, "broker")
    props.setProperty(ServerConfigs.BROKER_ID_CONFIG, "1")
    props.setProperty(QuorumConfig.QUORUM_BOOTSTRAP_SERVERS_CONFIG, "localhost:9093")
    props.setProperty(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER")

    // configuration with no listeners
    val conf = KafkaConfig.fromProps(props)
    assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.listeners)
    assertNull(conf.listeners.find(_.securityProtocol == SecurityProtocol.PLAINTEXT).get.host)
    assertEquals(listenerListToEndPoints(util.List.of("PLAINTEXT://:9092")).asScala, conf.effectiveAdvertisedBrokerListeners)
  }

  private def isValidKafkaConfig(props: Properties): Boolean = {
    try {
      KafkaConfig.fromProps(props)
      true
    } catch {
      case _: IllegalArgumentException | _: ConfigException => false
    }
  }

  @Test
  def testUncleanLeaderElectionDefault(): Unit = {
    val props = TestUtils.createBrokerConfig(0, port = 8181)
    val serverConfig = KafkaConfig.fromProps(props)

    assertEquals(serverConfig.uncleanLeaderElectionEnable, false)
  }

  @Test
  def testUncleanElectionDisabled(): Unit = {
    val props = TestUtils.createBrokerConfig(0, port = 8181)
    props.setProperty(ReplicationConfigs.UNCLEAN_LEADER_ELECTION_ENABLE_CONFIG, String.valueOf(false))
    val serverConfig = KafkaConfig.fromProps(props)

    assertEquals(serverConfig.uncleanLeaderElectionEnable, false)
  }