TERMINAL_STATE =
+ Collections.unmodifiableList(Arrays.asList(COMPLETED, ABORTED, ERROR));
/**
* Get a collection of terminal job state.
*
- * Terminal job state is final and will not change to any other state.
+ * Terminal job state is final and will not change to any other state.
*
* @return collection of terminal job state.
*/
- public static Collection getTerminalState(){
+ public static Collection getTerminalState() {
return TERMINAL_STATE;
}
}
diff --git a/core/src/main/java/feast/core/service/JobManagementService.java b/core/src/main/java/feast/core/service/JobManagementService.java
index 01257e33a7..02fa40b280 100644
--- a/core/src/main/java/feast/core/service/JobManagementService.java
+++ b/core/src/main/java/feast/core/service/JobManagementService.java
@@ -19,39 +19,50 @@
import com.google.common.base.Strings;
import feast.core.JobServiceProto.JobServiceTypes.JobDetail;
+import feast.core.config.ImportJobDefaults;
import feast.core.dao.JobInfoRepository;
import feast.core.dao.MetricsRepository;
+import feast.core.exception.JobExecutionException;
import feast.core.exception.RetrievalException;
import feast.core.job.JobManager;
+import feast.core.job.Runner;
import feast.core.log.Action;
import feast.core.log.AuditLogger;
import feast.core.log.Resource;
import feast.core.model.JobInfo;
import feast.core.model.JobStatus;
import feast.core.model.Metrics;
+import feast.specs.ImportSpecProto.ImportSpec;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
-
@Slf4j
@Service
public class JobManagementService {
- @Autowired private JobInfoRepository jobInfoRepository;
- @Autowired private MetricsRepository metricsRepository;
- @Autowired private JobManager jobManager;
+ private static final String JOB_PREFIX_DEFAULT = "feastimport";
+ private static final String UNKNOWN_EXT_JOB_ID = "";
+
+ private JobInfoRepository jobInfoRepository;
+ private MetricsRepository metricsRepository;
+ private JobManager jobManager;
+ private ImportJobDefaults defaults;
+ @Autowired
public JobManagementService(
JobInfoRepository jobInfoRepository,
MetricsRepository metricsRepository,
- JobManager jobManager) {
+ JobManager jobManager,
+ ImportJobDefaults defaults) {
this.jobInfoRepository = jobInfoRepository;
this.metricsRepository = metricsRepository;
this.jobManager = jobManager;
+ this.defaults = defaults;
}
/**
@@ -85,6 +96,64 @@ public JobDetail getJob(String id) {
return jobDetailBuilder.build();
}
+ /**
+ * Submit ingestion job to runner.
+ *
+ * @param importSpec import spec of the ingestion job
+ * @param namePrefix name prefix of the ingestion job
+ * @return feast job ID.
+ */
+ public String submitJob(ImportSpec importSpec, String namePrefix) {
+ String jobId = createJobId(namePrefix);
+ boolean isDirectRunner = Runner.DIRECT.getName().equals(defaults.getRunner());
+ try {
+ if (!isDirectRunner) {
+ JobInfo jobInfo =
+ new JobInfo(jobId, UNKNOWN_EXT_JOB_ID, defaults.getRunner(), importSpec, JobStatus.PENDING);
+ jobInfoRepository.save(jobInfo);
+ }
+
+ AuditLogger.log(
+ Resource.JOB,
+ jobId,
+ Action.SUBMIT,
+ "Building graph and submitting to %s",
+ defaults.getRunner());
+
+ String extId = jobManager.submitJob(importSpec, jobId);
+ if (extId.isEmpty()) {
+ throw new RuntimeException(
+ String.format("Could not submit job: \n%s", "unable to retrieve job external id"));
+ }
+
+ AuditLogger.log(
+ Resource.JOB,
+ jobId,
+ Action.STATUS_CHANGE,
+ "Job submitted to runner %s with ext id %s.",
+ defaults.getRunner(),
+ extId);
+
+ if (isDirectRunner) {
+ JobInfo jobInfo =
+ new JobInfo(jobId, extId, defaults.getRunner(), importSpec, JobStatus.COMPLETED);
+ jobInfoRepository.save(jobInfo);
+ } else {
+ updateJobExtId(jobId, extId);
+ }
+ return jobId;
+ } catch (Exception e) {
+ updateJobStatus(jobId, JobStatus.ERROR);
+ AuditLogger.log(
+ Resource.JOB,
+ jobId,
+ Action.STATUS_CHANGE,
+ "Job failed to be submitted to runner %s. Job status changed to ERROR.",
+ defaults.getRunner());
+ throw new JobExecutionException(String.format("Error running ingestion job: %s", e), e);
+ }
+ }
+
/**
* Drain the given job. If this is successful, the job will start the draining process. When the
* draining process is complete, the job will be cleaned up and removed.
@@ -108,4 +177,39 @@ public void abortJob(String id) {
AuditLogger.log(Resource.JOB, id, Action.ABORT, "Triggering draining of job");
jobInfoRepository.saveAndFlush(job);
}
+
+ /**
+ * Update a given job's status
+ *
+ * @param jobId
+ * @param status
+ */
+ void updateJobStatus(String jobId, JobStatus status) {
+ Optional jobRecordOptional = jobInfoRepository.findById(jobId);
+ if (jobRecordOptional.isPresent()) {
+ JobInfo jobRecord = jobRecordOptional.get();
+ jobRecord.setStatus(status);
+ jobInfoRepository.save(jobRecord);
+ }
+ }
+
+ /**
+ * Update a given job's external id
+ *
+ * @param jobId
+ * @param jobExtId
+ */
+ void updateJobExtId(String jobId, String jobExtId) {
+ Optional jobRecordOptional = jobInfoRepository.findById(jobId);
+ if (jobRecordOptional.isPresent()) {
+ JobInfo jobRecord = jobRecordOptional.get();
+ jobRecord.setExtId(jobExtId);
+ jobInfoRepository.save(jobRecord);
+ }
+ }
+
+ private String createJobId(String namePrefix) {
+ String dateSuffix = String.valueOf(Instant.now().toEpochMilli());
+ return namePrefix.isEmpty() ? JOB_PREFIX_DEFAULT + dateSuffix : namePrefix + dateSuffix;
+ }
}
diff --git a/core/src/main/resources/application.properties b/core/src/main/resources/application.properties
index ac8fb3534f..ef95352d24 100644
--- a/core/src/main/resources/application.properties
+++ b/core/src/main/resources/application.properties
@@ -26,6 +26,9 @@ feast.jobs.errorsStoreOptions=${JOB_ERRORS_STORE_OPTIONS:{}}
feast.jobs.dataflow.projectId = ${DATAFLOW_PROJECT_ID:}
feast.jobs.dataflow.location = ${DATAFLOW_LOCATION:}
+feast.jobs.flink.configDir = ${FLINK_CONF_DIR:/etc/flink/flink-1.5.5/conf}
+feast.jobs.flink.masterUrl = ${FLINK_MASTER_URL:localhost:8081}
+
feast.jobs.monitor.period = ${JOB_MONITOR_PERIOD_MS:5000}
feast.jobs.monitor.initialDelay = ${JOB_MONITOR_INITIAL_DELAY_MS:60000}
diff --git a/core/src/test/java/feast/core/job/ScheduledJobMonitorTest.java b/core/src/test/java/feast/core/job/ScheduledJobMonitorTest.java
index a7ee2bec70..d352044d16 100644
--- a/core/src/test/java/feast/core/job/ScheduledJobMonitorTest.java
+++ b/core/src/test/java/feast/core/job/ScheduledJobMonitorTest.java
@@ -20,10 +20,10 @@
import static org.hamcrest.core.IsEqual.equalTo;
import static org.junit.Assert.assertThat;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
-import com.google.common.collect.Lists;
import feast.core.dao.JobInfoRepository;
import feast.core.dao.MetricsRepository;
import feast.core.model.JobInfo;
@@ -39,22 +39,17 @@
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
-
public class ScheduledJobMonitorTest {
ScheduledJobMonitor scheduledJobMonitor;
- @Mock
- JobMonitor jobMonitor;
+ @Mock JobMonitor jobMonitor;
- @Mock
- StatsdMetricPusher stasdMetricPusher;
+ @Mock StatsdMetricPusher stasdMetricPusher;
- @Mock
- JobInfoRepository jobInfoRepository;
+ @Mock JobInfoRepository jobInfoRepository;
- @Mock
- MetricsRepository metricsRepository;
+ @Mock MetricsRepository metricsRepository;
@Before
public void setUp() throws Exception {
@@ -64,65 +59,71 @@ public void setUp() throws Exception {
@Test
public void getJobStatus_shouldUpdateJobInfoForRunningJob() {
- JobInfo job = new JobInfo("jobId", "extId1", "Streaming", "DataflowRunner", "",
- Collections.emptyList(),
- Collections.emptyList(),
- Collections.emptyList(),
- JobStatus.RUNNING,
- "");
-
- when(jobInfoRepository.findByStatusNotIn((Collection)any(Collection.class))).thenReturn(
- Arrays.asList(job));
- when(jobMonitor.getJobStatus("extId1")).thenReturn(JobStatus.COMPLETED);
+ JobInfo job =
+ new JobInfo(
+ "jobId",
+ "extId1",
+ "Streaming",
+ "DataflowRunner",
+ "",
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ JobStatus.RUNNING,
+ "");
+
+ when(jobInfoRepository.findByStatusNotIn((Collection) any(Collection.class)))
+ .thenReturn(Collections.singletonList(job));
+ when(jobMonitor.getJobStatus(job)).thenReturn(JobStatus.COMPLETED);
scheduledJobMonitor.getJobStatus();
- ArgumentCaptor> argCaptor = ArgumentCaptor.forClass(Iterable.class);
- verify(jobInfoRepository).saveAll(argCaptor.capture());
+ ArgumentCaptor argCaptor = ArgumentCaptor.forClass(JobInfo.class);
+ verify(jobInfoRepository).save(argCaptor.capture());
- List jobInfos = Lists.newArrayList(argCaptor.getValue());
- assertThat(jobInfos.size(), equalTo(1));
- assertThat(jobInfos.get(0).getStatus(), equalTo(JobStatus.COMPLETED));
+ JobInfo jobInfos = argCaptor.getValue();
+ assertThat(jobInfos.getStatus(), equalTo(JobStatus.COMPLETED));
}
@Test
public void getJobStatus_shouldNotUpdateJobInfoForTerminalJob() {
- when(jobInfoRepository.findByStatusNotIn((Collection)any(Collection.class))).thenReturn(
- Collections.emptyList());
+ when(jobInfoRepository.findByStatusNotIn((Collection) any(Collection.class)))
+ .thenReturn(Collections.emptyList());
scheduledJobMonitor.getJobStatus();
- ArgumentCaptor> argCaptor = ArgumentCaptor.forClass(Iterable.class);
- verify(jobInfoRepository).saveAll(argCaptor.capture());
-
- List jobInfos = Lists.newArrayList(argCaptor.getValue());
- assertThat(jobInfos.size(), equalTo(0));
+ verify(jobInfoRepository, never()).save(any(JobInfo.class));
}
@Test
public void getJobMetrics_shouldPushToStatsDMetricPusherAndSaveNewMetricToDb() {
- JobInfo job = new JobInfo("jobId", "extId1", "Streaming", "DataflowRunner", "",
- Collections.emptyList(),
- Collections.emptyList(),
- Collections.emptyList(),
- JobStatus.RUNNING,
- "");
+ JobInfo job =
+ new JobInfo(
+ "jobId",
+ "extId1",
+ "Streaming",
+ "DataflowRunner",
+ "",
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ JobStatus.RUNNING,
+ "");
Metrics metric1 = new Metrics(job, "metric1", 1);
Metrics metric2 = new Metrics(job, "metric2", 2);
List metrics = Arrays.asList(metric1, metric2);
- when(jobInfoRepository.findByStatusNotIn((Collection)any(Collection.class))).thenReturn(
- Arrays.asList(job));
+ when(jobInfoRepository.findByStatusNotIn((Collection) any(Collection.class)))
+ .thenReturn(Arrays.asList(job));
when(jobMonitor.getJobMetrics(job)).thenReturn(metrics);
scheduledJobMonitor.getJobMetrics();
verify(stasdMetricPusher).pushMetrics(metrics);
- ArgumentCaptor> argCaptor = ArgumentCaptor.forClass(Iterable.class);
- verify(jobInfoRepository).saveAll(argCaptor.capture());
+ ArgumentCaptor argCaptor = ArgumentCaptor.forClass(JobInfo.class);
+ verify(jobInfoRepository).save(argCaptor.capture());
assertThat(job.getMetrics(), equalTo(metrics));
}
-
-}
\ No newline at end of file
+}
diff --git a/core/src/test/java/feast/core/service/JobExecutionServiceTest.java b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java
similarity index 59%
rename from core/src/test/java/feast/core/service/JobExecutionServiceTest.java
rename to core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java
index ad29bb8b9e..678b76c104 100644
--- a/core/src/test/java/feast/core/service/JobExecutionServiceTest.java
+++ b/core/src/test/java/feast/core/job/dataflow/DataflowJobManagerTest.java
@@ -15,42 +15,37 @@
*
*/
-package feast.core.service;
+package feast.core.job.dataflow;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.equalTo;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.mockito.MockitoAnnotations.initMocks;
-import static org.mockito.internal.verification.VerificationModeFactory.times;
+import com.google.api.services.dataflow.Dataflow;
import com.google.common.collect.Lists;
import feast.core.config.ImportJobDefaults;
-import feast.core.dao.JobInfoRepository;
-import feast.core.model.JobInfo;
-import feast.core.model.JobStatus;
import feast.specs.ImportSpecProto.ImportSpec;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.List;
-import java.util.Optional;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
-import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.Mockito;
-public class JobExecutionServiceTest {
+public class DataflowJobManagerTest {
+
+ @Rule public final ExpectedException expectedException = ExpectedException.none();
+
+ @Mock Dataflow dataflow;
- @Rule
- public final ExpectedException expectedException = ExpectedException.none();
- @Mock
- JobInfoRepository jobInfoRepository;
private ImportJobDefaults defaults;
+ private DataflowJobManager dfJobManager;
@Before
public void setUp() {
@@ -58,25 +53,25 @@ public void setUp() {
defaults =
new ImportJobDefaults(
"localhost:8080",
- "DirectRunner",
+ "DataflowRunner",
"{\"key\":\"value\"}",
"ingestion.jar",
"STDOUT",
"{}");
+ dfJobManager = new DataflowJobManager(dataflow, "project", "location", defaults);
}
@Test
public void shouldBuildProcessBuilderWithCorrectOptions() {
- JobExecutionService jobExecutionService = new JobExecutionService(jobInfoRepository, defaults);
ImportSpec importSpec = ImportSpec.newBuilder().setType("file").build();
- ProcessBuilder pb = jobExecutionService.getProcessBuilder(importSpec, "test");
+ ProcessBuilder pb = dfJobManager.getProcessBuilder(importSpec, "test");
List expected =
Lists.newArrayList(
"java",
"-jar",
"ingestion.jar",
"--jobName=test",
- "--runner=DirectRunner",
+ "--runner=DataflowRunner",
"--importSpecBase64=CgRmaWxl",
"--coreApiUri=localhost:8080",
"--errorsStoreType=STDOUT",
@@ -85,38 +80,6 @@ public void shouldBuildProcessBuilderWithCorrectOptions() {
assertThat(pb.command(), equalTo(expected));
}
- @Test
- public void shouldUpdateJobStatusIfExists() {
- JobInfo jobInfo = new JobInfo();
- when(jobInfoRepository.findById("jobid")).thenReturn(Optional.of(jobInfo));
-
- ArgumentCaptor jobInfoArgumentCaptor = ArgumentCaptor.forClass(JobInfo.class);
- JobExecutionService jobExecutionService = new JobExecutionService(jobInfoRepository, defaults);
- jobExecutionService.updateJobStatus("jobid", JobStatus.PENDING);
-
- verify(jobInfoRepository, times(1)).saveAndFlush(jobInfoArgumentCaptor.capture());
-
- JobInfo jobInfoUpdated = new JobInfo();
- jobInfoUpdated.setStatus(JobStatus.PENDING);
- assertThat(jobInfoArgumentCaptor.getValue(), equalTo(jobInfoUpdated));
- }
-
- @Test
- public void shouldUpdateJobExtIdIfExists() {
- JobInfo jobInfo = new JobInfo();
- when(jobInfoRepository.findById("jobid")).thenReturn(Optional.of(jobInfo));
-
- ArgumentCaptor jobInfoArgumentCaptor = ArgumentCaptor.forClass(JobInfo.class);
- JobExecutionService jobExecutionService = new JobExecutionService(jobInfoRepository, defaults);
- jobExecutionService.updateJobExtId("jobid", "extid");
-
- verify(jobInfoRepository, times(1)).saveAndFlush(jobInfoArgumentCaptor.capture());
-
- JobInfo jobInfoUpdated = new JobInfo();
- jobInfoUpdated.setExtId("extid");
- assertThat(jobInfoArgumentCaptor.getValue(), equalTo(jobInfoUpdated));
- }
-
@Test
public void shouldRunProcessAndGetJobIdIfNoError() throws IOException {
Process process = Mockito.mock(Process.class);
@@ -130,8 +93,7 @@ public void shouldRunProcessAndGetJobIdIfNoError() throws IOException {
when(process.getErrorStream()).thenReturn(errorStream);
when(process.exitValue()).thenReturn(0);
when(process.isAlive()).thenReturn(true).thenReturn(false);
- JobExecutionService jobExecutionService = new JobExecutionService(jobInfoRepository, defaults);
- String jobId = jobExecutionService.runProcess(process);
+ String jobId = dfJobManager.runProcess(process);
assertThat(jobId, equalTo("1231231231"));
}
@@ -149,7 +111,6 @@ public void shouldThrowRuntimeExceptionIfErrorOccursInProcess() {
when(process.exitValue()).thenReturn(1);
when(process.isAlive()).thenReturn(true).thenReturn(false);
expectedException.expect(RuntimeException.class);
- JobExecutionService jobExecutionService = new JobExecutionService(jobInfoRepository, defaults);
- jobExecutionService.runProcess(process);
+ dfJobManager.runProcess(process);
}
}
diff --git a/core/src/test/java/feast/core/job/dataflow/DataflowJobMonitorTest.java b/core/src/test/java/feast/core/job/dataflow/DataflowJobMonitorTest.java
index 61ad74930d..3e1d3f89bf 100644
--- a/core/src/test/java/feast/core/job/dataflow/DataflowJobMonitorTest.java
+++ b/core/src/test/java/feast/core/job/dataflow/DataflowJobMonitorTest.java
@@ -28,6 +28,8 @@
import com.google.api.services.dataflow.Dataflow.Projects.Locations.Jobs;
import com.google.api.services.dataflow.Dataflow.Projects.Locations.Jobs.Get;
import com.google.api.services.dataflow.model.Job;
+import feast.core.job.Runner;
+import feast.core.model.JobInfo;
import feast.core.model.JobStatus;
import java.io.IOException;
import org.junit.Before;
@@ -66,7 +68,10 @@ public void getJobStatus_shouldReturnCorrectJobStatusForValidDataflowJobState()
when(job.getCurrentState()).thenReturn(DataflowJobState.JOB_STATE_RUNNING.toString());
when(jobService.get(projectId, location, jobId)).thenReturn(getOp);
- assertThat(monitor.getJobStatus(jobId), equalTo(JobStatus.RUNNING));
+ JobInfo jobInfo = mock(JobInfo.class);
+ when(jobInfo.getExtId()).thenReturn(jobId);
+ when(jobInfo.getRunner()).thenReturn(Runner.DATAFLOW.getName());
+ assertThat(monitor.getJobStatus(jobInfo), equalTo(JobStatus.RUNNING));
}
@Test
@@ -79,7 +84,10 @@ public void getJobStatus_shouldReturnUnknownStateForInvalidDataflowJobState() th
when(job.getCurrentState()).thenReturn("Random String");
when(jobService.get(projectId, location, jobId)).thenReturn(getOp);
- assertThat(monitor.getJobStatus(jobId), equalTo(JobStatus.UNKNOWN));
+ JobInfo jobInfo = mock(JobInfo.class);
+ when(jobInfo.getExtId()).thenReturn(jobId);
+ when(jobInfo.getRunner()).thenReturn(Runner.DATAFLOW.getName());
+ assertThat(monitor.getJobStatus(jobInfo), equalTo(JobStatus.UNKNOWN));
}
@Test
@@ -88,6 +96,9 @@ public void getJobStatus_shouldReturnUnknownStateWhenExceptionHappen() throws IO
when(jobService.get(projectId, location, jobId)).thenThrow(new RuntimeException("some thing wrong"));
- assertThat(monitor.getJobStatus(jobId), equalTo(JobStatus.UNKNOWN));
+ JobInfo jobInfo = mock(JobInfo.class);
+ when(jobInfo.getExtId()).thenReturn(jobId);
+ when(jobInfo.getRunner()).thenReturn(Runner.DATAFLOW.getName());
+ assertThat(monitor.getJobStatus(jobInfo), equalTo(JobStatus.UNKNOWN));
}
}
\ No newline at end of file
diff --git a/core/src/test/java/feast/core/job/flink/FlinkJobManagerTest.java b/core/src/test/java/feast/core/job/flink/FlinkJobManagerTest.java
new file mode 100644
index 0000000000..cd892696fb
--- /dev/null
+++ b/core/src/test/java/feast/core/job/flink/FlinkJobManagerTest.java
@@ -0,0 +1,107 @@
+package feast.core.job.flink;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import feast.core.config.ImportJobDefaults;
+import feast.specs.ImportSpecProto.ImportSpec;
+import java.util.Collections;
+import org.apache.flink.client.cli.CliFrontend;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+public class FlinkJobManagerTest {
+ @Mock private CliFrontend flinkCli;
+ @Mock private FlinkRestApi flinkRestApi;
+
+ private FlinkJobConfig config;
+ private ImportJobDefaults defaults;
+ private FlinkJobManager flinkJobManager;
+
+ @Before
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ config = new FlinkJobConfig("localhost:8081", "/etc/flink/conf");
+ defaults =
+ new ImportJobDefaults(
+ "localhost:8080",
+ "FlinkRunner",
+ "{\"key\":\"value\"}",
+ "ingestion.jar",
+ "stderr",
+ "{}");
+
+ flinkJobManager = new FlinkJobManager(flinkCli, config, flinkRestApi, defaults);
+ }
+
+ @Test
+ public void shouldPassCorrectArgumentForSubmittingJob() {
+ FlinkJobList response = new FlinkJobList();
+ response.setJobs(Collections.singletonList(new FlinkJob("1234", "job1", "RUNNING")));
+ when(flinkRestApi.getJobsOverview()).thenReturn(response);
+
+ ImportSpec importSpec = ImportSpec.newBuilder().setType("file").build();
+ String jobName = "importjob";
+ flinkJobManager.submitJob(importSpec, jobName);
+ String[] expected =
+ new String[] {
+ "run",
+ "-d",
+ "-m",
+ config.getMasterUrl(),
+ defaults.getExecutable(),
+ "--jobName=" + jobName,
+ "--runner=FlinkRunner",
+ "--importSpecBase64=CgRmaWxl",
+ "--coreApiUri=" + defaults.getCoreApiUri(),
+ "--errorsStoreType=" + defaults.getErrorsStoreType(),
+ "--errorsStoreOptions=" + defaults.getErrorsStoreOptions(),
+ "--key=value"
+ };
+
+ ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(String[].class);
+ verify(flinkCli).parseParameters(argumentCaptor.capture());
+
+ String[] actual = argumentCaptor.getValue();
+ assertThat(actual, equalTo(expected));
+ }
+
+ @Test
+ public void shouldReturnFlinkJobId() {
+ FlinkJobList response = new FlinkJobList();
+ String flinkJobId = "1234";
+ String jobName = "importjob";
+ response.setJobs(Collections.singletonList(new FlinkJob(flinkJobId, jobName, "RUNNING")));
+ when(flinkRestApi.getJobsOverview()).thenReturn(response);
+
+ ImportSpec importSpec = ImportSpec.newBuilder().setType("file").build();
+ String jobId = flinkJobManager.submitJob(importSpec, jobName);
+
+ assertThat(jobId, equalTo(flinkJobId));
+ }
+
+ @Test
+ public void shouldPassCorrectArgumentForStoppingJob() {
+ String jobId = "1234";
+
+ flinkJobManager.abortJob(jobId);
+
+ String[] expected = new String[]{
+ "cancel",
+ "-m",
+ config.getMasterUrl(),
+ jobId
+ };
+
+ ArgumentCaptor argumentCaptor = ArgumentCaptor.forClass(String[].class);
+ verify(flinkCli).parseParameters(argumentCaptor.capture());
+
+ String[] actual = argumentCaptor.getValue();
+ assertThat(actual, equalTo(expected));
+ }
+}
diff --git a/core/src/test/java/feast/core/job/flink/FlinkRestApiTest.java b/core/src/test/java/feast/core/job/flink/FlinkRestApiTest.java
new file mode 100644
index 0000000000..023791d2ee
--- /dev/null
+++ b/core/src/test/java/feast/core/job/flink/FlinkRestApiTest.java
@@ -0,0 +1,113 @@
+package feast.core.job.flink;
+
+import static org.hamcrest.core.IsEqual.equalTo;
+import static org.junit.Assert.assertThat;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Arrays;
+import okhttp3.HttpUrl;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.web.client.RestTemplate;
+
+public class FlinkRestApiTest {
+ FlinkRestApi flinkRestApi;
+ MockWebServer mockWebServer;
+
+ String host;
+ int port;
+
+ @Before
+ public void setUp() throws Exception {
+ mockWebServer = new MockWebServer();
+ mockWebServer.start();
+
+ port = mockWebServer.getPort();
+ host = mockWebServer.getHostName();
+
+ flinkRestApi = new FlinkRestApi(new RestTemplate(), String.format("%s:%d", host, port));
+ }
+
+ @Test
+ public void shouldSendCorrectRequest() throws InterruptedException {
+ MockResponse response = new MockResponse();
+ response.setResponseCode(200);
+ mockWebServer.enqueue(response);
+
+ flinkRestApi.getJobsOverview();
+
+ RecordedRequest recordedRequest = mockWebServer.takeRequest();
+ HttpUrl requestUrl = recordedRequest.getRequestUrl();
+
+ assertThat(requestUrl.host(), equalTo(host));
+ assertThat(requestUrl.port(), equalTo(port));
+ assertThat(requestUrl.encodedPath(), equalTo("/jobs/overview"));
+ }
+
+ @Test
+ public void shouldReturnEmptyJobListForEmptyBody() {
+ MockResponse response = new MockResponse();
+ response.setResponseCode(200);
+ mockWebServer.enqueue(response);
+
+ FlinkJobList jobList = flinkRestApi.getJobsOverview();
+ assertThat(jobList.getJobs().size(), equalTo(0));
+ }
+
+ @Test
+ public void shouldReturnEmptyJobListForEmptyJsonResponse() {
+ mockWebServer.enqueue(createMockResponse(200, "[]"));
+
+ FlinkJobList jobList = flinkRestApi.getJobsOverview();
+ assertThat(jobList.getJobs().size(), equalTo(0));
+
+ mockWebServer.enqueue(createMockResponse(200, "{}"));
+
+ jobList = flinkRestApi.getJobsOverview();
+ assertThat(jobList.getJobs().size(), equalTo(0));
+
+ mockWebServer.enqueue(createMockResponse(200, "{jobs: []}"));
+
+ jobList = flinkRestApi.getJobsOverview();
+ assertThat(jobList.getJobs().size(), equalTo(0));
+ }
+
+ @Test
+ public void shouldReturnCorrectResultForValidResponse() throws JsonProcessingException {
+ FlinkJobList jobList = new FlinkJobList();
+ FlinkJob job1 = new FlinkJob("1234", "job1", "RUNNING");
+ FlinkJob job2 = new FlinkJob("5678", "job2", "RUNNING");
+ FlinkJob job3 = new FlinkJob("1111", "job3", "RUNNING");
+
+ jobList.setJobs(Arrays.asList(job1, job2, job3));
+
+ mockWebServer.enqueue( createMockResponse(200, createResponseBody(jobList)));
+
+ FlinkJobList actual = flinkRestApi.getJobsOverview();
+
+ assertThat(actual.getJobs().size(), equalTo(3));
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ mockWebServer.shutdown();
+ }
+
+ private String createResponseBody(FlinkJobList jobList) throws JsonProcessingException {
+ ObjectMapper objectMapper = new ObjectMapper();
+ return objectMapper.writeValueAsString(jobList);
+ }
+
+ private MockResponse createMockResponse(int statusCode, String body) {
+ MockResponse response = new MockResponse();
+ response.setHeader("Content-Type", "application/json");
+ response.setResponseCode(statusCode);
+ response.setBody(body);
+ return response;
+ }
+}
diff --git a/core/src/test/java/feast/core/service/JobManagementServiceTest.java b/core/src/test/java/feast/core/service/JobManagementServiceTest.java
index 06228c28f5..709cf365b7 100644
--- a/core/src/test/java/feast/core/service/JobManagementServiceTest.java
+++ b/core/src/test/java/feast/core/service/JobManagementServiceTest.java
@@ -17,15 +17,28 @@
package feast.core.service;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.MockitoAnnotations.initMocks;
+
import com.google.common.collect.Lists;
import com.google.protobuf.Timestamp;
import feast.core.JobServiceProto.JobServiceTypes.JobDetail;
+import feast.core.config.ImportJobDefaults;
import feast.core.dao.JobInfoRepository;
import feast.core.dao.MetricsRepository;
import feast.core.exception.RetrievalException;
import feast.core.job.JobManager;
import feast.core.model.JobInfo;
import feast.core.model.JobStatus;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Optional;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -33,28 +46,19 @@
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
-import java.time.Instant;
-import java.util.Collections;
-import java.util.Date;
-import java.util.List;
-import java.util.Optional;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-import static org.mockito.MockitoAnnotations.initMocks;
-
public class JobManagementServiceTest {
+ @Rule public final ExpectedException exception = ExpectedException.none();
@Mock private JobInfoRepository jobInfoRepository;
@Mock private MetricsRepository metricsRepository;
@Mock private JobManager jobManager;
-
- @Rule public final ExpectedException exception = ExpectedException.none();
+ private ImportJobDefaults defaults;
@Before
public void setUp() {
initMocks(this);
+ defaults =
+ new ImportJobDefaults(
+ "localhost:8433", "DirectRunner", "", "/feast-import.jar", "stderr", "");
}
@Test
@@ -89,7 +93,7 @@ public void shouldListAllJobDetails() {
jobInfo2.setLastUpdated(Date.from(Instant.ofEpochSecond(1)));
when(jobInfoRepository.findAll()).thenReturn(Lists.newArrayList(jobInfo1, jobInfo2));
JobManagementService jobManagementService =
- new JobManagementService(jobInfoRepository, metricsRepository, jobManager);
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
List actual = jobManagementService.listJobs();
List expected =
Lists.newArrayList(
@@ -126,7 +130,7 @@ public void shouldReturnDetailOfRequestedJobId() {
jobInfo1.setLastUpdated(Date.from(Instant.ofEpochSecond(1)));
when(jobInfoRepository.findById("job1")).thenReturn(Optional.of(jobInfo1));
JobManagementService jobManagementService =
- new JobManagementService(jobInfoRepository, metricsRepository, jobManager);
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
JobDetail actual = jobManagementService.getJob("job1");
JobDetail expected =
JobDetail.newBuilder()
@@ -142,7 +146,7 @@ public void shouldReturnDetailOfRequestedJobId() {
public void shouldThrowErrorIfJobIdNotFoundWhenGettingJob() {
when(jobInfoRepository.findById("job1")).thenReturn(Optional.empty());
JobManagementService jobManagementService =
- new JobManagementService(jobInfoRepository, metricsRepository, jobManager);
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
exception.expect(RetrievalException.class);
exception.expectMessage("Unable to retrieve job with id job1");
jobManagementService.getJob("job1");
@@ -152,7 +156,7 @@ public void shouldThrowErrorIfJobIdNotFoundWhenGettingJob() {
public void shouldThrowErrorIfJobIdNotFoundWhenAbortingJob() {
when(jobInfoRepository.findById("job1")).thenReturn(Optional.empty());
JobManagementService jobManagementService =
- new JobManagementService(jobInfoRepository, metricsRepository, jobManager);
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
exception.expect(RetrievalException.class);
exception.expectMessage("Unable to retrieve job with id job1");
jobManagementService.abortJob("job1");
@@ -164,7 +168,7 @@ public void shouldThrowErrorIfJobInTerminalStateWhenAbortingJob() {
job.setStatus(JobStatus.COMPLETED);
when(jobInfoRepository.findById("job1")).thenReturn(Optional.of(job));
JobManagementService jobManagementService =
- new JobManagementService(jobInfoRepository, metricsRepository, jobManager);
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
exception.expect(IllegalStateException.class);
exception.expectMessage("Unable to stop job already in terminal state");
jobManagementService.abortJob("job1");
@@ -177,10 +181,44 @@ public void shouldUpdateJobAfterAborting() {
job.setExtId("extId1");
when(jobInfoRepository.findById("job1")).thenReturn(Optional.of(job));
JobManagementService jobManagementService =
- new JobManagementService(jobInfoRepository, metricsRepository, jobManager);
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
jobManagementService.abortJob("job1");
ArgumentCaptor jobCapture = ArgumentCaptor.forClass(JobInfo.class);
verify(jobInfoRepository).saveAndFlush(jobCapture.capture());
assertThat(jobCapture.getValue().getStatus(), equalTo(JobStatus.ABORTING));
}
+
+ @Test
+ public void shouldUpdateJobStatusIfExists() {
+ JobInfo jobInfo = new JobInfo();
+ when(jobInfoRepository.findById("jobid")).thenReturn(Optional.of(jobInfo));
+
+ ArgumentCaptor jobInfoArgumentCaptor = ArgumentCaptor.forClass(JobInfo.class);
+ JobManagementService jobExecutionService =
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
+ jobExecutionService.updateJobStatus("jobid", JobStatus.PENDING);
+
+ verify(jobInfoRepository, times(1)).save(jobInfoArgumentCaptor.capture());
+
+ JobInfo jobInfoUpdated = new JobInfo();
+ jobInfoUpdated.setStatus(JobStatus.PENDING);
+ assertThat(jobInfoArgumentCaptor.getValue(), equalTo(jobInfoUpdated));
+ }
+
+ @Test
+ public void shouldUpdateJobExtIdIfExists() {
+ JobInfo jobInfo = new JobInfo();
+ when(jobInfoRepository.findById("jobid")).thenReturn(Optional.of(jobInfo));
+
+ ArgumentCaptor jobInfoArgumentCaptor = ArgumentCaptor.forClass(JobInfo.class);
+ JobManagementService jobExecutionService =
+ new JobManagementService(jobInfoRepository, metricsRepository, jobManager, defaults);
+ jobExecutionService.updateJobExtId("jobid", "extid");
+
+ verify(jobInfoRepository, times(1)).save(jobInfoArgumentCaptor.capture());
+
+ JobInfo jobInfoUpdated = new JobInfo();
+ jobInfoUpdated.setExtId("extid");
+ assertThat(jobInfoArgumentCaptor.getValue(), equalTo(jobInfoUpdated));
+ }
}