This is an automated email from the ASF dual-hosted git repository.
mweiler pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie-kogito-apps.git
The following commit(s) were added to refs/heads/main by this push:
new 215f9a682 [Incubator kie issues#2177] Update existing job instances
instead of inserting/replacing in data index (#2282)
215f9a682 is described below
commit 215f9a68248a3242e4f23aef25bc2707da152429
Author: Abhiram Gundala <[email protected]>
AuthorDate: Thu Dec 11 17:49:29 2025 -0500
[Incubator kie issues#2177] Update existing job instances instead of
inserting/replacing in data index (#2282)
* added header
* update job details fix
* updated existing job instances
---
.../kie/kogito/index/service/IndexingService.java | 2 +-
.../index/storage/DataIndexStorageService.java | 3 +-
.../kogito/index/storage/JobInstanceStorage.java} | 22 ++--------
.../storage/ModelDataIndexStorageService.java | 4 +-
.../index/storage/ModelJobInstanceStorage.java} | 45 +++++++++++++--------
.../jpa/storage/JPADataIndexStorageService.java | 4 +-
.../kogito/index/jpa/storage/JobEntityStorage.java | 47 +++++++++++++++++++++-
7 files changed, 84 insertions(+), 43 deletions(-)
diff --git
a/data-index/data-index-common/src/main/java/org/kie/kogito/index/service/IndexingService.java
b/data-index/data-index-common/src/main/java/org/kie/kogito/index/service/IndexingService.java
index c4b468501..7993d4762 100644
---
a/data-index/data-index-common/src/main/java/org/kie/kogito/index/service/IndexingService.java
+++
b/data-index/data-index-common/src/main/java/org/kie/kogito/index/service/IndexingService.java
@@ -118,7 +118,7 @@ public class IndexingService {
}
public void indexJob(Job job) {
- manager.getJobsStorage().put(job.getId(), job);
+ manager.getJobsStorage().indexJob(job);
}
public void indexModel(ObjectNode updateData) {
diff --git
a/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
b/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
index 10e2d9195..4798020ae 100644
---
a/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
+++
b/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
@@ -21,7 +21,6 @@ package org.kie.kogito.index.storage;
import java.util.EnumSet;
import java.util.Set;
-import org.kie.kogito.index.model.Job;
import org.kie.kogito.index.model.ProcessDefinition;
import org.kie.kogito.index.model.ProcessDefinitionKey;
import org.kie.kogito.persistence.api.Storage;
@@ -36,7 +35,7 @@ public interface DataIndexStorageService {
UserTaskInstanceStorage getUserTaskInstanceStorage();
- Storage<String, Job> getJobsStorage();
+ JobInstanceStorage getJobsStorage();
Storage<String, ObjectNode> getDomainModelCache(String processId);
diff --git
a/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
b/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/JobInstanceStorage.java
similarity index 55%
copy from
data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
copy to
data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/JobInstanceStorage.java
index f9ad19520..61e1ebad4 100644
---
a/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
+++
b/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/JobInstanceStorage.java
@@ -16,25 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.index.jpa.storage;
+package org.kie.kogito.index.storage;
-import org.kie.kogito.index.jpa.mapper.JobEntityMapper;
-import org.kie.kogito.index.jpa.model.AbstractEntity;
-import org.kie.kogito.index.jpa.model.JobEntity;
import org.kie.kogito.index.model.Job;
+import org.kie.kogito.persistence.api.Storage;
-import jakarta.enterprise.context.ApplicationScoped;
-import jakarta.inject.Inject;
-import jakarta.persistence.EntityManager;
-
-@ApplicationScoped
-public class JobEntityStorage extends AbstractStorage<String, JobEntity, Job> {
-
- protected JobEntityStorage() {
- }
-
- @Inject
- public JobEntityStorage(EntityManager em) {
- super(em, Job.class, JobEntity.class,
JobEntityMapper.INSTANCE::mapToModel, JobEntityMapper.INSTANCE::mapToEntity,
AbstractEntity::getId);
- }
+public interface JobInstanceStorage extends Storage<String, Job> {
+ void indexJob(Job job);
}
diff --git
a/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelDataIndexStorageService.java
b/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelDataIndexStorageService.java
index 693928d99..47e435447 100644
---
a/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelDataIndexStorageService.java
+++
b/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelDataIndexStorageService.java
@@ -59,8 +59,8 @@ public class ModelDataIndexStorageService implements
DataIndexStorageService {
}
@Override
- public Storage<String, Job> getJobsStorage() {
- return storageService.getCache(JOBS_STORAGE, Job.class);
+ public JobInstanceStorage getJobsStorage() {
+ return new
ModelJobInstanceStorage(storageService.getCache(JOBS_STORAGE, Job.class));
}
@Override
diff --git
a/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
b/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelJobInstanceStorage.java
similarity index 52%
copy from
data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
copy to
data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelJobInstanceStorage.java
index 10e2d9195..6e57164e3 100644
---
a/data-index/data-index-storage/data-index-storage-api/src/main/java/org/kie/kogito/index/storage/DataIndexStorageService.java
+++
b/data-index/data-index-storage/data-index-storage-common/src/main/java/org/kie/kogito/index/storage/ModelJobInstanceStorage.java
@@ -18,33 +18,44 @@
*/
package org.kie.kogito.index.storage;
-import java.util.EnumSet;
-import java.util.Set;
+import java.util.Map;
import org.kie.kogito.index.model.Job;
-import org.kie.kogito.index.model.ProcessDefinition;
-import org.kie.kogito.index.model.ProcessDefinitionKey;
import org.kie.kogito.persistence.api.Storage;
-import org.kie.kogito.persistence.api.StorageServiceCapability;
-import com.fasterxml.jackson.databind.node.ObjectNode;
+public class ModelJobInstanceStorage extends ModelStorageFetcher<String, Job>
implements JobInstanceStorage {
-public interface DataIndexStorageService {
- Storage<ProcessDefinitionKey, ProcessDefinition>
getProcessDefinitionStorage();
-
- ProcessInstanceStorage getProcessInstanceStorage();
+ public ModelJobInstanceStorage(Storage<String, Job> storage) {
+ super(storage, key -> key, key -> key);
+ }
- UserTaskInstanceStorage getUserTaskInstanceStorage();
+ @Override
+ public void indexJob(Job job) {
+ put(job.getId(), job);
+ }
- Storage<String, Job> getJobsStorage();
+ @Override
+ public Job put(String key, Job value) {
+ return storage.put(key, value);
+ }
- Storage<String, ObjectNode> getDomainModelCache(String processId);
+ @Override
+ public Job remove(String key) {
+ return storage.remove(key);
+ }
- String getDomainModelCacheName(String processId);
+ @Override
+ public boolean containsKey(String key) {
+ return storage.containsKey(key);
+ }
- Storage<String, String> getProcessIdModelCache();
+ @Override
+ public Map<String, Job> entries() {
+ return storage.entries();
+ }
- default Set<StorageServiceCapability> capabilities() {
- return EnumSet.noneOf(StorageServiceCapability.class);
+ @Override
+ public String getRootType() {
+ return Job.class.getName();
}
}
diff --git
a/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JPADataIndexStorageService.java
b/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JPADataIndexStorageService.java
index bf0cf1a1c..3481147ad 100644
---
a/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JPADataIndexStorageService.java
+++
b/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JPADataIndexStorageService.java
@@ -20,10 +20,10 @@ package org.kie.kogito.index.jpa.storage;
import java.util.Set;
-import org.kie.kogito.index.model.Job;
import org.kie.kogito.index.model.ProcessDefinition;
import org.kie.kogito.index.model.ProcessDefinitionKey;
import org.kie.kogito.index.storage.DataIndexStorageService;
+import org.kie.kogito.index.storage.JobInstanceStorage;
import org.kie.kogito.index.storage.ProcessInstanceStorage;
import org.kie.kogito.index.storage.UserTaskInstanceStorage;
import org.kie.kogito.persistence.api.Storage;
@@ -70,7 +70,7 @@ public class JPADataIndexStorageService implements
DataIndexStorageService {
}
@Override
- public Storage<String, Job> getJobsStorage() {
+ public JobInstanceStorage getJobsStorage() {
return jobsStorage;
}
diff --git
a/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
b/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
index f9ad19520..8e767c679 100644
---
a/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
+++
b/data-index/data-index-storage/data-index-storage-jpa-common/src/main/java/org/kie/kogito/index/jpa/storage/JobEntityStorage.java
@@ -22,13 +22,15 @@ import org.kie.kogito.index.jpa.mapper.JobEntityMapper;
import org.kie.kogito.index.jpa.model.AbstractEntity;
import org.kie.kogito.index.jpa.model.JobEntity;
import org.kie.kogito.index.model.Job;
+import org.kie.kogito.index.storage.JobInstanceStorage;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import jakarta.persistence.EntityManager;
+import jakarta.transaction.Transactional;
@ApplicationScoped
-public class JobEntityStorage extends AbstractStorage<String, JobEntity, Job> {
+public class JobEntityStorage extends AbstractStorage<String, JobEntity, Job>
implements JobInstanceStorage {
protected JobEntityStorage() {
}
@@ -37,4 +39,47 @@ public class JobEntityStorage extends
AbstractStorage<String, JobEntity, Job> {
public JobEntityStorage(EntityManager em) {
super(em, Job.class, JobEntity.class,
JobEntityMapper.INSTANCE::mapToModel, JobEntityMapper.INSTANCE::mapToEntity,
AbstractEntity::getId);
}
+
+ @Transactional
+ @Override
+ public void indexJob(Job job) {
+ JobEntity entity = findOrInit(job.getId(), job);
+ updateJobEntity(entity, job);
+ }
+
+ private JobEntity findOrInit(String jobId, Job job) {
+ JobEntity entity = em.find(JobEntity.class, jobId);
+ if (entity == null) {
+ entity = new JobEntity();
+ entity.setId(jobId);
+ entity.setProcessId(job.getProcessId());
+ entity.setRootProcessId(job.getRootProcessId());
+ em.persist(entity);
+ } else {
+ if (entity.getProcessId() == null) {
+ entity.setProcessId(job.getProcessId());
+ }
+ if (entity.getRootProcessId() == null) {
+ entity.setRootProcessId(job.getRootProcessId());
+ }
+ }
+ return entity;
+ }
+
+ private void updateJobEntity(JobEntity entity, Job job) {
+ entity.setProcessInstanceId(job.getProcessInstanceId());
+ entity.setNodeInstanceId(job.getNodeInstanceId());
+ entity.setRootProcessInstanceId(job.getRootProcessInstanceId());
+ entity.setExpirationTime(job.getExpirationTime());
+ entity.setPriority(job.getPriority());
+ entity.setCallbackEndpoint(job.getCallbackEndpoint());
+ entity.setRepeatInterval(job.getRepeatInterval());
+ entity.setRepeatLimit(job.getRepeatLimit());
+ entity.setScheduledId(job.getScheduledId());
+ entity.setRetries(job.getRetries());
+ entity.setStatus(job.getStatus());
+ entity.setLastUpdate(job.getLastUpdate());
+ entity.setExecutionCounter(job.getExecutionCounter());
+ entity.setEndpoint(job.getEndpoint());
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
