MINOR: Port RestoreIntegrationTest to the new protocol #20347
Conversation
github-actions
bot
added
triage
aliehsaeedii
left a comment
aliehsaeedii
left a comment
There was a problem hiding this comment.
Thanks, @RaidenE1. Except calling streams-related admin API, I don't see any other concern.
|
|
||
| consumer.commitSync(); | ||
| consumer.close(); | ||
| admin.alterConsumerGroupOffsets(appId, offsetsToCommit).all().get(); |
There was a problem hiding this comment.
Please call admin.alterStreamsGroupOffsets
| if (admin != null) { | ||
| admin.close(); | ||
| } |
There was a problem hiding this comment.
FYI: there's a utility method in Kafka called Utils.closeQuietly() which can make closing resources a one-liner.
|
|
||
| consumer.commitSync(); | ||
| consumer.close(); | ||
| admin.alterStreamsGroupOffsets(appId, offsetsToCommit).all().get(); |
There was a problem hiding this comment.
Don't we need to make this configurable? If the test runs with classic protocol we would still need to use the corresponding consumer group admit-call?
In general, it seems that the whole test must be parametrized, to run with both classic and streams protocol? Atm, it seem that the test is only running with classic, and thus using this streams specific call, should actually break the test?
|
|
||
| consumer.commitSync(); | ||
| } | ||
| } |
| @ParameterizedTest | ||
| @ValueSource(booleans = {true, false}) | ||
| public void shouldRestoreStateFromSourceTopicForReadOnlyStore(final boolean stateUpdaterEnabled) throws Exception { | ||
| @CsvSource({ |
There was a problem hiding this comment.
We should do the same for all other test case, like shouldRestoreNullRecord above. There is 8 test methods I did not miscount, but you enable the new protocol only for two.
|
|
||
| if (useNewProtocol) { | ||
| // For new protocol, we need to stop the streams instance before altering offsets | ||
| kafkaStreams.close(); |
There was a problem hiding this comment.
Probably best to use close(Duration timeout) overload, otherwise the test can get stuck
| } | ||
| }); | ||
| kafkaStreams.setGlobalStateRestoreListener(new TrackingStateRestoreListener(restored)); | ||
| kafkaStreams.start(); |
There was a problem hiding this comment.
There's a IntegrationTestUtils.startApplicationAndWaitUntilRunning(KafkaStreams instance) method you can use that returns your KS instance once it's in a running state which will simplify your logic here - no need for CountDownLatch
|
|
||
| // Restart the streams instance with a new startup latch | ||
| final CountDownLatch restartLatch = new CountDownLatch(1); | ||
| kafkaStreams = new KafkaStreams(builder.build(props), props); |
| assertThat(CloseCountingInMemoryStore.numStoresClosed(), equalTo(expectedAfterStreams2Close)); | ||
| } finally { | ||
| streams1.close(); | ||
| } |
There was a problem hiding this comment.
I know this was pre-existing but I have the same comment about closing with the overload that accepts a timeout.
|
|
||
| admin.alterStreamsGroupOffsets(appId, offsetsToCommit).all().get(); | ||
| } catch (final Exception e) { | ||
| throw new RuntimeException("Failed to set committed offsets", e); |
|
merged #20347 into trunk |
5 participants
The test uses regular consumer to commit offsets. New protocol requires
a streams consumer since we are using streams groups, otherwise we run
into group ID conflicts.
Followed the addition of the KafkaAdmin interface for setting offsets, a
Kafka Admin client is created and used the interface in to set the
committed offsets instead of instantiating a consumer.
Also enable all tests for stream new protocol.
Reviewers: Alieh Saeediasaeedi@confluent.io, Kirk True
ktrue@confluent.io, Matthias Sax mjsax@apache.org, Bill Bejeck
bbejeck@apache.org