Merge pull request #5334 from adamnsch/jobid-in-configs

Add `jobId` field to algo and project configs

Author: adamnsch

algo-common/src/main/java/org/neo4j/gds/AlgorithmFactory.java

@@ -60,6 +60,7 @@ default ALGO build(
             progressTask,
             log,
             configuration.concurrency(),
+            configuration.jobId(),
             taskRegistryFactory,
             userLogRegistryFactory
         );

alpha/alpha-proc/src/test/java/org/neo4j/gds/KSpanningTreeProcTest.java

@@ -148,7 +148,7 @@ void shouldTrackProgress() {
         TestProcedureRunner.applyOnProcedure(db, KSpanningTreeMaxProc.class, proc -> {
             var taskStore = new GlobalTaskStore();
 
-            proc.taskRegistryFactory = () -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore));
+            proc.taskRegistryFactory = jobId -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore, jobId));
             proc.nodePropertyExporterBuilder = new NativeNodePropertiesExporterBuilder(
                 TransactionContext.of(proc.api, proc.procedureTransaction)
             );

alpha/alpha-proc/src/test/java/org/neo4j/gds/SpanningTreeProcTest.java

@@ -215,7 +215,7 @@ void shouldTrackProgress() {
         TestProcedureRunner.applyOnProcedure(db, SpanningTreeProcMin.class, proc -> {
             var taskStore = new GlobalTaskStore();
 
-            proc.taskRegistryFactory = () -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore));
+            proc.taskRegistryFactory = jobId -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore, jobId));
             proc.relationshipExporterBuilder = new NativeRelationshipExporterBuilder(
                 TransactionContext.of(proc.api, proc.procedureTransaction)
             );

alpha/alpha-proc/src/test/java/org/neo4j/gds/centrality/HarmonicCentralityProcTest.java

@@ -115,7 +115,7 @@ void testProgressTracking() {
         TestProcedureRunner.applyOnProcedure(db, HarmonicCentralityWriteProc.class, proc -> {
             var taskStore = new GlobalTaskStore();
 
-            proc.taskRegistryFactory = () -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore));
+            proc.taskRegistryFactory = jobId -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore, jobId));
             proc.nodePropertyExporterBuilder = new NativeNodePropertiesExporterBuilder(
                 TransactionContext.of(proc.api, proc.procedureTransaction)
             );

alpha/alpha-proc/src/test/java/org/neo4j/gds/scc/SccProcTest.java

@@ -147,7 +147,7 @@ void testProgressTracking() {
         TestProcedureRunner.applyOnProcedure(db, SccWriteProc.class, proc -> {
             var taskStore = new GlobalTaskStore();
 
-            proc.taskRegistryFactory = () -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore));
+            proc.taskRegistryFactory = jobId -> new NonReleasingTaskRegistry(new TaskRegistry(getUsername(), taskStore, jobId));
             proc.nodePropertyExporterBuilder = new NativeNodePropertiesExporterBuilder(
                 TransactionContext.of(proc.api, proc.procedureTransaction)
             );

alpha/alpha-proc/src/test/java/org/neo4j/gds/userlog/UserLogProcTest.java

@@ -27,6 +27,7 @@
 import org.neo4j.gds.GdsCypher;
 import org.neo4j.gds.catalog.GraphProjectProc;
 import org.neo4j.gds.compat.Neo4jProxy;
+import org.neo4j.gds.core.utils.progress.JobId;
 import org.neo4j.gds.core.utils.progress.ProgressFeatureSettings;
 import org.neo4j.gds.core.utils.progress.TaskRegistryExtension;
 import org.neo4j.gds.core.utils.progress.TaskRegistryFactory;
@@ -217,7 +218,7 @@ public Stream<FakeResult> foo(
         ) throws InterruptedException {
             var task = Tasks.task(taskName, Tasks.leaf("leaf", 3));
 
-            var taskProgressTracker = new TaskProgressTracker(task, Neo4jProxy.testLog(), 1, taskRegistryFactory,
+            var taskProgressTracker = new TaskProgressTracker(task, Neo4jProxy.testLog(), 1, new JobId(), taskRegistryFactory,
                 userLogRegistryFactory
             );
             taskProgressTracker.beginSubTask();

alpha/alpha-proc/src/test/java/org/neo4j/gds/walking/CollapsePathMutateProcTest.java

@@ -157,4 +157,8 @@ void testMutateYields() {
     @Disabled
     @Override
     public void shouldUnregisterTaskAfterComputation() {}
+
+    @Disabled
+    @Override
+    public void shouldRegisterTaskWithCorrectJobId() {}
 }

core/src/main/java/org/neo4j/gds/config/AlgoBaseConfig.java

@@ -31,7 +31,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
-public interface AlgoBaseConfig extends BaseConfig, ConcurrencyConfig {
+public interface AlgoBaseConfig extends BaseConfig, ConcurrencyConfig, JobIdConfig {
 
     String NODE_LABELS_KEY = "nodeLabels";
     String RELATIONSHIP_TYPES_KEY = "relationshipTypes";

core/src/main/java/org/neo4j/gds/config/GraphProjectConfig.java

@@ -36,7 +36,7 @@
 import static org.neo4j.gds.config.GraphProjectFromStoreConfig.NODE_PROJECTION_KEY;
 import static org.neo4j.gds.config.GraphProjectFromStoreConfig.RELATIONSHIP_PROJECTION_KEY;
 
-public interface GraphProjectConfig extends BaseConfig {
+public interface GraphProjectConfig extends BaseConfig, JobIdConfig {
 
     String IMPLICIT_GRAPH_NAME = "";
     String NODE_COUNT_KEY = "nodeCount";

core/src/main/java/org/neo4j/gds/config/JobIdConfig.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.gds.config;
+
+import org.immutables.value.Value;
+import org.neo4j.gds.annotation.Configuration;
+import org.neo4j.gds.core.utils.progress.JobId;
+
+@Configuration
+public interface JobIdConfig {
+    @Value.Default
+    @Value.Parameter(false)
+    @Configuration.ConvertWith("org.neo4j.gds.core.utils.progress.JobId#parse")
+    @Configuration.ToMapValue("org.neo4j.gds.core.utils.progress.JobId#asString")
+    default JobId jobId() {
+        return new JobId();
+    }
+}