[GitHub] [kafka] vvcephei commented on a change in pull request #9388: KAFKA-10562: Properly invoke new StateStoreContext init

[GitBox <gi...@apache.org>]

vvcephei commented on a change in pull request #9388:
URL: https://github.com/apache/kafka/pull/9388#discussion_r500715702



##########
File path: streams/examples/src/test/java/org/apache/kafka/streams/examples/wordcount/WordCountProcessorTest.java
##########
@@ -34,6 +34,7 @@
  * Demonstrate the use of {@link MockProcessorContext} for testing the {@link Processor} in the {@link WordCountProcessorDemo}.
  */
 public class WordCountProcessorTest {
+    @SuppressWarnings("deprecation") // TODO will be fixed in KAFKA-10437

Review comment:
       This ticket needs to go in to 2.7.0 also, but I split it out for reviewability.

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractReadOnlyDecorator.java
##########
@@ -45,12 +46,19 @@ public void flush() {
         throw new UnsupportedOperationException(ERROR_MESSAGE);
     }
 
+    @Deprecated
     @Override
     public void init(final ProcessorContext context,
                      final StateStore root) {
         throw new UnsupportedOperationException(ERROR_MESSAGE);
     }
 
+    @Override
+    public void init(final StateStoreContext context,
+                     final StateStore root) {
+        throw new UnsupportedOperationException(ERROR_MESSAGE);

Review comment:
       There are going to be a lot of duplicated init methods. It's not great, but hopefully we can drop the old API before too long.

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java
##########
@@ -47,9 +48,42 @@ public static StreamsMetricsImpl getMetricsImpl(final ProcessorContext context)
         return (StreamsMetricsImpl) context.metrics();
     }
 
+    /**
+     * Should be removed as part of KAFKA-10217
+     */
+    public static StreamsMetricsImpl getMetricsImpl(final StateStoreContext context) {
+        return (StreamsMetricsImpl) context.metrics();
+    }
+
     public static String changelogFor(final ProcessorContext context, final String storeName) {
         return context instanceof InternalProcessorContext
             ? ((InternalProcessorContext) context).changelogFor(storeName)
             : ProcessorStateManager.storeChangelogTopic(context.applicationId(), storeName);
     }
+
+    public static String changelogFor(final StateStoreContext context, final String storeName) {
+        return context instanceof InternalProcessorContext
+            ? ((InternalProcessorContext) context).changelogFor(storeName)
+            : ProcessorStateManager.storeChangelogTopic(context.applicationId(), storeName);
+    }
+
+    public static InternalProcessorContext asInternalProcessorContext(final ProcessorContext context) {
+        if (context instanceof InternalProcessorContext) {
+            return (InternalProcessorContext) context;
+        } else {
+            throw new IllegalArgumentException(
+                "This component requires internal features of Kafka Streams and must be disabled for unit tests."
+            );
+        }
+    }

Review comment:
       I replaced a lot of casts with this checked-cast method, which also lets us get rid of a lot of similar cast-checking blocks, which were inconsistently applied.

##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredKeyValueStore.java
##########
@@ -83,14 +85,40 @@
         this.valueSerde = valueSerde;
     }
 
+    @Deprecated
     @Override
     public void init(final ProcessorContext context,
                      final StateStore root) {
-        this.context = context;
+        this.context = context instanceof InternalProcessorContext ? (InternalProcessorContext) context : null;
         taskId = context.taskId().toString();
         initStoreSerde(context);
         streamsMetrics = (StreamsMetricsImpl) context.metrics();
 
+        registerMetrics();

Review comment:
       I wasn't able to extract out quite as much common code in the Metered implementations because they need to work regardless of whether the context is an InternalProcessorContext or whether it's a straight mock (for unit tests).

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/StateStore.java
##########
@@ -65,7 +65,11 @@
      *
      * @throws IllegalStateException If store gets registered after initialized is already finished
      * @throws StreamsException if the store's change log does not contain the partition
+     * @deprecated Since 2.7.0. Callers should invoke {@link this#init(StateStoreContext, StateStore)} instead.
+     *             Implementers may choose to implement this method for backward compatibility or to throw an
+     *             informative exception instead.
      */
+    @Deprecated

Review comment:
       Adding the deprecation tag right now lets us be sure we encountered all places this method appears in the codebase.

##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
##########
@@ -225,6 +225,7 @@ public String name() {
         return name;
     }
 
+    @Deprecated

Review comment:
       There are a handful of these also, just passing the deprecation on to the callers.

##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/KeyValueSegment.java
##########
@@ -45,10 +46,9 @@ public int compareTo(final KeyValueSegment segment) {
     }
 
     @Override
-    public void openDB(final ProcessorContext context) {
-        super.openDB(context);
+    public void openDB(final Map<String, Object> configs, final File stateDir) {
+        super.openDB(configs, stateDir);

Review comment:
       I was able to remove the type-dependency of the context by re-specifying the interface in terms of the only two properties it needed.

##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java
##########
@@ -47,9 +48,42 @@ public static StreamsMetricsImpl getMetricsImpl(final ProcessorContext context)
         return (StreamsMetricsImpl) context.metrics();
     }
 
+    /**
+     * Should be removed as part of KAFKA-10217
+     */
+    public static StreamsMetricsImpl getMetricsImpl(final StateStoreContext context) {
+        return (StreamsMetricsImpl) context.metrics();
+    }

Review comment:
       Added a bunch of duplicated extractors here to help keep the implementation classes clean.

##########
File path: streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
##########
@@ -122,7 +122,7 @@
     }
 
     @SuppressWarnings("unchecked")
-    void openDB(final ProcessorContext context) {
+    void openDB(final Map<String, Object> configs, final File stateDir) {

Review comment:
       Here's the interface change that saved us from needing two openDB methods.

##########
File path: streams/src/test/java/org/apache/kafka/streams/state/internals/CachingKeyValueStoreTest.java
##########
@@ -100,6 +100,31 @@ public void after() {
         return store;
     }
 
+    @SuppressWarnings("deprecation")
+    @Test
+    public void shouldDelegateDeprecatedInit() {

Review comment:
       These are the new unit tests I added to make sure that all the store builder wrappers transmit init calls correctly. They are frustratingly similar, but not exactly the same across different test classes because the test classes follow different idioms.
   
   I think it'd be nice to follow up with a general store-verification test that's parameterized by the exact store types so we can specify this test logic once and apply it to all the stores. That would also be handy for most of the rest of these tests. But I don't think we need to distract this PR with that concern.

##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImplTest.java
##########
@@ -147,7 +147,7 @@ public void shouldNotAllowToSchedulePunctuations() {
     public void shouldNotAllowInitForKeyValueStore() {
         final StateStore store = globalContext.getStateStore(GLOBAL_KEY_VALUE_STORE_NAME);
         try {
-            store.init((ProcessorContext) null, null);
+            store.init((StateStoreContext) null, null);

Review comment:
       This was actually a bug before, which this PR fixes: the wrapping layers should transmit the init call straight down, rather than translating it. There are a whole set of new unit tests making sure that this works properly for both the new and old init methods.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org