Skip to content

Commit

Permalink
update operation job details without jobtracker
Browse files Browse the repository at this point in the history
  • Loading branch information
ms9698 committed Nov 11, 2024
1 parent 906e3fa commit 93a3a3d
Show file tree
Hide file tree
Showing 2 changed files with 26 additions and 1 deletion.
25 changes: 25 additions & 0 deletions core/store/src/main/java/uk/gov/gchq/gaffer/store/Store.java
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@

import uk.gov.gchq.gaffer.cache.Cache;
import uk.gov.gchq.gaffer.cache.CacheServiceLoader;
import uk.gov.gchq.gaffer.cache.exception.CacheOperationException;
import uk.gov.gchq.gaffer.commonutil.CloseableUtil;
import uk.gov.gchq.gaffer.commonutil.ExecutorService;
import uk.gov.gchq.gaffer.core.exception.GafferRuntimeException;
Expand Down Expand Up @@ -186,7 +187,10 @@
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.StreamSupport;

import static java.util.Collections.unmodifiableList;
Expand Down Expand Up @@ -384,8 +388,29 @@ public void execute(final Operation operation, final Context context) throws Ope
public <O> O execute(final Output<O> operation, final Context context) throws OperationException {
return execute(OperationChain.wrap(operation), context);
}
private final ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>());

private void validateJobDetail(final JobDetail jobDetail) {
if (null == jobDetail) {
throw new IllegalArgumentException("JobDetail is required");
}

if (null == jobDetail.getJobId() || jobDetail.getJobId().isEmpty()) {
throw new IllegalArgumentException("jobId is required");
}
}

protected <O> O execute(final OperationChain<O> operation, final Context context) throws OperationException {
final JobDetail newJobDetail = new JobDetail(context.getJobId(), context.getUser(), operation, JobStatus.RUNNING, null);
final O result = (O) handleOperation(operation, context);
validateJobDetail(newJobDetail);
executor.submit(() -> {
try {
super.addToCache(newJobDetail.getJobId(), newJobDetail, true);
} catch (final CacheOperationException e) {
LOGGER.error("Failed to add jobDetail " + newJobDetail.toString() + " to the cache", e);
}
});
return result;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ public void shouldReturnSameJobIdInHeaderAsGetAllJobDetailsOperation() throws IO

// When
final Response response = client.executeOperation(new GetAllJobDetails());

// System.out.println(response.readEntity(String.class));
// Then
assertTrue(response.readEntity(String.class).contains(response.getHeaderString("job-id")));
}
Expand Down

0 comments on commit 93a3a3d

Please sign in to comment.