cameronlee314 commented on a change in pull request #951: SAMZA-2127: Upgrade
to Kafka 2.0
URL: https://github.com/apache/samza/pull/951#discussion_r267876940
##########
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. Can `topicDescription` be null?
2. If done is true, then can we break out of this loop?
----------------------------------------------------------------
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
