mynameborat commented on a change in pull request #951: SAMZA-2127: Upgrade to
Kafka 2.0
URL: https://github.com/apache/samza/pull/951#discussion_r268352274
##########
File path:
samza-kafka/src/test/scala/org/apache/samza/system/kafka/TestKafkaSystemAdmin.scala
##########
@@ -89,49 +91,49 @@ object TestKafkaSystemAdmin extends KafkaServerTestHarness
{
@AfterClass
override def tearDown() {
+ adminClient.close()
systemAdmin.stop()
producer.close()
super.tearDown()
}
def createTopic(topicName: String, partitionCount: Int) {
- createTopic(topicName, partitionCount, REPLICATION_FACTOR)
+ val topicCreationFutures =
adminClient.createTopics(Collections.singletonList(new NewTopic(topicName,
partitionCount, REPLICATION_FACTOR.shortValue())))
+ topicCreationFutures.all().get() // wait on the future to make sure create
topic call finishes
}
def validateTopic(topic: String, expectedPartitionCount: Int) {
var done = false
var retries = 0
- while (!done && retries < 100) {
+ while (!done && retries < 10) {
try {
- if (!zkClient.topicExists(topic)) {
- System.err.println("Test topic %s not found. Waiting and
retrying." format topic)
- retries += 1
- Thread.sleep(500)
- }
-
- val topicDescription =
-
adminClient.describeTopics(JavaConverters.asJavaCollectionConverter(Set(topic)).asJavaCollection)
- .all().get().get(topic)
+ val topicDescriptionFutures = adminClient.describeTopics(
+
JavaConverters.asJavaCollectionConverter(Set(topic)).asJavaCollection).all()
+ val topicDescription = topicDescriptionFutures.get(500,
TimeUnit.MILLISECONDS)
+ .get(topic)
done = expectedPartitionCount == topicDescription.partitions().size()
Review comment:
1. Nope.
2. Thanks for bringing it up. I found a corner case with current
implementation. I changed it a little without introducing additional control
flow statements.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
