KAFKA-7893 Refactor ConsumerBounceTest to reuse functionality from BaseConsumerTest by stanislavkozlovski · Pull Request #6238 · apache/kafka

stanislavkozlovski · 2019-02-07T21:36:48Z

This PR should help address the flakiness in the ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup test (https://issues.apache.org/jira/browse/KAFKA-7965). I tested this locally and have verified it significantly reduces flakiness - 25/25 tests now pass. Running the test 25 times in trunk, I'd get 18/25 passes.

It does so by reusing the less-flaky consumer integration testing functionality inside BaseConsumerTest. Most notably, the test now makes use of the ConsumerAssignmentPoller class - each consumer now polls non-stop rather than the more batch-oriented polling we had in ConsumerBounceTest#waitForRebalance().

stanislavkozlovski · 2019-02-08T17:26:12Z

retest this please

hachikuji

Nice cleanup! Left a few comments.

hachikuji · 2019-02-13T18:53:15Z

Not sure why this should be a requirement. I think validateGroupAssignment can still be satisfied as long as the subscription is non-empty.

isPartitionAssignmentValid(assignments, subscriptions)

* Check whether partition assignment is valid * Assumes partition assignment is valid if * 1. Every consumer got assigned at least one partition * 2. Each partition is assigned to only one consumer * 3. Every partition is assigned to one of the consumers

It checks

val allNonEmptyAssignments = assignments.forall(assignment => assignment.nonEmpty) if (!allNonEmptyAssignments) { // at least one consumer got empty assignment return false }

I don't think it will pass if that assertion doesn't itself

hachikuji · 2019-02-13T19:00:58Z

Seems like we should take either topicsToSubscribe or partitionsToAssign, but not both. Maybe we can have two separate constructors. Also, subscribe below should probably throw an illegal state error if partitionsToAssign is not empty.

I added two new secondary constructors - can you check?

hachikuji · 2019-02-13T19:02:50Z

Extending BaseConsumerTest means we inherit its test cases (i.e. testSimpleConsumption and testCoordinatorFailover). I wonder if we should move these tests somewhere else.

We have 6 test classes that make use of those test cases. Only ConsumerBounceTest is redundant. I don't think it hurts that much.

We could create a new class BasicConsumerTests which defines those tests if you think that's a good approach

Yeah, it is not that big of a deal. Just it's a bit annoying to extend the build time even more because of tests we know are redundant. I think I slightly prefer your suggestion for BasicConsumerTests. Or maybe we can call the new class BaseConsumerTest and rename the existing one to AbstractConsumerTest?

The use of inheritance in these test cases has been a bit problematic in general. It would probably be better to split out any shared utility logic into a separate utility class. We don't have to do that here though.

I like your thinking. Let's leave this as it is for the time being so we can merge this PR and reduce the flakiness of ConsumerBounceTest.

I've opened up a follow-up JIRA to track this - https://issues.apache.org/jira/browse/KAFKA-8176

hachikuji · 2019-02-13T19:04:51Z

Could probably get rid of this variable and just use kickedOutConsumerIdx.foreach.

stanislavkozlovski · 2019-02-22T15:34:43Z

@hachikuji apologies for squashing all the commits into one - I had to rebase and did not want to deal fixing merge conflicts with each commit. I shouldn't have done that as it now makes your review not as straightforward

Inheritance is ConsumerBounceTest->BaseConsumerTest->BaseRequestTest->IntegrationTestHarness Also refactored testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup to use consumer pollers from BaseConsumerTest Refactor more tests to use ConsumerPollers testConsumerReceivesFatalExceptionWhenGroupPassesMaxSize and testClose specifically Refactor rest of ConsumerBounceTest tests Remove unused imports Make the code a bit cleaner Remove unused KafkaConfig import Address PR comments

…sumerPoller

stanislavkozlovski · 2019-02-25T14:37:23Z

retest this please

stanislavkozlovski · 2019-02-25T18:00:08Z

retest this please

stanislavkozlovski · 2019-02-25T18:00:17Z

(I am also running the tests locally)

Adds a new protected method modifyConfigs, mainly to allow the inheriting BaseRequestTest to change the way it handles configs in a more succinct way and SaslPlaintextConsumerTest to override its wanted listeners Also rename BaseRequestTest#numBrokers to serverCount. This consolidates the use of the same method in IntegrationTestHarness

stanislavkozlovski · 2019-02-26T09:39:11Z

The SaslPlainPlaintextConsumerTest class had problems due to not initializing its config listeners properly. This is because the BaseRequestTest overrides generateConfigs.
I generalized the code a bit more to allow all classes to modify exactly what they want

stanislavkozlovski · 2019-02-27T10:10:44Z

retest this please

hachikuji · 2019-02-28T21:04:11Z

retest this please

hachikuji · 2019-02-28T21:36:41Z

retest this please

ijuma · 2019-03-01T17:03:29Z

@stanislavkozlovski you should use merge instead of rebase.

ijuma · 2019-03-01T17:07:49Z

+  override def serverCount: Int = 3

  // If required, override properties by mutating the passed Properties object
  protected def propertyOverrides(properties: Properties) {}


I think this name doesn't quite work generally since IntegrationTestHarness exposes brokers and clients. So it should probably be renamed to make it clear that it's only for broker properties.

That's a great catch, thanks

stanislavkozlovski · 2019-03-08T14:45:44Z

retest this please

stanislavkozlovski · 2019-03-11T10:50:45Z

retest this please

stanislavkozlovski · 2019-03-11T11:07:40Z

Testing this locally, I see the ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup gets stuck at the very last sendRecords() call.
Each producer seems to throw:

org.apache.kafka.common.errors.TimeoutException: Topic group-max-size-test not present in metadata after 60000 ms.

The issue was that we would not call restartDeadBrokers() after we had found a kicked-out consumer and were essentially running a reduced cluster.

hachikuji · 2019-03-15T18:02:01Z

retest this please

stanislavkozlovski · 2019-03-16T06:59:51Z

Strange, I could have sworn this built and passes checkstyle/findbugs locally. Sorry for the noise. I will rebase with master as well now as there seem to be conflicting files

hachikuji

Sorry for the delay. Left a few additional comments.

hachikuji · 2019-03-25T16:36:15Z

 class AuthorizerIntegrationTest extends BaseRequestTest {

-  override def numBrokers: Int = 1
+  override def serverCount: Int = 1


nit: we're a little inconsistent with naming (broker vs server). Perhaps this should be brokerCount to go along with brokerPropertyOverrides?

hachikuji · 2019-03-25T16:37:58Z

+  val group = "my-test"
  val producerClientId = "ConsumerTestProducer"
  val consumerClientId = "ConsumerTestConsumer"
+  val GroupMaxSessionTimeoutMs = 30000L


nit: why is only this one capitalized?

hachikuji · 2019-03-25T16:41:53Z


+
+  protected def createConsumerWithGroupId(groupId: String): KafkaConsumer[Array[Byte], Array[Byte]] = {
+    consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId)


It's a little surprising to find that we mutate consumerConfig here. So if you called createConsumerWithGroupId first and then createConsumer, the first call would affect the second call. Maybe instead we could duplicate consumerConfig and override group id?

Good call. I'll make use of the configOverrides var in createConsumer()

hachikuji · 2019-03-25T16:42:15Z

+    * @param consumerPollers current consumer pollers
+    * @param topicsToSubscribe topics to which new consumers will subscribe to
+    * @param subscriptions set of all topic partitions
+  */


nit: misaligned

hachikuji · 2019-03-25T16:45:55Z

  /**
   * Check whether partition assignment is valid
-   * Assumes partition assignment is valid iff
+   * Assumes partition assignment is valid if


The original was probably intentional, if pedantic.

hachikuji · 2019-03-25T16:58:42Z

Yeah, it is not that big of a deal. Just it's a bit annoying to extend the build time even more because of tests we know are redundant. I think I slightly prefer your suggestion for BasicConsumerTests. Or maybe we can call the new class BaseConsumerTest and rename the existing one to AbstractConsumerTest?

The use of inheritance in these test cases has been a bit problematic in general. It would probably be better to split out any shared utility logic into a separate utility class. We don't have to do that here though.

rename serverCount => brokerCount fix var naming capitalization fix javadoc indentation and typo

hachikuji · 2019-04-04T22:27:30Z

retest this please

hachikuji

LGTM. Merging to trunk.

* apache/trunk: MINOR: Add security considerations for remote JMX in Kafka docs (apache#6544) MINOR: Remove redundant access specifiers from metrics interfaces (apache#6527) MINOR: Correct KStream documentation (apache#6552) KAFKA-8013; Avoid underflow when reading a Struct from a partially correct buffer (apache#6340) KAFKA-8058: Fix ConnectClusterStateImpl.connectors() method (apache#6384) MINOR: Move common consumer tests out of abstract consumer class (apache#6548) KAFKA-8168; Add a generated ApiMessageType class KAFKA-7893; Refactor ConsumerBounceTest to reuse functionality from BaseConsumerTest (apache#6238) MINOR: Tighten up metadata upgrade test (apache#6531) KAFKA-8190; Don't update keystore modification time during validation (apache#6539) MINOR: Fixed a few warning in core and connects (apache#6545) KAFKA-7904; Add AtMinIsr partition metric and TopicCommand option (KIP-427) MINOR: fix throttling and status in ConnectionStressWorker KAFKA-8090: Use automatic RPC generation in ControlledShutdown KAFKA-6399: Remove Streams max.poll.interval override (apache#6509) KAFKA-8126: Flaky Test org.apache.kafka.connect.runtime.WorkerTest.testAddRemoveTask (apache#6475) HOTFIX: Update unit test for KIP-443 KAFKA-7190: KIP-443; Remove streams overrides on repartition topics (apache#6511) KAFKA-8183: Add retries to WorkerUtils#verifyTopics (apache#6532) KAFKA-8181: Removed Avro topic from TOC on kafka (apache#6529)

…aseConsumerTest (apache#6238) This PR should help address the flakiness in the ConsumerBounceTest#testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup test (https://issues.apache.org/jira/browse/KAFKA-7965). I tested this locally and have verified it significantly reduces flakiness - 25/25 tests now pass. Running the test 25 times in trunk, I'd get `18/25` passes. It does so by reusing the less-flaky consumer integration testing functionality inside `BaseConsumerTest`. Most notably, the test now makes use of the `ConsumerAssignmentPoller` class - each consumer now polls non-stop rather than the more batch-oriented polling we had in `ConsumerBounceTest#waitForRebalance()`. Reviewers: Jason Gustafson <jason@confluent.io>

hachikuji reviewed Feb 13, 2019

View reviewed changes

stanislavkozlovski added 2 commits February 22, 2019 15:37

Check proper partitionsToAssign collection before throwing ISE in Con…

d1a9d05

…sumerPoller

stanislavkozlovski added 2 commits February 25, 2019 18:38

Fix failing tests due to incorrect consumer group usage

7f1994c

Call super().modifyProps in SaslPlainPlaintextConsumerTest

9e5f7af

ijuma reviewed Mar 1, 2019

View reviewed changes

ijuma added the tests Test fixes (including flaky tests) label Mar 4, 2019

Rename 'propertyOverrides' -> 'brokerPropertyOverrides'

521ae06

Add missing restartDeadBrokers() call

145deee

Merge branch 'trunk' into KAFKA-7893-refactor-consumer-bounce-test

a396fff

hachikuji reviewed Mar 25, 2019

View reviewed changes

Address PR comments

a2acc8d

rename serverCount => brokerCount fix var naming capitalization fix javadoc indentation and typo

hachikuji approved these changes Apr 5, 2019

View reviewed changes

hachikuji merged commit cc4fde3 into apache:trunk Apr 5, 2019



		protected def createConsumerWithGroupId(groupId: String): KafkaConsumer[Array[Byte], Array[Byte]] = {
		consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId)

Conversation

stanislavkozlovski commented Feb 7, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

stanislavkozlovski commented Feb 8, 2019

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji Feb 13, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stanislavkozlovski commented Feb 22, 2019

Uh oh!

stanislavkozlovski commented Feb 25, 2019

Uh oh!

stanislavkozlovski commented Feb 25, 2019

Uh oh!

stanislavkozlovski commented Feb 25, 2019

Uh oh!

stanislavkozlovski commented Feb 26, 2019

Uh oh!

stanislavkozlovski commented Feb 27, 2019

Uh oh!

hachikuji commented Feb 28, 2019

Uh oh!

hachikuji commented Feb 28, 2019

Uh oh!

ijuma commented Mar 1, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stanislavkozlovski commented Mar 8, 2019

Uh oh!

stanislavkozlovski commented Mar 11, 2019

Uh oh!

stanislavkozlovski commented Mar 11, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji commented Mar 15, 2019

Uh oh!

stanislavkozlovski commented Mar 16, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stanislavkozlovski commented Feb 7, 2019 •

edited

Loading

hachikuji Feb 13, 2019 •

edited

Loading

stanislavkozlovski commented Mar 11, 2019 •

edited

Loading

stanislavkozlovski commented Mar 16, 2019 •

edited

Loading