-
Notifications
You must be signed in to change notification settings - Fork 14k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
KAFKA-10562: Properly invoke new StateStoreContext init #9388
Changes from 5 commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -65,7 +65,11 @@ public interface StateStore { | |
* | ||
* @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 | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Adding the deprecation tag right now lets us be sure we encountered all places this method appears in the codebase. |
||
void init(org.apache.kafka.streams.processor.ProcessorContext context, StateStore root); | ||
|
||
/** | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -20,6 +20,7 @@ | |
import org.apache.kafka.streams.kstream.Windowed; | ||
import org.apache.kafka.streams.processor.ProcessorContext; | ||
import org.apache.kafka.streams.processor.StateStore; | ||
import org.apache.kafka.streams.processor.StateStoreContext; | ||
import org.apache.kafka.streams.state.KeyValueIterator; | ||
import org.apache.kafka.streams.state.KeyValueStore; | ||
import org.apache.kafka.streams.state.SessionStore; | ||
|
@@ -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); | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 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. |
||
} | ||
|
||
@Override | ||
public void close() { | ||
throw new UnsupportedOperationException(ERROR_MESSAGE); | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -17,6 +17,7 @@ | |
package org.apache.kafka.streams.processor.internals; | ||
|
||
import org.apache.kafka.streams.processor.ProcessorContext; | ||
import org.apache.kafka.streams.processor.StateStoreContext; | ||
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; | ||
|
||
/** | ||
|
@@ -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(); | ||
} | ||
Comment on lines
+54
to
+56
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Added a bunch of duplicated extractors here to help keep the implementation classes clean. |
||
|
||
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." | ||
); | ||
} | ||
} | ||
Comment on lines
+70
to
+78
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 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. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Thanks! |
||
|
||
public static InternalProcessorContext asInternalProcessorContext(final StateStoreContext 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." | ||
); | ||
} | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -225,6 +225,7 @@ public String name() { | |
return name; | ||
} | ||
|
||
@Deprecated | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. There are a handful of these also, just passing the deprecation on to the callers. |
||
@Override | ||
public void init(final ProcessorContext context, | ||
final StateStore root) { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This ticket needs to go in to 2.7.0 also, but I split it out for reviewability.