KAFKA-10167: use the admin client to read end-offset by guozhangwang · Pull Request #8876 · apache/kafka

guozhangwang · 2020-06-16T00:44:55Z

Since admin client allows use to use flexible offset-spec, we can always set to use read-uncommitted regardless of the eos config.

Committer Checklist (excluded from commit message)

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

mjsax · 2020-06-16T01:34:33Z

        try {
-            return restoreConsumer.endOffsets(partitions);
-        } catch (final TimeoutException e) {
+            if (adminClient != null) {


Why do we need this distinction? Seems we set adminClient in any case and it should never be null?

I also do not understand the distinction. When would we want to call restoreConsumer.endOffsets(partitions)?

req: Could you add a test (or adapt an existing one) and verify whether during restore() a call to adminClient.listOffsets() with isolation level READ_UNCOMMITTED is done?

This is just a hack-around: I will always use admin-client by passing it via the constructor.

I will add a unit test.

Should we remove this check then?

I think we can remove this null check? If you really want to add one, add it to the constructor?

this.adminClient = Objects.notNull(adminClient);

I just saw that we kinda need to pass null as AdminClient in TDD -- however, StoreChangelogReader#restore() is never called by TTD, and thus I still think we should remove the null check and fail hard with a NPE as it would indicate a bug if adminClient is null in a regular deployment.

mjsax · 2020-06-16T01:37:21Z

    final ConsumerRebalanceListener rebalanceListener;
    final Consumer<byte[], byte[]> mainConsumer;
    final Consumer<byte[], byte[]> restoreConsumer;
+    final Admin adminClient;


Why is adminClient not final any longer? We still pass it into the StreamThread constructor.

You mean private, right? It is not private anymore.
Instead of making the adminClient field package private for testing, I would either add a setter setAdmin() to MockClientSupplier or instantiate the admin in a private field of the MockClientSupplier and use the existing getter to set up the admin, or instantiate the admin in a public field of the MockClientSupplier and accessing it directly to set it up (similarly to the producer and consumer).

Yeah I agree, will move all these variables to private and add getters.

cadonna

@guozhangwang Thanks very much for this!
This bug was driving me crazy! 😃

cadonna · 2020-06-16T12:07:58Z

    final ConsumerRebalanceListener rebalanceListener;
    final Consumer<byte[], byte[]> mainConsumer;
    final Consumer<byte[], byte[]> restoreConsumer;
+    final Admin adminClient;


You mean private, right? It is not private anymore.
Instead of making the adminClient field package private for testing, I would either add a setter setAdmin() to MockClientSupplier or instantiate the admin in a private field of the MockClientSupplier and use the existing getter to set up the admin, or instantiate the admin in a public field of the MockClientSupplier and accessing it directly to set it up (similarly to the producer and consumer).

cadonna · 2020-06-16T12:10:43Z


        final Consumer<byte[], byte[]> mainConsumer = clientSupplier.getConsumer(consumerConfigs);
        changelogReader.setMainConsumer(mainConsumer);
+        changelogReader.setAdminClient(adminClient);


Q: Why do you not pass the admin client in the constructor?

Yeah that's the plan, I was just hacking it around so that I do not need to change 30+ unit tests. Once it is confirmed to fix the issue I will refactor this PR.

cadonna · 2020-06-16T12:13:10Z

        try {
-            return restoreConsumer.endOffsets(partitions);
-        } catch (final TimeoutException e) {
+            if (adminClient != null) {


I also do not understand the distinction. When would we want to call restoreConsumer.endOffsets(partitions)?

cadonna · 2020-06-16T12:26:25Z

        try {
-            return restoreConsumer.endOffsets(partitions);
-        } catch (final TimeoutException e) {
+            if (adminClient != null) {


req: Could you add a test (or adapt an existing one) and verify whether during restore() a call to adminClient.listOffsets() with isolation level READ_UNCOMMITTED is done?

cadonna · 2020-06-16T12:27:06Z

        final StreamThread thread = createStreamThread("clientId", config, false);
        final MockConsumer<byte[], byte[]> mockConsumer = (MockConsumer<byte[], byte[]>) thread.mainConsumer;
        final MockConsumer<byte[], byte[]> mockRestoreConsumer = (MockConsumer<byte[], byte[]>) thread.restoreConsumer;
+        final MockAdminClient mockAdminClient = (MockAdminClient) thread.adminClient;


See my comment in StreamThread.

abbccdda · 2020-06-16T02:08:48Z

        this.mainConsumer = consumer;
    }

+    void setAdminClient(final Admin adminClient) {


This is not used.

…-offsets-uncommitted

mjsax · 2020-06-17T02:47:56Z

+                return result.all().get().entrySet().stream().collect(
+                        Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().offset()));
+            } else {
+                // we only fall back to use restore consumer if admin client is not set in TTD


Can this case ever apply? TTD should never restore any task from my understanding.

TTD would still call this function though... after some thinking I've decided to add a mock-admin to streams:test-utils (the existing mock is in clients:test and hence cannot be relied on).

mjsax · 2020-06-17T02:59:51Z

-        } catch (final TimeoutException e) {
+            if (adminClient != null) {
+                final ListOffsetsResult result = adminClient.listOffsets(partitions.stream().collect(
+                        Collectors.toMap(Function.identity(), tp -> OffsetSpec.latest())));


Should we set isolation.level explicitly? -- In case the default if ever changed, we would have a safe guard?

Good point, will do.

mjsax

Overall LGTM.

ableegoldman · 2020-06-17T23:30:23Z

        try {
-            return restoreConsumer.endOffsets(partitions);
-        } catch (final TimeoutException e) {
+            final ListOffsetsResult result = adminClient.listOffsets(


Did you check out the new methods in ClientUtils?

mjsax · 2020-06-18T00:05:35Z

@guozhangwang Lot of failing test that seem related?

guozhangwang · 2020-06-18T00:07:25Z

@guozhangwang Lot of failing test that seem related?

Hmm. My local tests passed twice.. I will double check.

…-offsets-uncommitted

mjsax · 2020-06-18T01:38:01Z

                    mockWallClockTime,
                    streamsConfig,
                    logContext,
+                    createAdminClient(processorTopology.storeToChangelogTopic()),


This is the only thing I am wondering about. Why not just pass null?

After thinking a bit more I agree with you now: I removed the mock admin client and instead just pass a mock register (since the register function may still be called)

mjsax

Thanks @guozhangwang. LGTM. Free free to merge after Jenkins passed.

Since admin client allows use to use flexible offset-spec, we can always set to use read-uncommitted regardless of the EOS config. Reviewers: A. Sophie Blee-Goldman <sophie@confluent.io>, Bruno Cadonna <bruno@confluent.io>, Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

guozhangwang · 2020-06-18T18:29:43Z

Cherry-pick to 2.6.

* 'trunk' of github.com:apache/kafka: KAFKA-10168: fix StreamsConfig parameter name variable (apache#8865) MINOR: code cleanup for inconsistent naming (apache#8871) KAFKA-10138: Prefer --bootstrap-server for reassign_partitions command in ducktape tests (apache#8898) KAFKA-10185: Restoration info logging (apache#8896) KAFKA-9891: add integration tests for EOS and StandbyTask (apache#8890) MINOR: Reduce build time by gating test coverage plugins behind a flag (apache#8899) KAFKA-10141; Add more detail to log segment delete messages (apache#8850) KAFKA-10113; Specify fetch offsets correctly in `LogTruncationException` (apache#8822) KAFKA-10167: use the admin client to read end-offset (apache#8876) MINOR: Upgrade ducktape to 0.7.8 (apache#8879) KAFKA-10123; Fix incorrect value for AWAIT_RESET#hasPosition (apache#8841) KAFKA-9896: fix flaky StandbyTaskEOSIntegrationTest (apache#8883) MINOR: clean up unused checkstyle suppressions for Streams (apache#8861) MINOR: reuse toConfigObject(Map) to generate Config (apache#8889) MINOR: Upgrade jetty to 9.4.27.v20200227 and jersey to 2.31 (apache#8859) MINOR: Fix flaky HighAvailabilityTaskAssignorIntegrationTest (apache#8884) KAFKA-10147 MockAdminClient#describeConfigs(Collection<ConfigResource>) is unable to handle broker resource (apache#8853) KAFKA-10165: Remove Percentiles from e2e metrics (apache#8882) # Conflicts: # core/src/main/scala/kafka/log/Log.scala

check

b9a27a8

mjsax added the streams label Jun 16, 2020

mjsax reviewed Jun 16, 2020

View reviewed changes

guozhangwang changed the title ~~KAFKA-10167: use the admin client to read end-offset~~ KAFKA-10167: use the admin client to read end-offset 【WIP】 Jun 16, 2020

cadonna reviewed Jun 16, 2020

View reviewed changes

abbccdda reviewed Jun 16, 2020

View reviewed changes

guozhangwang added 4 commits June 16, 2020 10:55

Merge branch 'trunk' of https://github.com/apache/kafka into KXXX-end…

a322765

…-offsets-uncommitted

continue revamping

b4589e6

Merge branch 'trunk' of https://github.com/apache/kafka into KXXX-end…

a1a4bb6

…-offsets-uncommitted

github comments

7f81699

guozhangwang changed the title ~~KAFKA-10167: use the admin client to read end-offset 【WIP】~~ KAFKA-10167: use the admin client to read end-offset Jun 17, 2020

mjsax reviewed Jun 17, 2020

View reviewed changes

guozhangwang added 3 commits June 17, 2020 11:01

add the mock-admin client to ttd

2fd0ea5

copy past the mock admin client

67bf693

github comments

6934cd6

ableegoldman reviewed Jun 17, 2020

View reviewed changes

guozhangwang added 2 commits June 17, 2020 18:11

not use changelog at all

ebe86f9

Merge branch 'trunk' of https://github.com/apache/kafka into KXXX-end…

097dbca

…-offsets-uncommitted

mjsax reviewed Jun 18, 2020

View reviewed changes

github comments

8d80cce

mjsax approved these changes Jun 18, 2020

View reviewed changes

guozhangwang merged commit d8cc6fe into apache:trunk Jun 18, 2020

Conversation

guozhangwang commented Jun 16, 2020

Committer Checklist (excluded from commit message)

Uh oh!

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!

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!

cadonna left a comment • 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!

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!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax commented Jun 18, 2020

Uh oh!

guozhangwang commented Jun 18, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jun 18, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

cadonna left a comment •

edited

Loading