[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17211320#comment-17211320
]
Bill Bejeck commented on KAFKA-8630:
[~vvcephei] ok for me to close this duplicate since there is
https://issues.apache.org/jira/browse/KAFKA-10200 with PR submitted with it?
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Assignee: John Roesler
>Priority: Critical
> Fix For: 2.7.0, 2.6.1
>
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
> com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
> at
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17146747#comment-17146747
]
John Roesler commented on KAFKA-8630:
-
Hey, all, I inadvertently started a new ticket for this KAFKA-10200, and
submitted a patch already. Sorry if I stepped on any toes.
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Assignee: John Roesler
>Priority: Critical
> Fix For: 2.7.0, 2.6.1
>
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
> com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17104758#comment-17104758
]
Matthias J. Sax commented on KAFKA-8630:
[~vamcrulz09] not so far – but we take PRs :)
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Critical
> Fix For: 2.6.0
>
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
> com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17104748#comment-17104748
]
Vamsi Krishna Koganti commented on KAFKA-8630:
--
Is there any update on this issue?
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Critical
> Fix For: 2.6.0
>
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
> com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16962448#comment-16962448
]
Bruno Cadonna commented on KAFKA-8630:
--
Agreed
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
> com.cantgetthistowork.InMemWindowProcessorTest.setup(InMemWindowProcessorTest.java:36)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16962413#comment-16962413
]
Sophie Blee-Goldman commented on KAFKA-8630:
That's fair, I agree that would be the ideal resolution to this ticket and
something we should do anyways. That said I've been thinking lately this ticket
is actually pretty important as without it users basically have no legitimate
way of unit testing a stateful transformer, processor, etc.
But if we aren't able to fix this issue holistically (ie resolve KAFKA-9109)
by the next release cycle, I would vote to implement the quick and dirty
solution I outlined in KAFKA-9088 just to give users something.
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16961820#comment-16961820
]
Bruno Cadonna commented on KAFKA-8630:
--
I also run into the {{ClassCastException}} during development. I had to work
around it. While I agree that the fastest way to solve the specific issue of
this ticket is on the test side, I also feel that it is somehow awkward to get
a {{ClassCastException}} when your code conforms to the specified APIs. The
cast to {{InternalProcessorContext}} is problematic and we should get rid of
it. Thus, I filed KAFKA-9109.
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16881603#comment-16881603
]
Sophie Blee-Goldman commented on KAFKA-8630:
Yep, that's the KIP. There was some ongoing discussion but I think it got
dropped on a floor a little while back as people became busy. You might be able
to kick start discussion again with this as motivation – someone on the mailing
list had a similar problem, but wanted to use key-value stores so for the time
being they can get by with just MockProcessorContext. Of course, even though
some stores don't currently need the InternalProcessorContext yet, we are
actively adding more metrics and this may not be true for long. It's worth
solving this holistically.
That's probably a lot more work than we need to do here. You're right about the
StreamsMetrics vs StreamsMetricsImpl – the window store, for example, needs to
access the storeLevelSensor method, but we don't want to expose that. I agree
it's pretty annoying that all this boils down to is needing to access some
internal metrics/sensors, but I do think the right way to solve it is on the
test side. From the source code side, everything is working as it should – it's
the MockProcessorContext that fails to adequately 'mock' the real thing. I
think some light refactoring to have the contexts implement some
"ProcessorContextMetrics" interface could be accomplished pretty easily
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16880923#comment-16880923
]
Justin Fetherolf commented on KAFKA-8630:
-
The `ProcessorContext` interface already offers a `metrics()` method, and
`InternalProcessorContext` for some reason overrides it to return a
`StreamsMetricsImpl` rather than `StreamsMetrics`.
The bit of poking around in the code I've done this evening it seems to be kind
of a mixed bag of stores needing just this `metrics()` method (and thus not
needing to do the cast) and needing the other methods that
`InternalProcessorContext` offers.
Is the possibly pre-existing KIP you're thinking of
[KIP-448|https://cwiki.apache.org/confluence/x/SAeZBg]?
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16880891#comment-16880891
]
Matthias J. Sax commented on KAFKA-8630:
I agree with Justin. Testing using internal classes should be avoided. That is
the reason why we added the test-utils package to begin with.
About the testing for the type of the context: this would introduce a
dependency from the "main" module to the test-utils model – beside that the
main code should not know anything about the test code, it would also be a
cyclic dependency.
We need to address this differently (maybe add a new interface) that both the
internal and mocked context class can implement and that we use for the cast.
This might require a KIP though.
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16880886#comment-16880886
]
Sophie Blee-Goldman commented on KAFKA-8630:
You're right, I overlooked that InternalMockProcessorContext is not part of the
public API and definitely shouldn't be used. In general users are encouraged to
do their testing through the TopologyTestDriver. You can access state stores
through it and they will be managed for you so you don't have to worry about
calling init or close. (This is especially important with the persistent
variants of the stores as they will leak memory if not properly closed.)
That said, I can see how this would be useful even if not necessary. If you are
interested in picking this up, I'd be happy to help! Just to outline the
problem clearly, along with some possible solutions:
This affects all window and session stores, and may affect key-value stores in
the future should we chose to add metrics.The problem is that the stores cast
to InternalMockProcessorContext in order to access the StreamsMetrics, but the
MockProcessorContext intended for testing does not extend this class. I don't
think it's ideal to just add "if" guards everywhere this internal context or
the related metrics/sensors are used in the actual streams code just to make
room for some test code.
One possibility would be to have the Internal and Mock processor contexts'
implement a common interface that has a `metrics()` method, and just return a
dummy metrics in the Mock case. Alternatively we could add some kind of Mock
stores that can hide the internals and just present a stable interface for
testing – either way we would need a KIP. Although I believe there already was
a KIP regarding mock stores, maybe this could be a part of that
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16880831#comment-16880831
]
Justin Fetherolf commented on KAFKA-8630:
-
[~ableegoldman] I personally feel that this is still a problem and shouldn't be
closed.
Trying to find `InternalMockProcessorContext` revealed another class
`MockInternalProcessorContext`, both of which are in test code and not
available via the public packages (or at least not in `kafka-streams` or
`kafka-streams-test-utils` which seems like the logical place for them).
I think the root problem here is not having necessary mock contexts in the
proper package, and not having the documentation to support their use cases.
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
>
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16880771#comment-16880771
]
Sophie Blee-Goldman commented on KAFKA-8630:
Yes, you're likely to hit this problem with some of the other store
implementations as well which also cast to `InternalProcessorContext`. This is
needed in order to access the metrics. However you should be able to run your
tests by just using an `InternalMockProcessorContext` instead.
[~fetherolfjd] Can I close this as "not a problem" ?
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
[jira] [Commented] (KAFKA-8630) Unit testing a streams processor with a WindowStore throws a ClassCastException
[
https://issues.apache.org/jira/browse/KAFKA-8630?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16880227#comment-16880227
]
Denis Washington commented on KAFKA-8630:
-
Looks like the {{InMemoryWindowStore}} needs the extra methods of the
{{InternalProcessorContext}} interface only for recording metrics. Perhaps it
could just surround those blocks with {{if (context instanceof
InternalProcessorContext)}}, but otherwise also accept other
{{ProcessorContext}} implementations like {{MockProcessorContext}}.
> Unit testing a streams processor with a WindowStore throws a
> ClassCastException
> ---
>
> Key: KAFKA-8630
> URL: https://issues.apache.org/jira/browse/KAFKA-8630
> Project: Kafka
> Issue Type: Bug
> Components: streams-test-utils
>Affects Versions: 2.3.0
>Reporter: Justin Fetherolf
>Priority: Major
>
> I was attempting to write a unit test for a class implementing the
> {{Processor}} interface that contained a {{WindowStore}}, but running the
> test fails with a {{ClassCastException}} coming out of
> {{InMemoryWindowStore.init}} attempting to cast {{MockProcessorContext}} to
> {{InternalProcessorContext}}.
> Minimal code to reproduce:
> {code:java}
> package com.cantgetthistowork;
> import org.apache.kafka.streams.processor.Processor;
> import org.apache.kafka.streams.processor.ProcessorContext;
> import org.apache.kafka.streams.state.WindowStore;
> public class InMemWindowProcessor implements Processor {
> private ProcessorContext context;
> private WindowStore windowStore;
> @Override
> public void init(ProcessorContext context) {
> this.context = context;
> windowStore = (WindowStore)
> context.getStateStore("my-win-store");
> }
> @Override
> public void process(String key, String value) {
> }
> @Override
> public void close() {
> }
> }
> {code}
> {code:java}
> package com.cantgetthistowork;
> import java.time.Duration;
> import java.time.Instant;
> import org.apache.kafka.common.serialization.Serdes;
> import org.apache.kafka.streams.processor.MockProcessorContext;
> import org.apache.kafka.streams.state.Stores;
> import org.apache.kafka.streams.state.WindowStore;
> import org.junit.Before;
> import org.junit.Test;
> public class InMemWindowProcessorTest {
> InMemWindowProcessor processor = null;
> MockProcessorContext context = null;
> @Before
> public void setup() {
> processor = new InMemWindowProcessor();
> context = new MockProcessorContext();
> WindowStore store =
> Stores.windowStoreBuilder(
> Stores.inMemoryWindowStore(
> "my-win-store",
> Duration.ofMinutes(10),
> Duration.ofSeconds(10),
> false
> ),
> Serdes.String(),
> Serdes.String()
> )
> .withLoggingDisabled()
> .build();
> store.init(context, store);
> context.register(store, null);
> processor.init(context);
> }
> @Test
> public void testThings() {
> Instant baseTime = Instant.now();
> context.setTimestamp(baseTime.toEpochMilli());
> context.setTopic("topic-name");
> processor.process("key1", "value1");
> }
> }
> {code}
>
> I was trying this with maven, with mvn --version outputting:
> {noformat}
> Apache Maven 3.5.0 (ff8f5e7444045639af65f6095c62210b5713f426;
> 2017-04-03T13:39:06-06:00)
> Maven home: ~/opt/apache-maven-3.5.0
> Java version: 1.8.0_212, vendor: Oracle Corporation
> Java home: /usr/lib/jvm/java-8-openjdk-amd64/jre
> Default locale: en_US, platform encoding: UTF-8
> OS name: "linux", version: "4.15.0-52-generic", arch: "amd64", family:
> "unix"{noformat}
> And finally the stack trace:
> {noformat}
> ---
> T E S T S
> ---
> Running com.cantgetthistowork.InMemWindowProcessorTest
> SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
> SLF4J: Defaulting to no-operation (NOP) logger implementation
> SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further
> details.
> Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.076 sec <<<
> FAILURE!
> testThings(com.cantgetthistowork.InMemWindowProcessorTest) Time elapsed:
> 0.05 sec <<< ERROR!
> java.lang.ClassCastException:
> org.apache.kafka.streams.processor.MockProcessorContext cannot be cast to
> org.apache.kafka.streams.processor.internals.InternalProcessorContext
> at
> org.apache.kafka.streams.state.internals.InMemoryWindowStore.init(InMemoryWindowStore.java:91)
> at
> org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48)
> at
> org.apache.kafka.streams.state.internals.MeteredWindowStore.init(MeteredWindowStore.java:90)
> at
>
