KAFKA-8816: Make offsets immutable to users of RecordCollector.offsets by cpettitt-confluent · Pull Request #7223 · apache/kafka

cpettitt-confluent · 2019-08-19T19:57:27Z

Make offsets immutable to users of RecordCollector.offsets. Fix up an
existing case where offsets could be modified in this way. Add a simple
test to verify offsets cannot be changed externally.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

cpettitt-confluent · 2019-08-19T20:00:51Z

This should have the "streams" label, but I don't seem to have enough permissions to set it.

mjsax

While I agree in general, that RecordCollectorImpl#offsets is conceptually a private field, I am not sure if I understand why there is a bug. RecordCollectorImpl never reads offsets but only blindly updates the map on write in the send() callback. Hence, this PR does not seem toy change the end-of-end behavior? Can you elaborate?

mjsax · 2019-08-20T01:15:15Z

Would it be simpler to hand out a deep copy of the map directly?

That would be totally fine for the current usage pattern, where we have a single query that ends up copying the map anyway. However, if we ever ended up with other queries then those queries would pay the cost of the copy whether they need it or not. It would be a bit surprising that a copy is happening without looking at the implementation. My bias would be towards defensive coding without surprise performance impact.

Fair enough.

mjsax · 2019-08-20T01:16:08Z

Seem to be unused?

Yep, copy-paste error. This can be removed.

mjsax · 2019-08-20T01:16:52Z

nit: use assertThrows instead of try-fail-catch construct.

Thanks! It's been a while since I've coded Java and I forgot about assertThrows.

mjsax · 2019-08-20T01:17:28Z

nit: use assertThat instead

Will do for both instances.

mjsax · 2019-08-20T01:17:33Z

cpettitt-confluent

Will follow up with a patch to address your comments in test. If you feel strongly about copying the map in offsets let me know and I will make that change; otherwise I'll leave that as is.

cpettitt-confluent · 2019-08-20T02:15:33Z

Yep, copy-paste error. This can be removed.

cpettitt-confluent · 2019-08-20T02:15:41Z

Will do for both instances.

cpettitt-confluent · 2019-08-20T02:16:09Z

Thanks! It's been a while since I've coded Java and I forgot about assertThrows.

cpettitt-confluent · 2019-08-20T02:20:18Z

That would be totally fine for the current usage pattern, where we have a single query that ends up copying the map anyway. However, if we ever ended up with other queries then those queries would pay the cost of the copy whether they need it or not. It would be a bit surprising that a copy is happening without looking at the implementation. My bias would be towards defensive coding without surprise performance impact.

cpettitt-confluent · 2019-08-20T02:32:01Z

Re. why is this a bug, based on my understanding of the code:

Per the doc on RecordCollector the contract is that the value returned from offsets is the latest acks from the producer. Prior to this change it was possible to get the offsets map and change it directly, with no involvement of a producer. That seems to be a violation of the documentation and my understanding of the intent of the class. Anybody can get the offsets map, do some manipulation to it for its own purpose and accidentally change the internal state of RecordCollector in a non-obvious way. In fact, that is what is happening from StreamTask during commit prior to this patch.

cpettitt-confluent · 2019-08-20T02:32:47Z

BTW, thanks for the review @mjsax and for reminding me about assertThrows. I'll get a new patch up tomorrow.

cpettitt-confluent · 2019-08-20T15:26:12Z

New commit is up and ready for review. I couldn't tell if you prefer force push or squash at merge and I see both in other PRs. I opted for squash at merge, but if you prefer force push let me know and I will make it so :).

cpettitt-confluent · 2019-08-20T15:38:30Z

W.r.t observable behavior change, I reran with this patch and without and here is a difference in how checkpointing works:

Good:

[2019-08-19 15:59:34,197] INFO task [0_0] *** Writing checkpoint: {input-topic-0=15} (org.apache.kafka.streams.processor.internals.ProcessorStateManager:354)
[2019-08-19 15:59:34,332] INFO task [0_0] *** Writing checkpoint: {input-topic-0=35} (org.apache.kafka.streams.processor.internals.ProcessorStateManager:354)
[2019-08-19 15:59:34,443] INFO task [0_0] *** Writing checkpoint: {input-topic-0=53} (org.apache.kafka.streams.processor.internals.ProcessorStateManager:354)

Bad:

[2019-08-20 09:28:39,009] INFO task [0_0] *** Writing checkpoint: {input-topic-0=15} (org.apache.kafka.streams.processor.internals.ProcessorStateManager:354)
[2019-08-20 09:28:39,153] INFO task [0_0] *** Writing checkpoint: {input-topic-0=15} (org.apache.kafka.streams.processor.internals.ProcessorStateManager:354)
[2019-08-20 09:28:39,264] INFO task [0_0] *** Writing checkpoint: {input-topic-0=15} (org.apache.kafka.streams.processor.internals.ProcessorStateManager:355)

In the bad case the consumed offset is not checkpointed because we added the first value we saw to the record collector and never update it (putIfAbsent):

    @Override
    protected Map<TopicPartition, Long> activeTaskCheckpointableOffsets() {
        final Map<TopicPartition, Long> checkpointableOffsets = recordCollector.offsets();
        for (final Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
            checkpointableOffsets.putIfAbsent(entry.getKey(), entry.getValue());
        }

        return checkpointableOffsets;
    }

I have no idea if that manifests in bad behavior, but it doesn't look right and doesn't match the behavior in the non-optimized graph where the checkpoints increase for the changelog.

mjsax · 2019-08-21T23:36:08Z

Thanks for the details.

I have no idea if that manifests in bad behavior, but it doesn't look right and doesn't match the behavior in the non-optimized graph where the checkpoints increase for the changelog.

It does not sound like a correctness issue, but a performance issue. If the checkpoint does not advance, on restore KS would re-read/re-play more data than necessary from the changelog.

mjsax

LGTM.

Call for second review @guozhangwang @bbejeck @vvcephei @ableegoldman @cadonna @abbccdda

mjsax · 2019-08-21T23:38:10Z

MIght be worth to back-port this fix to older versions... Maybe back to 2.0? (or even 1.0?)

mjsax · 2019-08-21T23:38:45Z

Java 11 / 2.12 and 2.13 passed.
Java8: kafka.api.SaslOAuthBearerSslEndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl

Retest this please

Make offsets immutable to users of RecordCollector.offsets. Fix up an existing case where offsets could be modified in this way. Add a simple test to verify offsets cannot be changed externally.

cpettitt-confluent · 2019-08-22T16:35:25Z

Seems like flaky tests. The first test passed this time and we picked up two connect test failures which were not there previously.

retest this please

cadonna

@cpettitt-confluent, Thank you for the PR.

Here my feedback.

cadonna · 2019-08-26T13:02:41Z

nit: This last check is not needed, since it verifies functionality of the Map returned by Collections.unmodifiableMap() and not of the code under test.

cadonna · 2019-08-26T13:03:30Z

Please remove ().

cadonna · 2019-08-26T13:47:26Z

FYI: Openend PR #7253 for a minor refactoring to activeTaskCheckpointableOffsets(). The PR should not conflict with this one. Reviews welcome.

cadonna · 2019-08-26T13:47:56Z

Retest this, please

cpettitt-confluent · 2019-08-26T13:54:04Z

Thanks for the feedback @cadonna! Good call on checking the map a second time. I will requery the offset directly from the collector, which more correctly completes verification. Patch coming shortly.

cpettitt-confluent · 2019-08-26T13:57:14Z

Ugh. Somehow my origin moved so this is pulling in a lot of irrelevant stuff. Let me see if I can clean it up.

cpettitt-confluent · 2019-08-26T14:00:39Z

All better.

Retest this please.

cadonna · 2019-08-26T14:03:35Z

+
+        assertThat(offsets.get(topicPartition), equalTo(2L));
+        assertThrows(UnsupportedOperationException.class, () -> offsets.put(new TopicPartition(topic, 0), 50L));
+


nit: I would remove this line

I would argue for keeping this because the change impacts the external behavior of the class. We're making a strong statement here: you will get an exception if you try to modify the contents of the returned map, you must copy this map if you want to make changes. This also distinguishes from the alternative approach we could have used in which we proactively copy the map for the user and where the user could have made a change to the map while still not impacting the underlying map. Given that this is externally facing and there is doc a couple levels up, I will fix that up.

Happy to discuss further if you strongly disagree.

cadonna · 2019-08-26T14:05:03Z

+        assertThat(offsets.get(topicPartition), equalTo(2L));
+        assertThrows(UnsupportedOperationException.class, () -> offsets.put(new TopicPartition(topic, 0), 50L));
+
+        // Verify that collector offsets were not updated


nit: I would remove this comment because the code it comments is clear enough.

Hah, this was probably for me more than anyone to highlight the subtle difference between this and the previous query. I'm fine pulling it out and since I have a doc change coming, I will make it so :).

mjsax · 2019-09-04T23:43:46Z

@guozhangwang Should we cherry-pick this to older branches? I would at least cherry-pick to 2.3, but maybe even back to 2.0. Thoughts?

#7223) Make offsets immutable to users of RecordCollector.offsets. Fix up an existing case where offsets could be modified in this way. Add a simple test to verify offsets cannot be changed externally. Reviewers: Bruno Cadonna <bruno@confluent.io>, Guozhang Wang <guozhang@confluent.io>, Matthias J. Sax <matthias@confluent.io>

guozhangwang · 2019-09-09T23:43:33Z

Cherry-picked all the way to 2.0.

mjsax added the streams label Aug 20, 2019

mjsax reviewed Aug 20, 2019

View reviewed changes

cpettitt-confluent commented Aug 20, 2019

View reviewed changes

mjsax approved these changes Aug 21, 2019

View reviewed changes

cpettitt-confluent added 2 commits August 22, 2019 08:42

KAFKA-8816: Make offsets immutable to users of RecordCollector.offsets

c540b38

Make offsets immutable to users of RecordCollector.offsets. Fix up an existing case where offsets could be modified in this way. Add a simple test to verify offsets cannot be changed externally.

Review fixes, to be squashed

c51b92c

cadonna reviewed Aug 26, 2019

View reviewed changes

More improvements based on review feedback

b44b9af

cpettitt-confluent force-pushed the kafka-8816 branch from 3971396 to b44b9af Compare August 26, 2019 13:59

cadonna approved these changes Aug 26, 2019

View reviewed changes

Tweak docs to reflect behavior change, cosmetic cleanup

2e0fd82

guozhangwang approved these changes Aug 26, 2019

View reviewed changes

guozhangwang merged commit 6b24b2e into apache:trunk Aug 26, 2019

cadonna mentioned this pull request Aug 26, 2019

KAFKA-8755: Fix state restore for standby tasks with optimized topology #7238

Merged

3 tasks

mjsax mentioned this pull request Sep 12, 2019

HOTFIX: fix compile error in 2.2 branch #7327

Merged


		assertThat(offsets.get(topicPartition), equalTo(2L));
		assertThrows(UnsupportedOperationException.class, () -> offsets.put(new TopicPartition(topic, 0), 50L));

Conversation

cpettitt-confluent commented Aug 19, 2019

Committer Checklist (excluded from commit message)

Uh oh!

cpettitt-confluent commented Aug 19, 2019

Uh oh!

mjsax 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!

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!

cpettitt-confluent 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!

cpettitt-confluent commented Aug 20, 2019

Uh oh!

cpettitt-confluent commented Aug 20, 2019

Uh oh!

cpettitt-confluent commented Aug 20, 2019

Uh oh!

cpettitt-confluent commented Aug 20, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax commented Aug 21, 2019

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

mjsax commented Aug 21, 2019

Uh oh!

mjsax commented Aug 21, 2019

Uh oh!

cpettitt-confluent commented Aug 22, 2019

Uh oh!

cadonna 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!

cadonna commented Aug 26, 2019

Uh oh!

cadonna commented Aug 26, 2019

Uh oh!

cpettitt-confluent commented Aug 26, 2019

Uh oh!

cpettitt-confluent commented Aug 26, 2019

Uh oh!

cpettitt-confluent commented Aug 26, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

cpettitt-confluent commented Aug 20, 2019 •

edited

Loading