MINOR: Clean up in DeleteConsumerGroupTest even if test fails
Author: Rajini Sivaram
Reviewers: Ismael Juma
Closes from rajinisivaram/MINOR-deleteconsumergrouptest-cleanup
@@ -163,7 +163,11 @@ class DeleteConsumerGroupTest extends KafkaServerTestHarness {
"Consumer group info on related topics should be deleted by DeleteAllConsumerGroupInfoForTopicInZK")
//produce events
val producer = TestUtils.createNewProducer(brokerList)
+ try {
produceEvents(producer, topic, List.fill(10)("test"))
+ } finally {
+ producer.close()
+ }
//consume events
val consumerProps = TestUtils.createConsumerProperties(zkConnect, group, "consumer")
@@ -173,11 +177,13 @@ class DeleteConsumerGroupTest extends KafkaServerTestHarness {
consumerProps.put("fetch.wait.max.ms", "0")
val consumerConfig = new ConsumerConfig(consumerProps)
val consumerConnector: ConsumerConnector = Consumer.create(consumerConfig)
+ try {
val messageStream = consumerConnector.createMessageStreams(Map(topic -> 1))(topic).head
consumeEvents(messageStream, 5)
consumerConnector.commitOffsets(false)
- producer.close()
+ } finally {
consumerConnector.shutdown()
+ }
TestUtils.waitUntilTrue(() => groupTopicOffsetAndOwnerDirsExist(dir),
"Consumer group info should exist after consuming from a recreated topic")
