MINOR: Relax Unsupported Version check on BrokerCompatibilityTest

[guozhangwang]

In BrokerCompatibilityTest.java, when older versioned broker is used (0.10.1, 0.10.2), LIST_OFFSET is not supported as well. Hence in the verification phase, there is a possibility that consumer hit the UnsupportedVersionException earlier than Streams actually hits it:

rg.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support LIST_OFFSETS with version in range [2,3]. The supported range is [0,1].

While the test is waiting for

ATAL: An unexpected exception org.apache.kafka.common.errors.UnsupportedVersionException: Cannot create a v0 FindCoordinator request because we require features supported only in 1 or later.

Call seek with specific offset would avoid sending LIST_OFFSET.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[guozhangwang]

@mjsax @vvcephei @bbejeck

The recent commit of using new poll() seems have increased the likelihood of this issue.

[guozhangwang]

https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1783

[mjsax]

A little unclear to me. The consumer should be backward compatible and not throw any UnsupportedVersionException... otherwise we broke something using the new poll(Duration) call (ie, poll(Duration) does something invalid introducing a regression). Or do I miss something?

[vvcephei]

Interesting... Why do you say the new poll call increases the likelihood of the issue? Or is that just empirical?

FWIW, the change looks good to me.

[guozhangwang]

@mjsax @vvcephei Here is what I'm suspecting: first of all, this race condition should have been existed since day one because there is no synchronization between streams.start and the loopUntilRecordReceived call. I.e. either Streams' leader failed to create topic during the rebalance, and printed the expected error message, or the Consumer failed to list offsets when starting up and printed the unexpected error message.

In the old API, we will call ensureCoordinatorReady with Long.MAX_VALUE as the timeout, and hence we are effectively doing long polling on the underlying network client, while with the new API we set a relatively low timeout in ensureCoordinatorReady and will fallback and retry the request pretty quickly. My un-validated suspicion is that this caused the consumer to 1) joined the group and 2) refreshed metadata quicker than before, and hence continue to updateFetchPositions and hence hit the error quicker, and that's why we now see this error more often than before.

[guozhangwang]

After some further discussion with @hachikuji I've decided to change the test, to relax the expect error message so that both producer's error:

[2018-06-08 01:24:55,313] DEBUG [Producer clientId=kafka-streams-system-test-broker-compatibility-bc8152cd-22b7-437f-8f19-7c56a7d40075-StreamThread-1-0_0-producer, transactionalId=kafka-streams-system-test-broker-compatibility-0_0] Version mismatch when attempting to send (type=FindCoordinatorRequest, coordinatorKey=kafka-streams-system-test-broker-compatibility-0_0, coordinatorType=TRANSACTION) with correlation id 2 to -1 (org.apache.kafka.clients.NetworkClient)
org.apache.kafka.common.errors.UnsupportedVersionException: Cannot create a v0 FindCoordinator request because we require features supported only in 1 or later.

and consumer's error

[2018-06-08 01:24:55,316] DEBUG [Consumer clientId=kafka-streams-system-test-broker-compatibility-bc8152cd-22b7-437f-8f19-7c56a7d40075-StreamThread-1-consumer, groupId=kafka-streams-system-test-broker-compatibility] Version mismatch when attempting to send (type=ListOffsetRequest, replicaId=-1, partitionTimestamps={brokerCompatibilitySourceTopic-0=-2}, isolationLevel=READ_COMMITTED) with correlation id 29 to 1 (org.apache.kafka.clients.NetworkClient)
org.apache.kafka.common.errors.UnsupportedVersionException: The broker does not support LIST_OFFSETS with version in range [2,3]. The supported range is [0,1].

are expected.

[bbejeck]

LGTM overall.

Can we get a new system test for this?

[guozhangwang]

Yes, https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1785/ has just passed (ran 5 times).

[mjsax]

LGTM.