Hi, Yi, I couldn't find any errors in the log indicating any issue writing to those particular changelog partitions. I event went ahead and removed all checkpoint, coordinator and changelog topics and started fresh. This issue is still manifested itself with offsets of 0:
partition offset ---------------------- 0 0 1 54251 2 54315 3 54196 4 53548 5 53581 6 55175 7 54599 8 53694 9 0 10 53456 11 53450 12 0 13 54442 14 54759 15 54958 16 54909 17 53396 18 55442 19 54121 In this case, we have partition 0, 2, 8 and 14 all running in the same YARN container, which means that it's not a container specific issue (since partition 2, 8 and 14 all get proper changelogs written). As I mentioned earlier, the changelog topic was auto-created by the samza job. So no manual overrides such as "auto-commit" was given. Thanks, David On Mon, Jun 13, 2016 at 9:40 AM, Yi Pan <nickpa...@gmail.com> wrote: > Hi, David, > > Did you check the log to see whether there is any log lines indicating the > producer issues on the three partitions that you suspect? And could you > also check whether you have auto-commit turned on? If your auto-commit is > on and producer does not report any issue writing to the changelog topic, > you may want to do a comparison between the local RocksDB and the one > persisted in changelog to verify that there indeed are some discrepancies > between them. Samza provides a command line tool state-storage-tool.sh to > recover the RocksDB state store from the changelog. You can use it to > recover the state store from changelog and compare w/ the local RocksDB to > verify if there is any discrepancies. > > Best. > > -Yi > > On Sun, Jun 12, 2016 at 12:49 PM, David Yu <david...@optimizely.com> > wrote: > > > Jagadish, > > > > All your description matches my understand. > > > > Here are our settings: > > - Our task aggregates user events into user sessions. > > - We have one k-v store for each task, which tracks active user sessions > > (with sessionId as the key). > > - When a user session expires, the session will be removed from the > store. > > - The changelog topic was auto created with cleanup.policy=*compact*. > > > > In terms of log compaction, I'm expecting it to keep the last log entry > for > > a given key and deletes all previous entries. For example, if we have: > > > > store.put("session1", Session1_1) // session created > > store.put("session1", Session1_2) // session updated > > store.delete("session1") // session expired > > > > > > I'm expecting something as following in the changelog (after compaction): > > > > 1 session1=Session1_1 > > 2 session1=Session1_2 > > 3 session1=NULL > > > > > > with only offset 3 retained. The next log entry should take offset 4. In > > that sense, the offsets should always increase monotonically, with lots > of > > gaps in between due to compaction. > > > > So again, I'm not sure why we have three changelog partitions that stop > > seeing movements in their offsets. > > > > Thanks, > > David > > > > On Sun, Jun 12, 2016 at 11:09 AM, Jagadish Venkatraman < > > jagadish1...@gmail.com> wrote: > > > > > Some context: Each k-v store has a changelog topic. The # of partitions > > in > > > that changelog topic is equal to the # of tasks. Each task's K-V store > > will > > > be mapped to a particular partition of that changelog topic. This > mapping > > > from taskNames-changeLogPartitionNumber is stored in coordinator > stream. > > > > > > > Of course, you don't want this k-v changelog topic to keep growing. So, > > > people configure it with some expiration. The expiration can either be: > > > 1. Time retention: Records older than the retention are purged. > > > 2. Compaction: Newer key-values will over-write older keys and only the > > > most recent value is retained. > > > > > > I'm not sure if offsets are always monotonically increasing in Kafka or > > > could change after a compaction/ a time based retention kicks in for > the > > > topic partition. > > > > > > > > > > > > > > > > > > On Sat, Jun 11, 2016 at 11:53 PM, David Yu <david...@optimizely.com> > > > wrote: > > > > > > > My understanding of store changelog is that, each task writes store > > > changes > > > > to a particular changelog partition for that task. (Does that mean > the > > > > changelog keys are task names?) > > > > > > > > One thing that confuses me is that, the last offsets of some > changelog > > > > partitions do not move. I'm using the kafka GetOffsetShell tool to > get > > > the > > > > last offsets for each partition. The result looks like this: > > > > > > > > partition offset > > > > 0 7090 > > > > 1 3737937 > > > > 2 3733222 > > > > 3 3719065 > > > > 4 3730208 > > > > 5 3731128 > > > > 6 3734669 > > > > 7 3691461 > > > > 8 3759133 > > > > 9 7286 > > > > 10 3690347 > > > > 11 3722450 > > > > 12 7376 > > > > 13 3738454 > > > > 14 3742316 > > > > 15 3710512 > > > > 16 3777267 > > > > 17 3750596 > > > > 18 3728185 > > > > 19 3694470 > > > > > > > > As you can see, three of the partitions barely got any updates. In > > fact, > > > > the offsets stopped moving for a while. The traffic for each task > > should > > > be > > > > fairly balanced. I checked the task log and made sure that the stores > > for > > > > these partitions are actively updated. > > > > > > > > Any idea why this is happening? Or am I missing something? > > > > > > > > Thanks, > > > > David > > > > > > > > > > > > > > > > -- > > > Jagadish V, > > > Graduate Student, > > > Department of Computer Science, > > > Stanford University > > > > > >