[FLINK-8894][REST] Set object codec for JsonGenerator used by CurrentJobIdsHandler
This closes #5711. Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/096a89ae Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/096a89ae Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/096a89ae Branch: refs/heads/master Commit: 096a89ae6d0db9d763648a0acca2334f14efd3b3 Parents: efd7336 Author: gyao <[email protected]> Authored: Fri Mar 16 20:52:11 2018 +0100 Committer: Till Rohrmann <[email protected]> Committed: Sun Mar 18 16:04:04 2018 +0100 ---------------------------------------------------------------------- .../handler/legacy/CurrentJobIdsHandler.java | 2 + .../legacy/CurrentJobIdsHandlerTest.java | 62 +++++++++++++++++--- 2 files changed, 57 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/096a89ae/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandler.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandler.java index cf8a3d6..ef02762 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandler.java @@ -22,6 +22,7 @@ import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.jobmaster.JobManagerGateway; import org.apache.flink.runtime.messages.webmonitor.JobIdsWithStatusOverview; import org.apache.flink.runtime.rest.messages.JobIdsWithStatusesOverviewHeaders; +import org.apache.flink.runtime.rest.util.RestMapperUtils; import org.apache.flink.util.FlinkException; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator; @@ -70,6 +71,7 @@ public class CurrentJobIdsHandler extends AbstractJsonRequestHandler { StringWriter writer = new StringWriter(); JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer); + gen.setCodec(RestMapperUtils.getStrictObjectMapper()); gen.writeStartObject(); gen.writeArrayFieldStart(JobIdsWithStatusOverview.FIELD_NAME_JOBS); http://git-wip-us.apache.org/repos/asf/flink/blob/096a89ae/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandlerTest.java ---------------------------------------------------------------------- diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandlerTest.java index 0ada30d..b193122 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandlerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/CurrentJobIdsHandlerTest.java @@ -18,21 +18,69 @@ package org.apache.flink.runtime.rest.handler.legacy; +import org.apache.flink.api.common.JobID; import org.apache.flink.api.common.time.Time; import org.apache.flink.runtime.concurrent.Executors; +import org.apache.flink.runtime.jobgraph.JobStatus; +import org.apache.flink.runtime.jobmaster.JobManagerGateway; +import org.apache.flink.runtime.messages.webmonitor.JobIdsWithStatusOverview; +import org.apache.flink.util.TestLogger; -import org.junit.Assert; +import org.junit.Before; import org.junit.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import java.util.Collections; +import java.util.concurrent.CompletableFuture; + +import static org.hamcrest.Matchers.containsString; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.when; /** - * Tests for the CurrentJobIdsHandler. + * Tests for {@link CurrentJobIdsHandler}. */ -public class CurrentJobIdsHandlerTest { +public class CurrentJobIdsHandlerTest extends TestLogger { + + private CurrentJobIdsHandler currentJobIdsHandler; + + @Mock + private JobManagerGateway mockJobManagerGateway; + + @Before + public void setUp() { + MockitoAnnotations.initMocks(this); + currentJobIdsHandler = new CurrentJobIdsHandler(Executors.directExecutor(), Time.seconds(0L)); + } + @Test public void testGetPaths() { - CurrentJobIdsHandler handler = new CurrentJobIdsHandler(Executors.directExecutor(), Time.seconds(0L)); - String[] paths = handler.getPaths(); - Assert.assertEquals(1, paths.length); - Assert.assertEquals("/jobs", paths[0]); + final String[] paths = currentJobIdsHandler.getPaths(); + assertEquals(1, paths.length); + assertEquals("/jobs", paths[0]); } + + @Test + public void testHandleJsonRequest() throws Exception { + final JobID jobId = new JobID(); + final JobStatus jobStatus = JobStatus.RUNNING; + + when(mockJobManagerGateway.requestJobsOverview(any(Time.class))).thenReturn( + CompletableFuture.completedFuture(new JobIdsWithStatusOverview(Collections.singleton( + new JobIdsWithStatusOverview.JobIdWithStatus(jobId, jobStatus))))); + + final CompletableFuture<String> jsonFuture = currentJobIdsHandler.handleJsonRequest( + Collections.emptyMap(), + Collections.emptyMap(), + mockJobManagerGateway); + + final String json = jsonFuture.get(); + + assertThat(json, containsString(jobId.toString())); + assertThat(json, containsString(jobStatus.name())); + } + }
