mindol1004/spring-cloud-stream-binder-kafka
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Ensure that topic is created where broker automatically creates it

Browse Source
This commit is contained in:
Soby Chacko
2016-10-14 17:31:18 -04:00
parent b292f81d46
commit 79643586dd
1 changed files with 18 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -1012,6 +1012,9 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
QueueChannel input = new QueueChannel();
Binding<MessageChannel> producerBinding = binder.bindProducer(testTopicName, output,
createProducerProperties());
waitUntilTopicCreated(getZkUtils(configurationProperties), testTopicName, 5000);
String testPayload = "foo1-" + UUID.randomUUID().toString();
output.send(new GenericMessage<>(testPayload.getBytes()));
@@ -1024,6 +1027,21 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
consumerBinding.unbind();
}
private void waitUntilTopicCreated(ZkUtils zkUtils, String testTopicName, int timeoutInMillis) throws Exception {
int partitionSize = 0;
long currentTimeMillis = System.currentTimeMillis();
do {
try {
partitionSize = invokePartitionSize(testTopicName, zkUtils);
Thread.sleep(100);
}
catch (Exception e) {
//falls through
}
}
while (partitionSize == 0 && (currentTimeMillis + timeoutInMillis) > System.currentTimeMillis());
}
@Test
@SuppressWarnings("unchecked")
public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception {