Hi, Jordi, The local KV-store is meant to be accessed by the Samza container locally on the machine. If you were referring to the use case that the local KV-store is accessed by a container from a different Samza job, it is not supported. And what does it exactly mean when you say "load the DB to be able to use it from the consuming job"? We may be of more help if we know your use case in more details.
Thanks! -Yi On Tue, Aug 11, 2015 at 3:00 AM, Jordi Blasi Uribarri <jbl...@nextel.es> wrote: > InitableTask tas kwas missing. That responds to another problem that I was > experiencing (and left for later). Anyway the exception was still there > until I commented the changelog definition line in the properties file: > > #stores.test12db.changelog=kafka.test12db-changelog > > As I understand it in case of job going down information will be lost. > That is not a real issue as I am storing temporal information there. > > What I am seeing is something that does not work as I expected. Maybe I am > not understanding correctly how the system works. I need that a job has > access to the information previously stored in the storage. I was planning > a loader job that on receiving some messages with data it stores them in > RocksDb and that information should be consumed by a different consumer job > to use it for calculation. > > I see that in the loader job I can put and get information correctly. When > I try to access the same storage from a different job I just get null > results. > > How I am supposed to load the DB to be able to use it from the consuming > job? Is RocksDB the tool to use or should I use any other technique? > > Thanks, > > Jordi > > > > -----Mensaje original----- > De: Yi Pan [mailto:nickpa...@gmail.com] > Enviado el: martes, 11 de agosto de 2015 3:27 > Para: dev@samza.apache.org > Asunto: Re: Missing a change log offset for SystemStreamPartition > > Hi, Jordi, > > Agree with Yan. More specifically, your class definition should be > something like: > {code} > public class testStore implements StreamTask, InitableTask { ... > } > {code} > > On Mon, Aug 10, 2015 at 6:08 PM, Yan Fang <yanfang...@gmail.com> wrote: > > > Hi Jordi, > > > > I think, you need to implement the *InitableTask* interface. > > Otherwise, the content in the init method will not be processed. > > > > Thanks, > > > > Fang, Yan > > yanfang...@gmail.com > > > > On Mon, Aug 10, 2015 at 3:24 AM, Jordi Blasi Uribarri > > <jbl...@nextel.es> > > wrote: > > > > > Just for making it easier to reproduce the problem I just reduced > > > the > > code > > > of the job to the minimum: > > > > > > package test; > > > > > > import org.apache.samza.config.Config; import > > > org.apache.samza.storage.kv.KeyValueStore; > > > import org.apache.samza.system.IncomingMessageEnvelope; > > > import org.apache.samza.task.MessageCollector; > > > import org.apache.samza.task.StreamTask; import > > > org.apache.samza.task.TaskContext; > > > import org.apache.samza.task.TaskCoordinator; > > > > > > public class testStore implements StreamTask { > > > private KeyValueStore<String, String> storestp; > > > > > > public void init(Config config, TaskContext context) { > > > this.storestp = (KeyValueStore<String, String>) > > > context.getStore("test11db"); > > > } > > > > > > public void process (IncomingMessageEnvelope envelope, > > > MessageCollector collector, > > > TaskCoordinator coordinator) > > > { > > > String msgin = (String) envelope.getMessage(); > > > storestp.put("test1",msgin); > > > } > > > } > > > > > > The properties file contains this: > > > > > > task.class=test.testStore > > > job.name=test.testStore > > > job.factory.class=org.apache.samza.job.yarn.YarnJobFactory > > > yarn.package.path=http://192.168.15.92/jobs/nxtBroker-0.0.1-bin.tar. > > > gz > > > > > > > > > > > systems.kafka.samza.factory=org.apache.samza.system.kafka.KafkaSystemF > > actory > > > > > systems.kafka.consumer.zookeeper.connect=kfk-kafka01:2181,kfk-kafka02: > > 2181 > > > > > > > > systems.kafka.producer.bootstrap.servers=kfk-kafka01:9092,kfk-kafka01: > > 9093,kfk-kafka02:9092,kfk-kafka02:9093 > > > > > > > > systems.kafka.producer.metadata.broker.list=kfk-kafka01:9092,kfk-kafka > > 01:9093,kfk-kafka02:9092,kfk-kafka02:909 > > > > > > # Declare that we want our job's checkpoints to be written to Kafka > > > > > > > > task.checkpoint.factory=org.apache.samza.checkpoint.kafka.KafkaCheckpo > > intManagerFactory > > > task.checkpoint.system=kafka > > > > > > # The job consumes a topic called "configtpc" from the "kafka" > > > system task.inputs=kafka.configtpc > > > > > > # Define a serializer/deserializer called "json" which parses JSON > > messages > > > > > > > > serializers.registry.json.class=org.apache.samza.serializers.JsonSerde > > Factory > > > > > > > > serializers.registry.string.class=org.apache.samza.serializers.StringS > > erdeFactory > > > > > > # Serializer for the system > > > systems.kafka.samza.msg.serde=string > > > systems.kafka.streams.tracetpc.samza.msg.serde=json > > > > > > # Use the key-value store implementation for a store called "my-store" > > > > > > > > stores.test11db.factory=org.apache.samza.storage.kv.RocksDbKeyValueSto > > rageEngineFactory > > > > > > # Use the Kafka topic "routingdb-changelog" as the changelog stream > > > for this store. > > > # This enables automatic recovery of the store after a failure. If > > > you don't # configure this, no changelog stream will be generated. > > > stores.test11db.changelog=kafka.test11db-changelog > > > > > > # Encode keys and values in the store as UTF-8 strings. > > > stores.test11db.key.serde=string > > > stores.test11db.msg.serde=string > > > > > > # Commit checkpoints every 1 seconds task.commit.ms=1000 > > > > > > With this, I am getting just the same error: > > > > > > java version "1.7.0_79" > > > OpenJDK Runtime Environment (IcedTea 2.5.6) (7u79-2.5.6-1~deb7u1) > > > OpenJDK 64-Bit Server VM (build 24.79-b02, mixed mode) log4j:WARN No > > > appenders could be found for logger > > > (org.apache.samza.metrics.JmxServer). > > > log4j:WARN Please initialize the log4j system properly. > > > log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig > > > for more info. > > > Exception in thread "main" org.apache.samza.SamzaException: Missing > > > a change log offset for SystemStreamPartition [kafka, > > > test11db-changelog, > > 2]. > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3$ > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3$ > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > at scala.collection.MapLike$class.getOrElse(MapLike.scala:128) > > > at scala.collection.AbstractMap.getOrElse(Map.scala:58) > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > apply(TaskStorageManager.scala:87) > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > apply(TaskStorageManager.scala:84) > > > at > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(T > > raversableLike.scala:772) > > > at > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(MapLike > > .scala:245) > > > at > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(MapLike > > .scala:245) > > > at > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(T > > raversableLike.scala:772) > > > at scala.collection.immutable.Map$Map1.foreach(Map.scala:109) > > > at > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.sc > > ala:771) > > > at > > scala.collection.MapLike$MappedValues.foreach(MapLike.scala:245) > > > at > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.sc > > ala:771) > > > at > > > > > org.apache.samza.storage.TaskStorageManager.startConsumers(TaskStorage > > Manager.scala:84) > > > at > > > > > org.apache.samza.storage.TaskStorageManager.init(TaskStorageManager.sc > > ala:63) > > > at > > > > > org.apache.samza.container.TaskInstance.startStores(TaskInstance.scala > > :88) > > > at > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.apply > > (SamzaContainer.scala:607) > > > at > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.apply > > (SamzaContainer.scala:607) > > > at scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > at > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > at > > > > scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:206) > > > at > > > > > org.apache.samza.container.SamzaContainer.startStores(SamzaContainer.s > > cala:607) > > > at > > > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:550) > > > at > > > > > org.apache.samza.container.SamzaContainer$.safeMain(SamzaContainer.sca > > la:108) > > > at > > > > org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala:87) > > > at > > > org.apache.samza.container.SamzaContainer.main(SamzaContainer.scala) > > > > > > > > > The job fails even when there is no message sent to the input topic. > > > > > > Samza is version 0.9.1 and kafka 0.8.2. > > > > > > Thanks, > > > > > > Jordi > > > > > > -----Mensaje original----- > > > De: Jordi Blasi Uribarri [mailto:jbl...@nextel.es] Enviado el: > > > lunes, 10 de agosto de 2015 10:26 > > > Para: dev@samza.apache.org > > > Asunto: RE: Missing a change log offset for SystemStreamPartition > > > > > > Hi, > > > > > > I have migrated samza to the last versión and recreated the job with > > > a > > new > > > store name so the streams were created clean. I am getting the same > > error: > > > > > > java version "1.7.0_79" > > > OpenJDK Runtime Environment (IcedTea 2.5.6) (7u79-2.5.6-1~deb7u1) > > > OpenJDK 64-Bit Server VM (build 24.79-b02, mixed mode) log4j:WARN No > > > appenders could be found for logger > (org.apache.samza.metrics.JmxServer). > > > log4j:WARN Please initialize the log4j system properly. > > > log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig > > > for more info. > > > Exception in thread "main" org.apache.samza.SamzaException: Missing > > > a change log offset for SystemStreamPartition [kafka, > commdb-changelog, 2]. > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3$ > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3$ > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > at scala.collection.MapLike$class.getOrElse(MapLike.scala:128) > > > at scala.collection.AbstractMap.getOrElse(Map.scala:58) > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > apply(TaskStorageManager.scala:87) > > > at > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > apply(TaskStorageManager.scala:84) > > > at > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(T > > raversableLike.scala:772) > > > at > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(MapLike > > .scala:245) > > > at > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(MapLike > > .scala:245) > > > at > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(T > > raversableLike.scala:772) > > > at scala.collection.immutable.Map$Map1.foreach(Map.scala:109) > > > at > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.sc > > ala:771) > > > at > > scala.collection.MapLike$MappedValues.foreach(MapLike.scala:245) > > > at > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.sc > > ala:771) > > > at > > > > > org.apache.samza.storage.TaskStorageManager.startConsumers(TaskStorage > > Manager.scala:84) > > > at > > > > > org.apache.samza.storage.TaskStorageManager.init(TaskStorageManager.sc > > ala:63) > > > at > > > > > org.apache.samza.container.TaskInstance.startStores(TaskInstance.scala > > :88) > > > at > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.apply > > (SamzaContainer.scala:607) > > > at > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.apply > > (SamzaContainer.scala:607) > > > at scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > at > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > at > > > > scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:206) > > > at > > > > > org.apache.samza.container.SamzaContainer.startStores(SamzaContainer.s > > cala:607) > > > at > > > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:550) > > > at > > > > > org.apache.samza.container.SamzaContainer$.safeMain(SamzaContainer.sca > > la:108) > > > at > > > > org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala:87) > > > at > > > org.apache.samza.container.SamzaContainer.main(SamzaContainer.scala) > > > > > > Is there any other info I can attach to help find the problem? > > > > > > Thanks, > > > > > > Jordi > > > > > > -----Mensaje original----- > > > De: Yan Fang [mailto:yanfang...@gmail.com] Enviado el: viernes, 07 > > > de agosto de 2015 23:21 > > > Para: dev@samza.apache.org > > > Asunto: Re: Missing a change log offset for SystemStreamPartition > > > > > > Hi Jordi, > > > > > > Sorry for getting you back late. Was quite busy yesterday. > > > > > > I think the reason of your error is that you mismatched Samza > > > version and Kafka version. > > > > > > Seems that, you are using Samza 0.8.1 with Kafka 0.8.2, which is not > > > supported. > > > > > > So my suggestion is to upgrade to *Samza 0.9.1*, then use *Kafka > 0.8.2*. > > > This match is proved working. > > > > > > Hope this helps you. > > > > > > Thanks, > > > > > > > > > Fang, Yan > > > yanfang...@gmail.com > > > > > > On Thu, Aug 6, 2015 at 1:16 AM, Jordi Blasi Uribarri > > > <jbl...@nextel.es> > > > wrote: > > > > > > > I changed the job name and the store name. I was defining two > > > > different stores and in case that was the problem, I also > > > > eliminated > > the > > > second one. > > > > I am getting the same exception. > > > > > > > > Exception in thread "main" org.apache.samza.SamzaException: > > > > Missing a change log offset for SystemStreamPartition [kafka, > > > > testdb-changelog, > > 2]. > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3$ > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3$ > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > > at > scala.collection.MapLike$class.getOrElse(MapLike.scala:128) > > > > at scala.collection.AbstractMap.getOrElse(Map.scala:58) > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > apply(TaskStorageManager.scala:87) > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > apply(TaskStorageManager.scala:84) > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(T > > raversableLike.scala:772) > > > > at > > > > > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(MapLike > > .scala:245) > > > > at > > > > > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(MapLike > > .scala:245) > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(T > > raversableLike.scala:772) > > > > at scala.collection.immutable.Map$Map1.foreach(Map.scala:109) > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.sc > > ala:771) > > > > at > > > scala.collection.MapLike$MappedValues.foreach(MapLike.scala:245) > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.sc > > ala:771) > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager.startConsumers(TaskStorage > > Manager.scala:84) > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager.init(TaskStorageManager.sc > > ala:63) > > > > at > > > > > > > > > org.apache.samza.container.TaskInstance.startStores(TaskInstance.scala > > :88) > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.apply > > (SamzaContainer.scala:607) > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.apply > > (SamzaContainer.scala:607) > > > > at > scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > > at > > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > > at > > > > > > scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:2 > > 06) > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer.startStores(SamzaContainer.s > > cala:607) > > > > at > > > > > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:550) > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$.safeMain(SamzaContainer.sca > > la:108) > > > > at > > > > > > org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala:8 > > 7) > > > > at > > > > org.apache.samza.container.SamzaContainer.main(SamzaContainer.scal > > > > a) > > > > > > > > As I have the autocreate configured in Kafka I am not creating > > > > anything for the store. Is that ok? > > > > > > > > By the way, is there any problem on having two different stores? > > > > > > > > Thanks, > > > > > > > > Jordi > > > > > > > > -----Mensaje original----- > > > > De: Yan Fang [mailto:yanfang...@gmail.com] Enviado el: miércoles, > > > > 05 de agosto de 2015 20:23 > > > > Para: dev@samza.apache.org > > > > Asunto: Re: Missing a change log offset for SystemStreamPartition > > > > > > > > Hi Jordi, > > > > > > > > I wonder, the reason of your first exception is that, you changed > > > > the task number (partition number of your input stream), but still > > > > were using the same changelog stream. It is trying to send to the > > > > partition 2, which does not exist? > > > > > > > > Can you reproduce this exception in a new job? (new store name, > > > > new job > > > > name) > > > > > > > > The second exception is caused by the wrong offset format, I believe. > > > > > > > > Let me know how the new job goes. > > > > > > > > Thanks, > > > > > > > > Fang, Yan > > > > yanfang...@gmail.com > > > > > > > > On Wed, Aug 5, 2015 at 12:51 AM, Jordi Blasi Uribarri > > > > <jbl...@nextel.es> > > > > wrote: > > > > > > > > > Hi, > > > > > > > > > > I am trying to use the Keystore to manage some state information. > > > > > Basically this is the code I am using. As long as I have tested, > > > > > the rest is working correctly. > > > > > > > > > > private KeyValueStore<String, String> storestp; > > > > > > > > > > public void init(Config config, TaskContext context) { > > > > > this.storestp = (KeyValueStore<String, String>) > > > > > context.getStore("stepdb"); > > > > > } > > > > > > > > > > public void process(IncomingMessageEnvelope envelope, > > > > > MessageCollector collector, > > > > > TaskCoordinator coordinator) > > > > > { > > > > > … > > > > > String str = storestp.get(code) > > > > > … > > > > > } > > > > > > > > > > When I load it, it goes to running but, whe I send the messages > > > > > through Kafka stream It goes to Failed state. I have found this > > > > Exception: > > > > > Exception in thread "main" org.apache.samza.SamzaException: > > > > > Missing a change log offset for SystemStreamPartition [kafka, > > > stepdb-changelog, 2]. > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumer > > > > s$3$ > > > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumer > > > > s$3$ > > > > $anonfun$1.apply(TaskStorageManager.scala:87) > > > > > at > > scala.collection.MapLike$class.getOrElse(MapLike.scala:128) > > > > > at scala.collection.AbstractMap.getOrElse(Map.scala:58) > > > > > at > > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > > > apply(TaskStorageManager.scala:87) > > > > > at > > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$startConsumers$3. > > > > apply(TaskStorageManager.scala:84) > > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.app > > > > ly(T > > > > raversableLike.scala:772) > > > > > at > > > > > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(Map > > > > Like > > > > .scala:245) > > > > > at > > > > > > > > > scala.collection.MapLike$MappedValues$$anonfun$foreach$3.apply(Map > > > > Like > > > > .scala:245) > > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.app > > > > ly(T > > > > raversableLike.scala:772) > > > > > at > scala.collection.immutable.Map$Map2.foreach(Map.scala:130) > > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLik > > > > e.sc > > > > ala:771) > > > > > at > > > > scala.collection.MapLike$MappedValues.foreach(MapLike.scala:245) > > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLik > > > > e.sc > > > > ala:771) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager.startConsumers(TaskSto > > > > rage > > > > Manager.scala:84) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager.init(TaskStorageManage > > > > r.sc > > > > ala:63) > > > > > at > > > > > > > > > org.apache.samza.container.TaskInstance.startStores(TaskInstance.s > > > > cala > > > > :88) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.a > > > > pply > > > > (SamzaContainer.scala:607) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.a > > > > pply > > > > (SamzaContainer.scala:607) > > > > > at > > scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > > > at > > > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > > > at > > > > > > > > scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala > > > :206) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer.startStores(SamzaContain > > > > er.s > > > > cala:607) > > > > > at > > > > > > > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:550 > > ) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$.safeMain(SamzaContainer > > > > .sca > > > > la:108) > > > > > at > > > > > > > > org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala > > > :87) > > > > > at > > > > > org.apache.samza.container.SamzaContainer.main(SamzaContainer.sc > > > > > ala) > > > > > > > > > > I have seen that the stepdb-changelog stream exists in Kafka. As > > > > > a try to regenerate the missing offset and tes it I have > > > > > connected through the command line and send a message to the > > > > > stream. It was received > > > > correctly. > > > > > Now I am seeing the following Exception: > > > > > > > > > > Exception in thread "main" java.lang.NullPointerException > > > > > at > > > > > > > > > scala.collection.mutable.ArrayOps$ofByte$.length$extension(ArrayOp > > > > s.sc > > > > ala:126) > > > > > at > > > > > scala.collection.mutable.ArrayOps$ofByte.length(ArrayOps.scala:126) > > > > > at scala.collection.SeqLike$class.size(SeqLike.scala:106) > > > > > at > > > > > scala.collection.mutable.ArrayOps$ofByte.size(ArrayOps.scala:120) > > > > > at > > > > > > > > > org.apache.samza.storage.kv.KeyValueStorageEngine$$anonfun$restore > > > > $1.a > > > > pply(KeyValueStorageEngine.scala:94) > > > > > at > > > > > > > > > org.apache.samza.storage.kv.KeyValueStorageEngine$$anonfun$restore > > > > $1.a > > > > pply(KeyValueStorageEngine.scala:79) > > > > > at > > scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > > > at > > > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > > > at > > > > > > > > > org.apache.samza.storage.kv.KeyValueStorageEngine.restore(KeyValue > > > > Stor > > > > ageEngine.scala:79) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$restoreStores > > > > $3.a > > > > pply(TaskStorageManager.scala:112) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager$$anonfun$restoreStores > > > > $3.a > > > > pply(TaskStorageManager.scala:106) > > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.app > > > > ly(T > > > > raversableLike.scala:772) > > > > > at > scala.collection.immutable.Map$Map1.foreach(Map.scala:109) > > > > > at > > > > > > > > > scala.collection.TraversableLike$WithFilter.foreach(TraversableLik > > > > e.sc > > > > ala:771) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager.restoreStores(TaskStor > > > > ageM > > > > anager.scala:106) > > > > > at > > > > > > > > > org.apache.samza.storage.TaskStorageManager.init(TaskStorageManage > > > > r.sc > > > > ala:64) > > > > > at > > > > > > > > > org.apache.samza.container.TaskInstance.startStores(TaskInstance.s > > > > cala > > > > :88) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.a > > > > pply > > > > (SamzaContainer.scala:607) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$$anonfun$startStores$2.a > > > > pply > > > > (SamzaContainer.scala:607) > > > > > at > > scala.collection.Iterator$class.foreach(Iterator.scala:727) > > > > > at > > > scala.collection.AbstractIterator.foreach(Iterator.scala:1157) > > > > > at > > > > > > > > scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala > > > :206) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer.startStores(SamzaContain > > > > er.s > > > > cala:607) > > > > > at > > > > > > > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:550 > > ) > > > > > at > > > > > > > > > org.apache.samza.container.SamzaContainer$.safeMain(SamzaContainer > > > > .sca > > > > la:108) > > > > > at > > > > > > > > org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala > > > :87) > > > > > at > > > > > org.apache.samza.container.SamzaContainer.main(SamzaContainer.sc > > > > > ala) > > > > > > > > > > Is there something wrong? > > > > > > > > > > Thanks, > > > > > > > > > > Jordi > > > > > ________________________________ Jordi Blasi Uribarri Área I+D+i > > > > > > > > > > jbl...@nextel.es > > > > > Oficina Bilbao > > > > > > > > > > [http://www.nextel.es/wp-content/uploads/Firma_Nextel_2015.png] > > > > > > > > > > > > > > >
