Spring Cloud unable to create topics when using KafkaContainer

java.lang.IllegalStateException: Failed to load ApplicationContext

	at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:124)
	at org.springframework.test.context.support.DefaultTestContext.getApplicationContext(DefaultTestContext.java:83)
	at org.springframework.test.context.web.ServletTestExecutionListener.setUpRequestContextIfNecessary(ServletTestExecutionListener.java:189)
	at org.springframework.test.context.web.ServletTestExecutionListener.prepareTestInstance(ServletTestExecutionListener.java:131)
	at org.springframework.test.context.TestContextManager.prepareTestInstance(TestContextManager.java:230)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.createTest(SpringJUnit4ClassRunner.java:228)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner$1.runReflectiveCall(SpringJUnit4ClassRunner.java:287)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.methodBlock(SpringJUnit4ClassRunner.java:289)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:247)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:94)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
	at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
	at org.testcontainers.containers.FailureDetectingExternalResource$1.evaluate(FailureDetectingExternalResource.java:30)
	at org.testcontainers.containers.FailureDetectingExternalResource$1.evaluate(FailureDetectingExternalResource.java:30)
	at org.junit.rules.RunRules.evaluate(RunRules.java:20)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:191)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
	at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
	at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
	at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
Caused by: org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: 
	at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178)
	at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:50)
	at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:348)
	at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:151)
	at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:114)
	at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:880)
	at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546)
	at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693)
	at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360)
	at org.springframework.boot.SpringApplication.run(SpringApplication.java:303)
	at org.springframework.boot.test.context.SpringBootContextLoader.loadContext(SpringBootContextLoader.java:120)
	at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContextInternal(DefaultCacheAwareContextLoaderDelegate.java:98)
	at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:116)
	... 28 more
Caused by: org.springframework.cloud.stream.binder.BinderException: Exception thrown while starting consumer: 
	at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:276)
	at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:66)
	at org.springframework.cloud.stream.binder.AbstractBinder.bindConsumer(AbstractBinder.java:131)
	at org.springframework.cloud.stream.binding.BindingService.bindConsumer(BindingService.java:97)
	at org.springframework.cloud.stream.binding.BindableProxyFactory.bindInputs(BindableProxyFactory.java:221)
	at org.springframework.cloud.stream.binding.InputBindingLifecycle.start(InputBindingLifecycle.java:55)
	at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:175)
	... 40 more
Caused by: kafka.common.KafkaException: Failed to parse the broker info from zookeeper: {"listener_security_protocol_map":{"BROKER":"PLAINTEXT","PLAINTEXT":"PLAINTEXT"},"endpoints":["BROKER://tc-qicnGmJA:9093","PLAINTEXT://localhost:32775"],"jmx_port":-1,"host":"tc-qicnGmJA","timestamp":"1517772725630","port":9093,"version":4}
	at kafka.cluster.Broker$.createBroker(Broker.scala:98)
	at kafka.utils.ZkUtils.getBrokerInfo(ZkUtils.scala:814)
	at kafka.utils.ZkUtils$$anonfun$getAllBrokersInCluster$2.apply(ZkUtils.scala:250)
	at kafka.utils.ZkUtils$$anonfun$getAllBrokersInCluster$2.apply(ZkUtils.scala:250)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
	at kafka.utils.ZkUtils.getAllBrokersInCluster(ZkUtils.scala:250)
	at kafka.admin.AdminUtils$.getBrokerMetadatas(AdminUtils.scala:391)
	at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:413)
	at kafka.admin.AdminUtils.createTopic(AdminUtils.scala)
	at org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation.invokeCreateTopic(Kafka10AdminUtilsOperation.java:51)
	at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner$1.doWithRetry(KafkaTopicProvisioner.java:212)
	at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:287)
	at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:164)
	at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicAndPartitions(KafkaTopicProvisioner.java:206)
	at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(KafkaTopicProvisioner.java:153)
	at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionConsumerDestination(KafkaTopicProvisioner.java:132)
	at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionConsumerDestination(KafkaTopicProvisioner.java:60)
	at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindConsumer(AbstractMessageChannelBinder.java:225)
	... 46 more
Caused by: java.lang.IllegalArgumentException: No enum constant org.apache.kafka.common.protocol.SecurityProtocol.BROKER
	at java.lang.Enum.valueOf(Enum.java:238)
	at org.apache.kafka.common.protocol.SecurityProtocol.valueOf(SecurityProtocol.java:28)
	at org.apache.kafka.common.protocol.SecurityProtocol.forName(SecurityProtocol.java:89)
	at kafka.cluster.EndPoint$.createEndPoint(EndPoint.scala:49)
	at kafka.cluster.Broker$$anonfun$1.apply(Broker.scala:87)
	at kafka.cluster.Broker$$anonfun$1.apply(Broker.scala:86)
	at scala.collection.immutable.List.map(List.scala:273)
	at kafka.cluster.Broker$.createBroker(Broker.scala:86)
	... 68 more

TC 1.6.0, Spring Boot 1.5.10.RELEASE run with

      System.setProperty("spring.kafka.bootstrap-servers", kafka.bootstrapServers)
      System.setProperty("spring.cloud.stream.kafka.binder.defaultZkPort", kafka.getMappedPort(2181).toString())

Start with KafkaContainer's bootstrapServers and mapped ZooKeeper port, then trace the failure through KafkaTopicProvisioner and Kafka10AdminUtilsOperation in the stack trace. Reproduce the Spring Cloud Stream context with the shown system properties and inspect the broker metadata and security-protocol parsing; done means topics can be created without the reported IllegalArgumentException.