Every time, a job scans the directory and picks the valid latest job state. It solves problem 1 and 2 but is not efficient especially in the case where a job has a long running history.  

This approach can be improved by periodically archiving old job states. Listing and sorting over the non-archived job states will become more efficient. Data integrity can be done by storing a checksum into the job state. However, we are seeing performance degrade caused by saving state store to HDFS.   

Option 3: `current.jst.bk` 

Before deleting the `current.jst` in phase 3, a copy of it is persisted as a back file. If a job can't find `current.jst`, it will read `current.jst.bk`. Problem 1 is resolved, but it doesn't address problem 2.  

 Option 4: DB 

All of the efforts lead to an implementation of transactional storage, which is what DB is good at. This approach can solve problem 1 and 2 efficiently.  

In terms of what DB to use and whether it is a client/server based or embedded db, the table below demonstrates the pros/cons of MySQL C/S and Embedded RocksDB, which supports HDFS.  

Job instance isolation means isolating the state store of the same job running by different users in the same cluster or environment.  

If a job requires fault tolerant state store, it should use a MySQL (C/S) state store. Task: Migrating jobs to use MySQL state store. 

Access control to MySQL state store in different environments(dev, prod) is required to prevent engineers working in dev to break state in production.  Add permission on if a user is allowed to use any primary key, avoid corrupting states stored by its actual owner  

Kafka audit count exactly once 

Currently, the kafka audit count is reported but the job may fail or exit without persisting its state. Next time, a new job is launched and the same audit count is reported again.  To approach the solution:  

Step1: Report aggregated Kafka audit count in the driver 

A dataset may have multiple task states. Kafka audit count will be sent in a dataset level instead of task level:  

1. Aggregate Kafka audit count from task states of the dataset 
2.  Put the aggregated count into the dataset state 
3.  Persist dataset state is successfully 
4.  Report the aggregated count   

With step1, it's guaranteed that failed or killed job won't report bogus audit count   

Step 2: Exactly once assistance from Kafka audit server  

If Kafka audit server can ignore duplicate audit count report by looking at its key/content, an exactly once semantic can be achieved by adding a checkpoint operation:   

5. Mark the dataset state is `Audited`   

Next time, the new job resends the audit count of the last known good dataset state if its status is not `Audited`   

Conclusion 

It is not clear whether Kafka Audit server can help us with the exactly once semantic, although the team is working on some kind of deduping audit messages. Step 1 only just reduces the failure window and error cases. It doesn't solve the issue.  

Lock revocation 

When a job suddenly exits withoutt cleaning its lock, no subsequent execution can be successfully launched as the existence of the lock is perceived as a previous job is running, even though it is not.  

Option 1: Remove the lock file in shutdown hook 

It's likely that removing the lock, especially from a distributed file system, takes a long time that exceeds the time limit of a shutdown hook in JVM. We could configure a reasonable ttl. But it won't work all the time.  

Option 2: Remove the lock while the rebooting gobblin service 

It's likely that the same gobblin service can be accidently deployed twice, booting the second service ends up removing the lock placed by the first one, breaking the semantic of the job lock.  

 Option 3: Use locks that require heartbeat renewal 

Use a zookeeper based lock. The owner renews the lease of the lock periodically. A lock renewal service needs implementing and designing to avoid hanging in any case, otherwise, the lease will expire and the lock will be claimed by another process, breaking the state of the current job.  

Conclusion 

If a job requires lock auto revocation, it should use a zookeeper based lock, which is already implemented. Task: implement lock renewal service  

Summary 

There are various implementations of state stores and locks, each providing different set of features. To achieve gracefully shutdown, a job should use MySQL state store and Zookeeper based lock. Tasks are, in order of priority:  

1. Migrating jobs to use MySQL state store. If the time commitment is urgent, we can implement Option 2 (with archiving and checksum) for the sake of the existing jobs. New jobs are recommended to use MySQL state store   
2. Implement log renewal service  
3.  Implement reporting aggregated Kafka audit count in the driver    

Without support from kafka audit server, we haven't found a solution to fix the exactly once semantic of kafka audit count, However, implementing step 1 (task 3) only will mitigate the issue, reducing the failure window and error cases.