Jeremiah, If you ensure the base directory is exclusive to your state stores, e.g. "./state-stores" or any similarly exclusive path, that should work.
Thanks, Brett On Tue, Jan 28, 2020 at 8:07 AM Jeremiah Adams <jad...@helixeducation.com.invalid> wrote: > Thanks Brett. > > In the short term, can I fix this by specifying an alternate directory? It > will still be empty on initial startup. > > > > Jeremiah Adams > Software Engineer > www.helixeducation.com > Blog | Twitter | Facebook | LinkedIn > > ________________________________________ > From: Brett Konold <brettkon...@gmail.com> > Sent: Tuesday, January 28, 2020 1:27 AM > To: dev@samza.apache.org > Subject: Re: Nullpointer After Upgrade from samza 1.0.0 to 1.3.0 > > Thanks for sharing Jeremiah. The infringing code is calling listFiles on > every file in the base directory, and since you'd configured that to be the > current directory of the job, there will be other non-store related > contents there as well some of which may be empty directories or not > directories at all. This is the reason for the NPE. > > I've created > https://url.emailprotection.link/?bb4dcsrIoLyiCTjGiqSsT_0LnLYpW7LTqyGmsCgDxt72IyPMS1mxdVxwJ7bI1sDQscs04YbqL0eWxl9A9wpYR3u44rRH8NrkqY28e6GFAlDrQ7EbO2m0JAgkH2_um6A4d > to track, you > can follow that for updates as I get a patch ready. > > Thanks, > Brett > > On Fri, Jan 24, 2020 at 7:03 AM Jeremiah Adams > <jad...@helixeducation.com.invalid> wrote: > > > Thanks for following up Brett. > > > > It is a RocksDB data store on the local disk. I am trying to get it > > runnign before deploying it as a standalone job in a docker container. > Yes, > > the directory is empty in this case. > > > > local config: > > job.logged.store.base.dir=. > > job.non-logged.store.base.dir=. > > > > new RocksDbTableDescriptor(tableName, KVSerde.of(new StringSerde(), new > > StringSerde()))).withTtl(this.rocksDbTtl); > > > > > > Jeremiah Adams > > Software Engineer > > > https://url.emailprotection.link/?bM9S-3pRw1lv8pYfwa-TwdjElP4W2K6b9vP5Crz22L_YcgsRJ-13h-OgPZSwFtU7GSNTDi1z-jdaRvWESRhtTVA~~ > > Blog | Twitter | Facebook | LinkedIn > > > > ________________________________________ > > From: Brett Konold <brettkon...@gmail.com> > > Sent: Thursday, January 23, 2020 3:45 PM > > To: Prateek Maheshwari > > Cc: dev@samza.apache.org > > Subject: Re: Nullpointer After Upgrade from samza 1.0.0 to 1.3.0 > > > > Hey Jeremiah, > > > > It looks like TransactionalStateTaskStorageManager.removeOldCheckpoints > is > > relying on an assumption that the store directory actually has some > > contents, which may not be true depending on the type of store being > used. > > > > Can you let me know what kind of KeyValueStore you've configured for your > > state? > > > > Brett > > > > On Thu, Jan 23, 2020 at 2:31 PM Prateek Maheshwari <prateek...@gmail.com > > > > wrote: > > > > > Brett, can you take a look at this? > > > > > > - Prateek > > > > > > On Wed, Jan 15, 2020 at 9:41 AM Jeremiah Adams > > > <jad...@helixeducation.com.invalid> wrote: > > > > > >> I am updating our jobs to use samza 1.3.0. I'm getting a null pointer > > >> when manually committing via taskCoordinator.commit(). > > >> > > >> > > >> Below is the stack trace - can anyone point me in the right direction? > > >> > > >> Thanks. > > >> > > >> > > >> 2020-01-15 10:33:35 RunLoop [ERROR] Task Partition 0 commit failed > > >> java.lang.NullPointerException > > >> at > > >> > > > scala.collection.mutable.ArrayOps$ofRef$.newBuilder$extension(ArrayOps.scala:190) > > >> at > > >> scala.collection.mutable.ArrayOps$ofRef.newBuilder(ArrayOps.scala:186) > > >> at > > >> > > > scala.collection.TraversableLike$class.filterImpl(TraversableLike.scala:246) > > >> at > > >> > scala.collection.TraversableLike$class.filter(TraversableLike.scala:259) > > >> at > > scala.collection.mutable.ArrayOps$ofRef.filter(ArrayOps.scala:186) > > >> at > > >> > > > org.apache.samza.storage.TransactionalStateTaskStorageManager$$anonfun$removeOldCheckpoints$2.apply(TransactionalStateTaskStorageManager.scala:94) > > >> at > > >> > > > org.apache.samza.storage.TransactionalStateTaskStorageManager$$anonfun$removeOldCheckpoints$2.apply(TransactionalStateTaskStorageManager.scala:86) > > >> at > > >> > > > scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) > > >> at > > scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186) > > >> at > > >> > > > org.apache.samza.storage.TransactionalStateTaskStorageManager.removeOldCheckpoints(TransactionalStateTaskStorageManager.scala:86) > > >> at > > >> org.apache.samza.container.TaskInstance.commit(TaskInstance.scala:277) > > >> at > > >> > > > org.apache.samza.container.RunLoop$AsyncTaskWorker$5.run(RunLoop.java:547) > > >> at > > >> > > > org.apache.samza.container.RunLoop$AsyncTaskWorker.commit(RunLoop.java:566) > > >> at > > >> > org.apache.samza.container.RunLoop$AsyncTaskWorker.run(RunLoop.java:432) > > >> at > > >> > > > org.apache.samza.container.RunLoop$AsyncTaskWorker.access$300(RunLoop.java:357) > > >> at org.apache.samza.container.RunLoop.runTasks(RunLoop.java:244) > > >> at org.apache.samza.container.RunLoop.run(RunLoop.java:176) > > >> at > > >> > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:768) > > >> at > > >> > java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) > > >> at > > java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) > > >> at java.util.concurrent.FutureTask.run(FutureTask.java) > > >> at > > >> > > > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > > >> at > > >> > > > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > > >> at java.lang.Thread.run(Thread.java:748) > > >> 2020-01-15 10:33:35 RunLoop [ERROR] Caught throwable and stopping run > > loop > > >> java.lang.NullPointerException > > >> at > > >> > > > scala.collection.mutable.ArrayOps$ofRef$.newBuilder$extension(ArrayOps.scala:190) > > >> at > > >> scala.collection.mutable.ArrayOps$ofRef.newBuilder(ArrayOps.scala:186) > > >> at > > >> > > > scala.collection.TraversableLike$class.filterImpl(TraversableLike.scala:246) > > >> at > > >> > scala.collection.TraversableLike$class.filter(TraversableLike.scala:259) > > >> at > > scala.collection.mutable.ArrayOps$ofRef.filter(ArrayOps.scala:186) > > >> at > > >> > > > org.apache.samza.storage.TransactionalStateTaskStorageManager$$anonfun$removeOldCheckpoints$2.apply(TransactionalStateTaskStorageManager.scala:94) > > >> at > > >> > > > org.apache.samza.storage.TransactionalStateTaskStorageManager$$anonfun$removeOldCheckpoints$2.apply(TransactionalStateTaskStorageManager.scala:86) > > >> at > > >> > > > scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) > > >> at > > scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186) > > >> at > > >> > > > org.apache.samza.storage.TransactionalStateTaskStorageManager.removeOldCheckpoints(TransactionalStateTaskStorageManager.scala:86) > > >> at > > >> org.apache.samza.container.TaskInstance.commit(TaskInstance.scala:277) > > >> at > > >> > > > org.apache.samza.container.RunLoop$AsyncTaskWorker$5.run(RunLoop.java:547) > > >> at > > >> > > > org.apache.samza.container.RunLoop$AsyncTaskWorker.commit(RunLoop.java:566) > > >> at > > >> > org.apache.samza.container.RunLoop$AsyncTaskWorker.run(RunLoop.java:432) > > >> at > > >> > > > org.apache.samza.container.RunLoop$AsyncTaskWorker.access$300(RunLoop.java:357) > > >> at org.apache.samza.container.RunLoop.runTasks(RunLoop.java:244) > > >> at org.apache.samza.container.RunLoop.run(RunLoop.java:176) > > >> at > > >> > org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:768) > > >> at > > >> > java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) > > >> at > > java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) > > >> at java.util.concurrent.FutureTask.run(FutureTask.java) > > >> at > > >> > > > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) > > >> at > > >> > > > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) > > >> at java.lang.Thread.run(Thread.java:748) > > >> > > >> > > >> Jeremiah Adams > > >> Software Engineer > > >> > > > https://url.emailprotection.link/?bM9S-3pRw1lv8pYfwa-TwdjElP4W2K6b9vP5Crz22L_YcgsRJ-13h-OgPZSwFtU7GSNTDi1z-jdaRvWESRhtTVA~~ > > < > > > https://url.emailprotection.link/?basKr9vk92a8vVw0XMnK5bmaSKuBc0AuEZ7YasYc7Df8YVt3SYmcjmLWdKMWzAAINWlUUA33ebGI7pSoTl9cg1g~~ > > > > > >> Blog< > > > https://url.emailprotection.link/?basKr9vk92a8vVw0XMnK5bmaSKuBc0AuEZ7YasYc7Df-lAcqG1fqHPpNw-wd9z7HtUJeCG5_8UjCf2mHtn6C_zQ~~ > > > > | Twitter< > > >> > > > https://url.emailprotection.link/?bVO2q0UXR235wN_yOnM0FjqITPdBYMD3reLGNddq-zPV5ChMQK9JwV4Be-QnrbRoXpJl8IcknAqKzYtA3RABKww~~ > > > > | Facebook< > > >> > > > https://url.emailprotection.link/?bUU7m4NfMS_EWGtH1yojBHX9sWZ6uxVdT1eQUkmU5vWY01WFZiS2KJ-c9iLIncdHB7Uw1lRYCprEEpPPQCdiK6Q~~ > > > > | LinkedIn< > > >> > > > https://url.emailprotection.link/?b0ZQfJ1pZYnASyoShs9MJI46-r1lxPhA-JS5VSkR7so-DFP0_HxbOo2LsajGOaoYXxb1ZCOMAu7hZscPCnIKWpXz0cpgQ386SnNHjPcwsu4z90mzBkuwoZc6YxOCzMGA0 > > > > > >> > > > > > >
