vvcephei commented on a change in pull request #9467: URL: https://github.com/apache/kafka/pull/9467#discussion_r509373215
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java ########## @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.processor.internals; + +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; + +/** + * This class bridges the gap for components that _should_ be compatible with + * the public ProcessorContext interface, but have come to depend on features + * in InternalProcessorContext. In theory, all the features adapted here could + * migrate to the public interface, so each method in this class should reference + * the ticket that would ultimately obviate it. + */ +public final class ProcessorContextUtils { + + private ProcessorContextUtils() {} + + /** + * Note that KIP-622 would move currentSystemTimeMs to ProcessorContext, + * removing the need for this method. + */ + public static long getCurrentSystemTime(final ProcessorContext context) { + return context instanceof InternalProcessorContext + ? ((InternalProcessorContext) context).currentSystemTimeMs() + : System.currentTimeMillis(); + } + + /** + * Should be removed as part of KAFKA-10217 + */ + public static StreamsMetricsImpl getMetricsImpl(final ProcessorContext context) { + return (StreamsMetricsImpl) context.metrics(); + } + + 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: Note, there will also be an unused import after removing these methods. I don't know if you know, but you can run `./gradlew :streams:testAll` to fully run all the verifications locally. ########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextUtils.java ########## @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.processor.internals; + +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; + +/** + * This class bridges the gap for components that _should_ be compatible with + * the public ProcessorContext interface, but have come to depend on features + * in InternalProcessorContext. In theory, all the features adapted here could + * migrate to the public interface, so each method in this class should reference + * the ticket that would ultimately obviate it. + */ +public final class ProcessorContextUtils { + + private ProcessorContextUtils() {} + + /** + * Note that KIP-622 would move currentSystemTimeMs to ProcessorContext, + * removing the need for this method. + */ + public static long getCurrentSystemTime(final ProcessorContext context) { + return context instanceof InternalProcessorContext + ? ((InternalProcessorContext) context).currentSystemTimeMs() + : System.currentTimeMillis(); + } + + /** + * Should be removed as part of KAFKA-10217 + */ + public static StreamsMetricsImpl getMetricsImpl(final ProcessorContext context) { + return (StreamsMetricsImpl) context.metrics(); + } + + 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: ```suggestion ``` Let's not backport these methods, which are unused in 2.6. If we need them later, we can backport them at that time. ---------------------------------------------------------------- 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: us...@infra.apache.org
