subject:"Re\: Kafka Streams not auto\-creating the state store changelog topic"

Re: Kafka Streams not auto-creating the state store changelog topic

2017-08-07 Thread Anish Mashankar

Hello Eno,
Yes, I have followed a similar code to setup the streams application. Does any 
other code inside the application affect the bootstrapping steps? 
I have a custom interface in which I populate Streams properties using 
environment variables. I attach the state store to the builder using:

builder.addProcessor(...)
builder.addStateStore(stateStoreSupplier, PROCESSOR_NAME)
…
builder.build()

The state store is created in this way:
Stores.create(name)
.withKeys(keySerde)
.withValues(valueSerde)
.persistent()
.build();

I am using the following streams configurations:
props.put(StreamsConfig.APPLICATION_ID_CONFIG, appId);
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, 
timeStampExtractor);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, keySerdes);
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, valueSerdes);
props.put(StreamsConfig.PARTITION_GROUPER_CLASS_CONFIG, partitionGrouper);
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, numStandbyReplicas);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numStreamThreads);
props.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, replicationFactor);
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, applicationServer);
props.put(StreamsConfig.STATE_DIR_CONFIG, stateStoreDir);

Are there any new properties I am missing? I am not seeing any errors on the 
broker. Do I need to enable auto topic creation on the broker for this to work?

> On 07-Aug-2017, at 1:05 PM, Eno Thereska  wrote:
> 
> HI Anish,
> 
> Yeah, changing the input topic partitions at runtime could be problematic. 
> But it doesn’t seem like that’s what’s going on here. (For regex the 
> application it will work fine).
> 
> Are there any broker failures going on while test is running? Also, I wonder 
> about how the rest of your code looks like. There is some code here 
> https://github.com/confluentinc/examples/blob/3.3.0-post/kafka-streams/src/test/java/io/confluent/examples/streams/StateStoresInTheDSLIntegrationTest.java#L158
>  
> 
>  that shows how to create the state stores and initialize Kafka Streams and 
> the order of doing things. Could you please double check if it matches your 
> code?
> 
> Thanks
> Eno
> 
> 
>> On Aug 5, 2017, at 3:22 AM, Anish Mashankar  wrote:
>> 
>> Hello Eno,
>> So, if I change the input topic partitions, it affects the ability of kafka
>> streams to find partitions for the state store changelog? I think I'm
>> missing something here.
>> In my case, the application was new, so it's for sure that there were no
>> changes.
>> Also, if I have regex for the input topic on kafka streams and a new topic
>> is added to kafka matching the regex, the application will break?
>> 
>> On Fri, Aug 4, 2017, 8:33 PM Eno Thereska > > wrote:
>> 
>>> Hi,
>>> 
>>> Could you check if this helps:
>>> 
>>> https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
>>>  
>>> 
>>> <
>>> https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
>>>  
>>> 
 
>>> 
>>> Thanks
>>> Eno
 On Aug 4, 2017, at 12:48 PM, Anish Mashankar 
>>> wrote:
 
 Hello Eno,
 Thanks for considering the question.
 
 How I am creating the state stores:
 
 StateStoreSupplier stateStoreSupplier =
 
>>> StateStorStores.create("testing-2-store").withKeys(keySerde).withValues(valueSerde).persistent().build();
 TopologyBuilder builder = ...
 builder.addStateStore(stateStoreSupplier, "ProcessorUsingStateStore");
 
 The Error Message with stack trace is as follows:
 
 2017-08-04 17:11:23,184 53205
 [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
 o.a.k.s.p.internals.StreamThread - stream-thread
 [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] Created
 active task -727063541_0 with assigned partitions [testing-topic-0]
 
 2017-08-04 17:11:23,185 53206
 [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
 o.a.k.s.p.internals.StreamThread - stream-thread
 [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] partition
 assignment took 41778 ms.
 current active tasks: []
 current standby tasks: []
 
 2017-08-04 17:11:23,187 53208
 [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] ERROR
 o.a.k.c.c.i.ConsumerCoordinator - User provid

Re: Kafka Streams not auto-creating the state store changelog topic

2017-08-07 Thread Eno Thereska

HI Anish,

Yeah, changing the input topic partitions at runtime could be problematic. But 
it doesn’t seem like that’s what’s going on here. (For regex the application it 
will work fine).

Are there any broker failures going on while test is running? Also, I wonder 
about how the rest of your code looks like. There is some code here 
https://github.com/confluentinc/examples/blob/3.3.0-post/kafka-streams/src/test/java/io/confluent/examples/streams/StateStoresInTheDSLIntegrationTest.java#L158
 

 that shows how to create the state stores and initialize Kafka Streams and the 
order of doing things. Could you please double check if it matches your code?

Thanks
Eno


> On Aug 5, 2017, at 3:22 AM, Anish Mashankar  wrote:
> 
> Hello Eno,
> So, if I change the input topic partitions, it affects the ability of kafka
> streams to find partitions for the state store changelog? I think I'm
> missing something here.
> In my case, the application was new, so it's for sure that there were no
> changes.
> Also, if I have regex for the input topic on kafka streams and a new topic
> is added to kafka matching the regex, the application will break?
> 
> On Fri, Aug 4, 2017, 8:33 PM Eno Thereska  > wrote:
> 
>> Hi,
>> 
>> Could you check if this helps:
>> 
>> https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
>>  
>> 
>> <
>> https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
>>  
>> 
>>> 
>> 
>> Thanks
>> Eno
>>> On Aug 4, 2017, at 12:48 PM, Anish Mashankar 
>> wrote:
>>> 
>>> Hello Eno,
>>> Thanks for considering the question.
>>> 
>>> How I am creating the state stores:
>>> 
>>> StateStoreSupplier stateStoreSupplier =
>>> 
>> StateStorStores.create("testing-2-store").withKeys(keySerde).withValues(valueSerde).persistent().build();
>>> TopologyBuilder builder = ...
>>> builder.addStateStore(stateStoreSupplier, "ProcessorUsingStateStore");
>>> 
>>> The Error Message with stack trace is as follows:
>>> 
>>> 2017-08-04 17:11:23,184 53205
>>> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
>>> o.a.k.s.p.internals.StreamThread - stream-thread
>>> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] Created
>>> active task -727063541_0 with assigned partitions [testing-topic-0]
>>> 
>>> 2017-08-04 17:11:23,185 53206
>>> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
>>> o.a.k.s.p.internals.StreamThread - stream-thread
>>> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] partition
>>> assignment took 41778 ms.
>>> current active tasks: []
>>> current standby tasks: []
>>> 
>>> 2017-08-04 17:11:23,187 53208
>>> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] ERROR
>>> o.a.k.c.c.i.ConsumerCoordinator - User provided listener
>>> 
>> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener
>>> for group testing-2 failed on partition assignment
>>> org.apache.kafka.streams.errors.StreamsException: Store testing-2-store's
>>> change log (testing-2-testing-2-store-changelog) does not contain
>> partition
>>> 0
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.StoreChangelogReader.validatePartitionExists(StoreChangelogReader.java:87)
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:165)
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:100)
>>> at
>>> 
>> org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:177)
>>> at
>>> 
>> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:40)
>>> at
>>> 
>> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueStore.init(ChangeLoggingKeyValueStore.java:57)
>>> at
>>> 
>> org.apache.kafka.streams.state.internals.MeteredKeyValueStore$7.run(MeteredKeyValueStore.java:99)
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)
>>> at
>>> 
>> org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:130)
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:201)
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:140)
>>> at
>>> 
>> org.apache.kafka.streams.processor.internals.StreamThread.creat

Re: Kafka Streams not auto-creating the state store changelog topic

2017-08-04 Thread Anish Mashankar

Hello Eno,
So, if I change the input topic partitions, it affects the ability of kafka
streams to find partitions for the state store changelog? I think I'm
missing something here.
In my case, the application was new, so it's for sure that there were no
changes.
Also, if I have regex for the input topic on kafka streams and a new topic
is added to kafka matching the regex, the application will break?

On Fri, Aug 4, 2017, 8:33 PM Eno Thereska  wrote:

> Hi,
>
> Could you check if this helps:
>
> https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
> <
> https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
> >
>
> Thanks
> Eno
> > On Aug 4, 2017, at 12:48 PM, Anish Mashankar 
> wrote:
> >
> > Hello Eno,
> > Thanks for considering the question.
> >
> > How I am creating the state stores:
> >
> > StateStoreSupplier stateStoreSupplier =
> >
> StateStorStores.create("testing-2-store").withKeys(keySerde).withValues(valueSerde).persistent().build();
> > TopologyBuilder builder = ...
> > builder.addStateStore(stateStoreSupplier, "ProcessorUsingStateStore");
> >
> > The Error Message with stack trace is as follows:
> >
> > 2017-08-04 17:11:23,184 53205
> > [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
> > o.a.k.s.p.internals.StreamThread - stream-thread
> > [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] Created
> > active task -727063541_0 with assigned partitions [testing-topic-0]
> >
> > 2017-08-04 17:11:23,185 53206
> > [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
> > o.a.k.s.p.internals.StreamThread - stream-thread
> > [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] partition
> > assignment took 41778 ms.
> > current active tasks: []
> > current standby tasks: []
> >
> > 2017-08-04 17:11:23,187 53208
> > [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] ERROR
> > o.a.k.c.c.i.ConsumerCoordinator - User provided listener
> >
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener
> > for group testing-2 failed on partition assignment
> > org.apache.kafka.streams.errors.StreamsException: Store testing-2-store's
> > change log (testing-2-testing-2-store-changelog) does not contain
> partition
> > 0
> > at
> >
> org.apache.kafka.streams.processor.internals.StoreChangelogReader.validatePartitionExists(StoreChangelogReader.java:87)
> > at
> >
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:165)
> > at
> >
> org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:100)
> > at
> >
> org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:177)
> > at
> >
> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:40)
> > at
> >
> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueStore.init(ChangeLoggingKeyValueStore.java:57)
> > at
> >
> org.apache.kafka.streams.state.internals.MeteredKeyValueStore$7.run(MeteredKeyValueStore.java:99)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)
> > at
> >
> org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:130)
> > at
> >
> org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:201)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:140)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:1234)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:294)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:254)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:1313)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread.access$1100(StreamThread.java:73)
> > at
> >
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:183)
> > at
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:265)
> > at
> >
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:363)
> > at
> >
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:310)
> > at
> >
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:297)
> > at
> >
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
> > at
> >
> org.apache.kafka.clients.consumer.

Re: Kafka Streams not auto-creating the state store changelog topic

2017-08-04 Thread Eno Thereska

Hi,

Could you check if this helps:
https://stackoverflow.com/questions/42329387/failed-to-rebalance-error-in-kafka-streams-with-more-than-one-topic-partition
 


Thanks
Eno
> On Aug 4, 2017, at 12:48 PM, Anish Mashankar  wrote:
> 
> Hello Eno,
> Thanks for considering the question.
> 
> How I am creating the state stores:
> 
> StateStoreSupplier stateStoreSupplier =
> StateStorStores.create("testing-2-store").withKeys(keySerde).withValues(valueSerde).persistent().build();
> TopologyBuilder builder = ...
> builder.addStateStore(stateStoreSupplier, "ProcessorUsingStateStore");
> 
> The Error Message with stack trace is as follows:
> 
> 2017-08-04 17:11:23,184 53205
> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
> o.a.k.s.p.internals.StreamThread - stream-thread
> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] Created
> active task -727063541_0 with assigned partitions [testing-topic-0]
> 
> 2017-08-04 17:11:23,185 53206
> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
> o.a.k.s.p.internals.StreamThread - stream-thread
> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] partition
> assignment took 41778 ms.
> current active tasks: []
> current standby tasks: []
> 
> 2017-08-04 17:11:23,187 53208
> [testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] ERROR
> o.a.k.c.c.i.ConsumerCoordinator - User provided listener
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener
> for group testing-2 failed on partition assignment
> org.apache.kafka.streams.errors.StreamsException: Store testing-2-store's
> change log (testing-2-testing-2-store-changelog) does not contain partition
> 0
> at
> org.apache.kafka.streams.processor.internals.StoreChangelogReader.validatePartitionExists(StoreChangelogReader.java:87)
> at
> org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:165)
> at
> org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:100)
> at
> org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:177)
> at
> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:40)
> at
> org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueStore.init(ChangeLoggingKeyValueStore.java:57)
> at
> org.apache.kafka.streams.state.internals.MeteredKeyValueStore$7.run(MeteredKeyValueStore.java:99)
> at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)
> at
> org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:130)
> at
> org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:201)
> at
> org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:140)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:1234)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:294)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:254)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:1313)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.access$1100(StreamThread.java:73)
> at
> org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:183)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:265)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:363)
> at
> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:310)
> at
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:297)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:582)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:553)
> at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527)
> 
> I hope this shares more light on the situation.
> Thanks
> 
> On Fri, Aug 4, 2017 at 2:28 PM Eno Thereska  > wrote:
> 
>> Hi Anish,
>> 
>> Could you give more info on how you create the state stores in your code?
>> Also could you copy-paste the exact error message from the log?
>> 
>> Thanks
>> Eno
>>> On Aug 4, 201

Re: Kafka Streams not auto-creating the state store changelog topic

2017-08-04 Thread Anish Mashankar

Hello Eno,
Thanks for considering the question.

How I am creating the state stores:

StateStoreSupplier stateStoreSupplier =
StateStorStores.create("testing-2-store").withKeys(keySerde).withValues(valueSerde).persistent().build();
TopologyBuilder builder = ...
builder.addStateStore(stateStoreSupplier, "ProcessorUsingStateStore");

The Error Message with stack trace is as follows:

2017-08-04 17:11:23,184 53205
[testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
 o.a.k.s.p.internals.StreamThread - stream-thread
[testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] Created
active task -727063541_0 with assigned partitions [testing-topic-0]

2017-08-04 17:11:23,185 53206
[testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] INFO
 o.a.k.s.p.internals.StreamThread - stream-thread
[testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] partition
assignment took 41778 ms.
current active tasks: []
current standby tasks: []

2017-08-04 17:11:23,187 53208
[testing-2-9f5aa1d8-35c7-4f0c-9593-be31738cb4c0-StreamThread-1] ERROR
o.a.k.c.c.i.ConsumerCoordinator - User provided listener
org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener
for group testing-2 failed on partition assignment
org.apache.kafka.streams.errors.StreamsException: Store testing-2-store's
change log (testing-2-testing-2-store-changelog) does not contain partition
0
at
org.apache.kafka.streams.processor.internals.StoreChangelogReader.validatePartitionExists(StoreChangelogReader.java:87)
at
org.apache.kafka.streams.processor.internals.ProcessorStateManager.register(ProcessorStateManager.java:165)
at
org.apache.kafka.streams.processor.internals.AbstractProcessorContext.register(AbstractProcessorContext.java:100)
at
org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:177)
at
org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:40)
at
org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueStore.init(ChangeLoggingKeyValueStore.java:57)
at
org.apache.kafka.streams.state.internals.MeteredKeyValueStore$7.run(MeteredKeyValueStore.java:99)
at
org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:187)
at
org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:130)
at
org.apache.kafka.streams.processor.internals.AbstractTask.initializeStateStores(AbstractTask.java:201)
at
org.apache.kafka.streams.processor.internals.StreamTask.(StreamTask.java:140)
at
org.apache.kafka.streams.processor.internals.StreamThread.createStreamTask(StreamThread.java:1234)
at
org.apache.kafka.streams.processor.internals.StreamThread$TaskCreator.createTask(StreamThread.java:294)
at
org.apache.kafka.streams.processor.internals.StreamThread$AbstractTaskCreator.retryWithBackoff(StreamThread.java:254)
at
org.apache.kafka.streams.processor.internals.StreamThread.addStreamTasks(StreamThread.java:1313)
at
org.apache.kafka.streams.processor.internals.StreamThread.access$1100(StreamThread.java:73)
at
org.apache.kafka.streams.processor.internals.StreamThread$RebalanceListener.onPartitionsAssigned(StreamThread.java:183)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:265)
at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:363)
at
org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:310)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:297)
at
org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1078)
at
org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1043)
at
org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:582)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:553)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:527)

I hope this shares more light on the situation.
Thanks

On Fri, Aug 4, 2017 at 2:28 PM Eno Thereska  wrote:

> Hi Anish,
>
> Could you give more info on how you create the state stores in your code?
> Also could you copy-paste the exact error message from the log?
>
> Thanks
> Eno
> > On Aug 4, 2017, at 9:05 AM, Anish Mashankar 
> wrote:
> >
> > I have a new application, call it streamsApp with state stores S1 and S2.
> > So, according to the documentation, upon the first time startup, the
> > application should've created the changelog topics
> streamsApp-S1-changelog
> > and streamsApp-S2-changelog. But I see that these topics are not created.
> > Also, the application throws an error that it couldn't find any partition
> > for topics *streamsApp-S1-changelog and streamsApp-S2-changelog *and then
> > exits*. *To get it working, I manually created the topics, but I

Re: Kafka Streams not auto-creating the state store changelog topic

2017-08-04 Thread Eno Thereska

Hi Anish,

Could you give more info on how you create the state stores in your code? Also 
could you copy-paste the exact error message from the log?

Thanks
Eno
> On Aug 4, 2017, at 9:05 AM, Anish Mashankar  wrote:
> 
> I have a new application, call it streamsApp with state stores S1 and S2.
> So, according to the documentation, upon the first time startup, the
> application should've created the changelog topics streamsApp-S1-changelog
> and streamsApp-S2-changelog. But I see that these topics are not created.
> Also, the application throws an error that it couldn't find any partition
> for topics *streamsApp-S1-changelog and streamsApp-S2-changelog *and then
> exits*. *To get it working, I manually created the topics, but I am
> skeptical because the docs say that this convention might change any time.
> I am using Kafka Streams v0.11, with a Kafka Broker v0.11, but message
> protocol set to v0.10.0. Am I missing something?
> -- 
> 
> Regards,
> Anish Samir Mashankar
> R&D Engineer
> System Insights
> +91-9789870733

Re: Kafka Streams not auto-creating the state store changelog topic

Re: Kafka Streams not auto-creating the state store changelog topic

Re: Kafka Streams not auto-creating the state store changelog topic

Re: Kafka Streams not auto-creating the state store changelog topic

Re: Kafka Streams not auto-creating the state store changelog topic

Re: Kafka Streams not auto-creating the state store changelog topic

6 matches

Site Navigation

Mail list logo

Footer information