HOTFIX: skip listOffsets request for newly created changelog topics

[ableegoldman]

This seems to be the final piece in the EosBetaUpgradeTest flakiness puzzle. From time to time the listOffsets request would fail during the initial startup rebalance, causing a probing rebalance to be scheduled for 10 min later (rather than immediately after). The assignor was newly creating the changelogs and then almost immediately calling listOffsets which then failed as the internal topics were not yet ready.

To avoid this race condition, we keep track of which changelogs we just created and skip fetching the end offsets for them. If a changelog did not exist before this moment, we can safely assume that its end offset is zero.

[ableegoldman]

call for review any of @guozhangwang @vvcephei @cadonna @mjsax

[ableegoldman]

Am I crazy or is this entire block not actually doing anything?

[mjsax]

It should do some sanity checks -- but I agree it's not easy to grok... Maybe we also incrementally refactored the code and this method become useless? All tests passed?

[ableegoldman]

I guess to be fair it was still validating that every topic had the numberOfPartitions set, but I thought it might make more sense to do this inside the InternalTopicManager since that checks the partitions anyways

[guozhangwang]

The logic indeed seem redundant to me.

[mjsax]

Retest this please

[mjsax]

I like immutability. Should we call this allExpetcedChangelogPartitions and introduce allExistingChangelogPartitions = allExpetcedChangelogPartitions - newlyCreatedChangelogs

[ableegoldman]

Fair enough

[mjsax]

Overall LGTM.

Can we extend unit test accordingly to verify that we set offset=0L for newly created topcis?

[ableegoldman]

@mjsax addressed your comments and added a unit test to verify that we don't ask the admin client to request offsets for newly created topics

[mjsax]

Don't we need to reply adminClient, too?

[ableegoldman]

It gets replayed during configuration (at the end of configureDefault below)

[mjsax]

Retest this please

[mjsax]

Retest this please

[mjsax]

Retest this please.

[mjsax]

Retest this please.

[ableegoldman]

Nooooooo EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta[true] still had one failure on the Java 8 build 😭

But it failed on a different line which seems more in line with real flakiness. FWIW I ran this locally 40 times without failures (technically 80 in total for both true/false variations) ... I think it's worth still merging this PR and we can continue investigating it from there.

java.lang.AssertionError: Did not receive all 20 records from topic multiPartitionOutputTopic within 60000 ms
Expected: is a value equal to or greater than <20>
     but: <5> was less than <20>
	at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
	at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.lambda$waitUntilMinKeyValueRecordsReceived$1(IntegrationTestUtils.java:563)
	at org.apache.kafka.test.TestUtils.retryOnExceptionWithTimeout(TestUtils.java:429)
	at org.apache.kafka.test.TestUtils.retryOnExceptionWithTimeout(TestUtils.java:397)
	at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(IntegrationTestUtils.java:559)
	at org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(IntegrationTestUtils.java:530)
	at org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.readResult(EosBetaUpgradeIntegrationTest.java:973)
	at org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.verifyCommitted(EosBetaUpgradeIntegrationTest.java:961)
	at org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest.shouldUpgradeFromEosAlphaToEosBeta(EosBetaUpgradeIntegrationTest.java:427)

[vvcephei]

Oops, never submitted my review. Feel free to ignore my minor suggestion.

[vvcephei]

It's not what you signed up for, but I'm wondering if we should at least submit a Jira to give some of these AdminClient methods a "full consistency" mode. In other words, since the command returns a future anyway, it would be nice to be able to tell it not to return until it can guarantee the topic will appear to be fully created on all brokers.

I'm mildly concerned that we're just kicking the can down the road a little ways with this PR. I.e., we let the assignment happen, but then some other metadata (or data) operation for that topic will just fail shortly thereafter.

More generally, we jump through a lot of hoops in our own tests to try and make sure that the topics are really, actually created (or deleted) before proceeding with the test, and I'm sure that our users also suffer from the same problem in their testing and production code.

[ableegoldman]

I think this race condition was particularly severe since we do the listOffsets request pretty much immediately after creating the topics, whereas whatever we're doing with that topic next will not be until the rebalance was completed.

AFAIK we've never had any users report subsequent operations failing after the first rebalance due to not-yet-fully-created topics, but it could have just slipped past us

[ableegoldman]

I do agree it would be useful though. Feel free to create a ticket :P

[vvcephei]

Thanks! Will do. I just wanted to bounce the idea off you first, in case it was stupid.

[mjsax]

Retest this please.

[mjsax]

Retest this please.

[ableegoldman]

Java 14 build passed, Java 8 din't build due to

11:29:13 # There is insufficient memory for the Java Runtime Environment to continue.
11:29:13 # Cannot create GC thread. Out of system resources.

[vvcephei]

Thanks, @ableegoldman !