mjsax commented on code in PR #16906:
URL: https://github.com/apache/kafka/pull/16906#discussion_r1730428926
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java:
##########
@@ -72,14 +71,6 @@ class CachingWindowStore
this.maxObservedTimestamp = new AtomicLong(RecordQueue.UNKNOWN);
}
- @Deprecated
- @Override
- public void init(final ProcessorContext context, final StateStore root) {
- final String changelogTopic =
ProcessorContextUtils.changelogFor(context, name(), Boolean.TRUE);
- initInternal(asInternalProcessorContext(context), changelogTopic);
- super.init(context, root);
- }
-
@Override
public void init(final StateStoreContext context, final StateStore root) {
final String changelogTopic =
ProcessorContextUtils.changelogFor(context, name(), Boolean.TRUE);
Review Comment:
Can we inline `initInternal` here, too?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java:
##########
@@ -95,16 +93,17 @@ public String name() {
return name;
}
- @Deprecated
@Override
- public void init(final ProcessorContext context, final StateStore root) {
+ public void init(final StateStoreContext stateStoreContext,
+ final StateStore root) {
+ this.stateStoreContext = stateStoreContext;
final String threadId = Thread.currentThread().getName();
- final String taskName = context.taskId().toString();
+ final String taskName = stateStoreContext.taskId().toString();
// The provided context is not required to implement
InternalProcessorContext,
// If it doesn't, we can't record this metric.
- if (context instanceof InternalProcessorContext) {
- this.context = (InternalProcessorContext) context;
+ if (stateStoreContext instanceof InternalProcessorContext) {
Review Comment:
Should we refactor this method and use `asInternalContext` similar to other
methods instead?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTimeOrderedKeyValueChangeBuffer.java:
##########
@@ -196,15 +195,6 @@ public void setSerdesIfNull(final SerdeGetter getter) {
keySerde = keySerde == null ? (Serde<K>) getter.keySerde() : keySerde;
valueSerde = valueSerde == null ? FullChangeSerde.wrap((Serde<V>)
getter.valueSerde()) : valueSerde;
}
-
- @Deprecated
- @Override
- public void init(final ProcessorContext context, final StateStore root) {
- this.context =
ProcessorContextUtils.asInternalProcessorContext(context);
- changelogTopic = ProcessorContextUtils.changelogFor(context, name(),
Boolean.TRUE);
- init(root);
Review Comment:
Seem this `init` is the same and `internalInit` in other classes -- can we
inline it?
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
