KAFKA-6970: All standard state stores guarded with read only wrapper

[nizhikov]

Committer Checklist (excluded from commit message)

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

[mjsax]

Tests fail with

java.lang.UnsupportedOperationException: Global store is read only

Seems this PR breaks something.

[mjsax]

Also note: only init() and close() should be guarded. Regular read/write access must be preserved.

[vvcephei]

Hey @nizhikov ,

Thanks for picking this up!

After reading the ticket, I think the current change doesn't quite do what we want...

James (the reporter) mentioned this method being used for getting a state store for use in Transformer and Processor, both of which need read-write access to the store. It sounds like the idea is to protect users from calling methods that are intended only to be called from the framework.

This is similar to the KAFKA-7420, as you mentioned in the ticket, but local state stores may safely be read/write, where as global state stores must be read-only.

Plus, James specifically called out protecting init() and close(), neither of which are protected by the current decorators.

I'd propose:

1. throw UnsupportedOperationException in StateStoreReadOnlyDecorator methods init and close
2. change the ERROR_MESSAGE to "This method may only be called by Kafka Steams"
3. add new XXReadWriteDecorator classes for the three kinds of stores
4. use the ReadOnly decorators for global stores and ReadWrite decorators for local stores

How does that sound?

(edit) I didn't refresh, so I didn't see @mjsax 's prior comments... sorry about the overlap.

[nizhikov]

Hello, @mjsax @vvcephei . Thanks for a quick review and detailed explanation of the ticket.
I will rework PR in a few hours.

[nizhikov]

I reworked PR as follows:

1. Usage of existing AbstractStateStore implements WrappedStateStore.

Kafka, already have an abstract class for a StateStore implementations that wraps some other state store.
Also, we have marker interface for such stores WrappedStateStore.
If wrapping state store doesn't implement WrappedStateStore it leads to wrong TupleForwarder behaviour(TupleForwarder#cachedStateStore(StateStore) method).
Wrong TupleForward behaviour leads to org.apache.kafka.streams.scala.WordCountTest fails.

2. Fixed TupleForwarder to work with several WrappedStateStore that wraps each other.

Right now TupleForwarder#cachedStateStore(StateStore) works correctly only if we have one layer of wrapped state store(and my patch introduce second). I fix this method to work with any count of wrapping state stores.

3. I also change StateStoreReadOnlyDecorator to extends AbstractStateStore(and implements WrappedStateStore).

If we don't extend AbstractStateStore we still have green tests(see #5865), but it seems that we must implements WrappedStateStore for all decorators.

[mjsax]

Thanks for the PR. Couple of comments/nits.

[vvcephei]

Looks good! I had one nitpick...

Thanks for this,
-John

[vvcephei]

Good catch!

[vvcephei]

The java 11 test failure is unrelated:

kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsToEarliestOnOneTopicAndPartition

Error Message
java.lang.AssertionError: expected:<Map(bar-0 -> 50, bar-1 -> 0)> but was:<Map(bar-0 -> 50, bar-1 -> 50)>
Stacktrace
java.lang.AssertionError: expected:<Map(bar-0 -> 50, bar-1 -> 0)> but was:<Map(bar-0 -> 50, bar-1 -> 50)>
	at org.junit.Assert.fail(Assert.java:88)
	at org.junit.Assert.failNotEquals(Assert.java:834)
	at org.junit.Assert.assertEquals(Assert.java:118)
	at org.junit.Assert.assertEquals(Assert.java:144)
	at kafka.admin.ResetConsumerGroupOffsetTest.resetAndAssertOffsetsCommitted(ResetConsumerGroupOffsetTest.scala:388)
	at kafka.admin.ResetConsumerGroupOffsetTest.testResetOffsetsToEarliestOnOneTopicAndPartition(ResetConsumerGroupOffsetTest.scala:247)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/1162/

[vvcephei]

Retest this, please.

[nizhikov]

@mjsax I fixed all your comments. Tests are green. Please, review.

[mjsax]

Merged to trunk.

Thanks for the PR @nizhikov!

[nizhikov]

@mjsax @vvcephei

Guys, thanks for the review and ticket description. I Appreciate it.

[vvcephei]

No problem, @nizhikov ; thanks for the contribution!