Fixes KAFKA-2327

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -725,8 +725,7 @@ case class KafkaConfig (props: java.util.Map[_, _]) extends AbstractConfig(Kafka
       validateUniquePortAndProtocol(getString(KafkaConfig.AdvertisedListenersProp))
       CoreUtils.listenerListToEndPoints(getString(KafkaConfig.AdvertisedListenersProp))
     } else if (getString(KafkaConfig.AdvertisedHostNameProp) != null || getInt(KafkaConfig.AdvertisedPortProp) != null) {
-      CoreUtils.listenerListToEndPoints("PLAINTEXT://" +
-            getString(KafkaConfig.AdvertisedHostNameProp) + ":" + getInt(KafkaConfig.AdvertisedPortProp))
+      CoreUtils.listenerListToEndPoints("PLAINTEXT://" + advertisedHostName + ":" + advertisedPort)
     } else {
       getListeners()
     }

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -170,8 +170,41 @@ class KafkaConfigTest extends JUnit3Suite {
     assertEquals(endpoint.host, advertisedHostName)
     assertEquals(endpoint.port, advertisedPort.toInt)
   }
+    
+  @Test
+  def testAdvertisePortDefault() {
+    val advertisedHostName = "routable-host"
+    val port = "9999"
 
+    val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect)
+    props.put(KafkaConfig.AdvertisedHostNameProp, advertisedHostName)
+    props.put(KafkaConfig.PortProp, port.toString)
 
+    val serverConfig = KafkaConfig.fromProps(props)
+    val endpoints = serverConfig.advertisedListeners
+    val endpoint = endpoints.get(SecurityProtocol.PLAINTEXT).get
+
+    assertEquals(endpoint.host, advertisedHostName)
+    assertEquals(endpoint.port, port.toInt)
+  }
+  
+  @Test
+  def testAdvertiseHostNameDefault() {
+    val hostName = "routable-host"
+    val advertisedPort = "9999"
+
+    val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect)
+    props.put(KafkaConfig.HostNameProp, hostName)
+    props.put(KafkaConfig.AdvertisedPortProp, advertisedPort.toString)
+
+    val serverConfig = KafkaConfig.fromProps(props)
+    val endpoints = serverConfig.advertisedListeners
+    val endpoint = endpoints.get(SecurityProtocol.PLAINTEXT).get
+
+    assertEquals(endpoint.host, hostName)
+    assertEquals(endpoint.port, advertisedPort.toInt)
+  }    
+    
   @Test
   def testDuplicateListeners() {
     val props = new Properties()