update ScheduledJobParser to use the JobDocVersion

sample-extension-plugin/src/main/java/com/amazon/opendistroforelasticsearch/jobscheduler/sampleextension/SampleExtensionPlugin.java

@@ -91,7 +91,7 @@ public ScheduledJobRunner getJobRunner() {
 
     @Override
     public ScheduledJobParser getJobParser() {
-        return (parser, id, version) -> {
+        return (parser, id, jobDocVersion) -> {
             SampleJobParameter jobParameter = new SampleJobParameter();
             XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser::getTokenLocation);
 

spi/src/main/java/com/amazon/opendistroforelasticsearch/jobscheduler/spi/ScheduledJobParser.java

@@ -20,5 +20,5 @@
 import java.io.IOException;
 
 public interface ScheduledJobParser {
-    ScheduledJobParameter parse(XContentParser xContentParser, String id, Long version) throws IOException;
+    ScheduledJobParameter parse(XContentParser xContentParser, String id, JobDocVersion jobDocVersion) throws IOException;
 }

src/main/java/com/amazon/opendistroforelasticsearch/jobscheduler/sweeper/JobSweeper.java

@@ -214,18 +214,18 @@ public void postDelete(ShardId shardId, Engine.Delete delete, Engine.DeleteResul
     }
 
     @VisibleForTesting
-    void sweep(ShardId shardId, String docId, BytesReference jobSource, JobDocVersion version) {
+    void sweep(ShardId shardId, String docId, BytesReference jobSource, JobDocVersion jobDocVersion) {
         ConcurrentHashMap<String, JobDocVersion> jobVersionMap;
         if (this.sweptJobs.containsKey(shardId)) {
             jobVersionMap = this.sweptJobs.get(shardId);
         } else {
             jobVersionMap = new ConcurrentHashMap<>();
             this.sweptJobs.put(shardId, jobVersionMap);
         }
-        jobVersionMap.compute(docId, (id, currentVersion) -> {
-            if (version.compareTo(currentVersion) <= 0) {
-                log.debug("Skipping job {}, new version {} <= current version {}", docId, version, currentVersion);
-                return currentVersion;
+        jobVersionMap.compute(docId, (id, currentJobDocVersion) -> {
+            if (jobDocVersion.compareTo(currentJobDocVersion) <= 0) {
+                log.debug("Skipping job {}, new version {} <= current version {}", docId, jobDocVersion, currentJobDocVersion);
+                return currentJobDocVersion;
             }
 
             if (this.scheduler.getScheduledJobIds(shardId.getIndexName()).contains(docId)) {
@@ -236,20 +236,20 @@ void sweep(ShardId shardId, String docId, BytesReference jobSource, JobDocVersio
                     ScheduledJobProvider provider = this.indexToProviders.get(shardId.getIndexName());
                     XContentParser parser = XContentHelper.createParser(this.xContentRegistry, LoggingDeprecationHandler.INSTANCE,
                             jobSource, XContentType.JSON);
-                    ScheduledJobParameter jobParameter = provider.getJobParser().parse(parser, docId, version.getVersion());
+                    ScheduledJobParameter jobParameter = provider.getJobParser().parse(parser, docId, jobDocVersion);
                     if (jobParameter == null) {
                         // allow parser to return null, which means this is not a scheduled job document.
                         return null;
                     }
                     ScheduledJobRunner jobRunner = this.indexToProviders.get(shardId.getIndexName()).getJobRunner();
                     if (jobParameter.isEnabled()) {
-                        this.scheduler.schedule(shardId.getIndexName(), docId, jobParameter, jobRunner, version);
+                        this.scheduler.schedule(shardId.getIndexName(), docId, jobParameter, jobRunner, jobDocVersion);
                     }
-                    return version;
+                    return jobDocVersion;
                 } catch (Exception e) {
                     log.warn("Unable to parse job, error message: {} , message source: {}", e.getMessage(),
                             Strings.cleanTruncate(jobSource.utf8ToString(), 1000));
-                    return currentVersion;
+                    return currentJobDocVersion;
                 }
             } else {
                 return null;

src/test/java/com/amazon/opendistroforelasticsearch/jobscheduler/sweeper/JobSweeperTests.java

@@ -261,7 +261,8 @@ public void testSweep() throws IOException {
 
         ScheduledJobParameter mockJobParameter = Mockito.mock(ScheduledJobParameter.class);
         Mockito.when(mockJobParameter.isEnabled()).thenReturn(true);
-        Mockito.when(this.jobParser.parse(Mockito.any(), Mockito.anyString(), Mockito.anyLong())).thenReturn(mockJobParameter);
+        Mockito.when(this.jobParser.parse(Mockito.any(), Mockito.anyString(), Mockito.any(JobDocVersion.class)))
+            .thenReturn(mockJobParameter);
 
         this.sweeper.sweep(shardId, "id", this.getTestJsonSource(), new JobDocVersion(1L, 1L, 2L));
         Mockito.verify(this.scheduler).schedule(Mockito.anyString(), Mockito.anyString(), Mockito.any(), Mockito.any(),