MINOR: Make assignment expectation explicit in testConsumptionWithBrokerFailures
Author: Jason Gustafson
Reviewers: Ismael Juma
Closes from hachikuji/bouncetest-assert-assignment
@@ -100,12 +100,15 @@ class ConsumerBounceTest extends IntegrationTestHarness with Logging {
scheduler.start()
while (scheduler.isRunning.get()) {
- for (record <- consumer.poll(100).asScala) {
+ val records = consumer.poll(100).asScala
+ assertEquals(Set(tp), consumer.assignment.asScala)
+
+ for (record <- records) {
assertEquals(consumed, record.offset())
consumed += 1
}
- try {
+ if (records.nonEmpty) {
consumer.commitSync()
assertEquals(consumer.position(tp), consumer.committed(tp).offset)
@@ -113,10 +116,6 @@ class ConsumerBounceTest extends IntegrationTestHarness with Logging {
consumer.seekToBeginning(Collections.emptyList())
consumed = 0
}
- } catch {
- // TODO: should be no need to catch these exceptions once KAFKA-2017 is
- // merged since coordinator fail-over will not cause a rebalance
- case _: CommitFailedException =>
}
}
scheduler.shutdown()
