[jira] [Updated] (KAFKA-7536) TopologyTestDriver cannot pre-populate KTable
[ https://issues.apache.org/jira/browse/KAFKA-7536?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dmitry Minkovsky updated KAFKA-7536: Summary: TopologyTestDriver cannot pre-populate KTable (was: TopologyTestDriver cannot pre-populate GlobalKTable) > TopologyTestDriver cannot pre-populate KTable > - > > Key: KAFKA-7536 > URL: https://issues.apache.org/jira/browse/KAFKA-7536 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.0 >Reporter: Dmitry Minkovsky >Priority: Minor > > I have a GlobalKTable that's defined as > {code} > GlobalKTable userIdsByEmail = topology >.globalTable(USER_IDS_BY_EMAIL.name, >USER_IDS_BY_EMAIL.consumed(), >Materialized.as("user-ids-by-email")); > {code} > And the following test in Spock: > {code} > def topology = // my topology > def driver = new TopologyTestDriver(topology, config()) > def cleanup() { > driver.close() > } > def "create from email request"() { > def store = driver.getKeyValueStore('user-ids-by-email') > store.put('string', ByteString.copyFrom(new byte[0])) > // more, but it fails at the `put` above > {code} > When I run this, I get the following: > {code} > [2018-10-23 19:35:27,055] INFO > (org.apache.kafka.streams.processor.internals.GlobalStateManagerImpl) mock > Restoring state for global store user-ids-by-email > java.lang.NullPointerException > at > org.apache.kafka.streams.processor.internals.AbstractProcessorContext.topic(AbstractProcessorContext.java:115) > at > org.apache.kafka.streams.state.internals.CachingKeyValueStore.putInternal(CachingKeyValueStore.java:237) > at > org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:220) > at > org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:38) > at > org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.put(InnerMeteredKeyValueStore.java:206) > at > org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.put(MeteredKeyValueBytesStore.java:117) > at pony.message.MessageWriteStreamsTest.create from mailgun email > request(MessageWriteStreamsTest.groovy:52) > [2018-10-23 19:35:27,189] INFO > (org.apache.kafka.streams.processor.internals.StateDirectory) stream-thread > [main] Deleting state directory 0_0 for task 0_0 as user calling cleanup. > {code} > I've noticed that I can {{put()}} to the store if I first write to it with > {{driver.pipeInput}}. But otherwise I get the above error. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (KAFKA-7536) TopologyTestDriver cannot pre-populate KTable
[ https://issues.apache.org/jira/browse/KAFKA-7536?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dmitry Minkovsky updated KAFKA-7536: Description: I have a GlobalKTable that's defined as {code} GlobalKTable userIdsByEmail = topology .globalTable(USER_IDS_BY_EMAIL.name, USER_IDS_BY_EMAIL.consumed(), Materialized.as("user-ids-by-email")); {code} And the following test in Spock: {code} def topology = // my topology def driver = new TopologyTestDriver(topology, config()) def cleanup() { driver.close() } def "create from email request"() { def store = driver.getKeyValueStore('user-ids-by-email') store.put('string', ByteString.copyFrom(new byte[0])) // more, but it fails at the `put` above {code} When I run this, I get the following: {code} [2018-10-23 19:35:27,055] INFO (org.apache.kafka.streams.processor.internals.GlobalStateManagerImpl) mock Restoring state for global store user-ids-by-email java.lang.NullPointerException at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.topic(AbstractProcessorContext.java:115) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.putInternal(CachingKeyValueStore.java:237) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:220) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:38) at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.put(InnerMeteredKeyValueStore.java:206) at org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.put(MeteredKeyValueBytesStore.java:117) at pony.message.MessageWriteStreamsTest.create from mailgun email request(MessageWriteStreamsTest.groovy:52) [2018-10-23 19:35:27,189] INFO (org.apache.kafka.streams.processor.internals.StateDirectory) stream-thread [main] Deleting state directory 0_0 for task 0_0 as user calling cleanup. {code} The same issue applies to KTable. I've noticed that I can {{put()}} to the store if I first write to it with {{driver.pipeInput}}. But otherwise I get the above error. was: I have a GlobalKTable that's defined as {code} GlobalKTable userIdsByEmail = topology .globalTable(USER_IDS_BY_EMAIL.name, USER_IDS_BY_EMAIL.consumed(), Materialized.as("user-ids-by-email")); {code} And the following test in Spock: {code} def topology = // my topology def driver = new TopologyTestDriver(topology, config()) def cleanup() { driver.close() } def "create from email request"() { def store = driver.getKeyValueStore('user-ids-by-email') store.put('string', ByteString.copyFrom(new byte[0])) // more, but it fails at the `put` above {code} When I run this, I get the following: {code} [2018-10-23 19:35:27,055] INFO (org.apache.kafka.streams.processor.internals.GlobalStateManagerImpl) mock Restoring state for global store user-ids-by-email java.lang.NullPointerException at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.topic(AbstractProcessorContext.java:115) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.putInternal(CachingKeyValueStore.java:237) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:220) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:38) at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.put(InnerMeteredKeyValueStore.java:206) at org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.put(MeteredKeyValueBytesStore.java:117) at pony.message.MessageWriteStreamsTest.create from mailgun email request(MessageWriteStreamsTest.groovy:52) [2018-10-23 19:35:27,189] INFO (org.apache.kafka.streams.processor.internals.StateDirectory) stream-thread [main] Deleting state directory 0_0 for task 0_0 as user calling cleanup. {code} I've noticed that I can {{put()}} to the store if I first write to it with {{driver.pipeInput}}. But otherwise I get the above error. > TopologyTestDriver cannot pre-populate KTable > - > > Key: KAFKA-7536 > URL: https://issues.apache.org/jira/browse/KAFKA-7536 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.0 >Reporter: Dmitry Minkovsky >Priority: Minor > > I have a GlobalKTable that's defined as > {code} > GlobalKTable userIdsByEmail = topology >.globalTable(USER_IDS_BY_EMAIL.name, >USER_IDS_BY_EMAIL.consumed(), >Materialized.as("u
[jira] [Updated] (KAFKA-7536) TopologyTestDriver cannot pre-populate KTable or GlobalKTable
[ https://issues.apache.org/jira/browse/KAFKA-7536?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Matthias J. Sax updated KAFKA-7536: --- Summary: TopologyTestDriver cannot pre-populate KTable or GlobalKTable (was: TopologyTestDriver cannot pre-populate KTable) > TopologyTestDriver cannot pre-populate KTable or GlobalKTable > - > > Key: KAFKA-7536 > URL: https://issues.apache.org/jira/browse/KAFKA-7536 > Project: Kafka > Issue Type: Bug > Components: streams >Affects Versions: 2.0.0 >Reporter: Dmitry Minkovsky >Priority: Minor > > I have a GlobalKTable that's defined as > {code} > GlobalKTable userIdsByEmail = topology >.globalTable(USER_IDS_BY_EMAIL.name, >USER_IDS_BY_EMAIL.consumed(), >Materialized.as("user-ids-by-email")); > {code} > And the following test in Spock: > {code} > def topology = // my topology > def driver = new TopologyTestDriver(topology, config()) > def cleanup() { > driver.close() > } > def "create from email request"() { > def store = driver.getKeyValueStore('user-ids-by-email') > store.put('string', ByteString.copyFrom(new byte[0])) > // more, but it fails at the `put` above > {code} > When I run this, I get the following: > {code} > [2018-10-23 19:35:27,055] INFO > (org.apache.kafka.streams.processor.internals.GlobalStateManagerImpl) mock > Restoring state for global store user-ids-by-email > java.lang.NullPointerException > at > org.apache.kafka.streams.processor.internals.AbstractProcessorContext.topic(AbstractProcessorContext.java:115) > at > org.apache.kafka.streams.state.internals.CachingKeyValueStore.putInternal(CachingKeyValueStore.java:237) > at > org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:220) > at > org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:38) > at > org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore.put(InnerMeteredKeyValueStore.java:206) > at > org.apache.kafka.streams.state.internals.MeteredKeyValueBytesStore.put(MeteredKeyValueBytesStore.java:117) > at pony.message.MessageWriteStreamsTest.create from mailgun email > request(MessageWriteStreamsTest.groovy:52) > [2018-10-23 19:35:27,189] INFO > (org.apache.kafka.streams.processor.internals.StateDirectory) stream-thread > [main] Deleting state directory 0_0 for task 0_0 as user calling cleanup. > {code} > The same issue applies to KTable. > I've noticed that I can {{put()}} to the store if I first write to it with > {{driver.pipeInput}}. But otherwise I get the above error. -- This message was sent by Atlassian JIRA (v7.6.3#76005)