[FLINK-9212][flip6] Port SubtasksAllAccumulatorsHandler to new REST endpoint #5893

yew1eb · 2018-04-22T11:04:02Z

What is the purpose of the change

Port SubtasksAllAccumulatorsHandler to new REST endpoint

Brief change log

Add SubtasksAllAccumulatorsHandler for WebMonitorEndpoint
Add SubtasksAllAccumulatorsHeaders/SubtasksAllAccumulatorsInfo for rest response
Add SubtasksAllAccumulatorsInfoTest

Verifying this change

Added test case SubtasksAllAccumulatorsInfoTest to generate the json response

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

…ndpoint

zentol · 2018-04-23T11:14:03Z

.../src/main/java/org/apache/flink/runtime/rest/handler/job/SubtasksAllAccumulatorsHandler.java

+public class SubtasksAllAccumulatorsHandler extends AbstractJobVertexHandler<SubtasksAllAccumulatorsInfo, JobVertexMessageParameters> {
+
+	/**
+	 * Instantiates a new subtasks all accumulators handler.


we typically don't have javadocs on constructors as they rarely offer any value

zentol · 2018-04-23T11:15:01Z

...me/src/main/java/org/apache/flink/runtime/rest/messages/job/SubtasksAllAccumulatorsInfo.java

+import java.util.Collection;
+import java.util.Objects;
+
+


remove empty line

zentol · 2018-04-23T11:15:54Z

.../src/main/java/org/apache/flink/runtime/rest/handler/job/SubtasksAllAccumulatorsHandler.java

+				));
+		}
+
+		return new SubtasksAllAccumulatorsInfo(vertexId, parallelism, subtaskAccumulatorsInfos);


the vertex ID should be passed as a JobVertexID, not as a string.

zentol · 2018-04-23T11:21:34Z

...me/src/main/java/org/apache/flink/runtime/rest/messages/job/SubtasksAllAccumulatorsInfo.java

+	public static final String FILED_NMAE_SUBTASKS = "subtasks";
+
+	@JsonProperty(FIELD_NAME_ID)
+	private String id;


this should be a JobVertexID, and a more descriptive field name would be great. (the JSON field name should remain the same ;) )

zentol · 2018-04-23T11:24:14Z

.../src/main/java/org/apache/flink/runtime/rest/handler/job/SubtasksAllAccumulatorsHandler.java

+
+		final List<SubtasksAllAccumulatorsInfo.SubtaskAccumulatorsInfo> subtaskAccumulatorsInfos = new ArrayList<>();
+
+		int index = 0;


This is not the actual subtask index. Use vertex.getCurrentExecutionAttempt().getParallelSubtaskIndex() instead.

zentol · 2018-04-23T11:26:26Z

...rc/test/java/org/apache/flink/runtime/rest/messages/job/SubtasksAllAccumulatorsInfoTest.java

+
+	@Override
+	protected SubtasksAllAccumulatorsInfo getTestResponseInstance() throws Exception {
+		List<SubtasksAllAccumulatorsInfo.SubtaskAccumulatorsInfo> subtaskAccumulatorsInfos = new ArrayList<>();


you now the size of this array already. (3)

zentol · 2018-04-23T11:26:35Z

...rc/test/java/org/apache/flink/runtime/rest/messages/job/SubtasksAllAccumulatorsInfoTest.java

+	protected SubtasksAllAccumulatorsInfo getTestResponseInstance() throws Exception {
+		List<SubtasksAllAccumulatorsInfo.SubtaskAccumulatorsInfo> subtaskAccumulatorsInfos = new ArrayList<>();
+
+		List<UserAccumulator> userAccumulators = new ArrayList<>(3);


you actually only have 2 accumulators.

yew1eb · 2018-04-23T12:05:08Z

@zentol thanks for your review. I have updated the PR.

zentol · 2018-04-23T12:12:44Z

You have not changed the type of SubtasksAllAccumulatorsInfo#jobVertexId to JobVertexID.

yew1eb · 2018-04-23T12:17:12Z

i saw legacy/SubtasksAllAccumulatorsHandler.java used gen.writeStringField("id", jobVertex.getJobVertexId().toString());.

if i change the type of SubtasksAllAccumulatorsInfo#jobVertexId to JobVertexID ,
run the SubtasksAllAccumulatorsInfoTest will fail, error log as follow:

org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "bytes" (class org.apache.flink.runtime.jobgraph.JobVertexID), not marked as ignorable (2 known properties: "lowerPart", "upperPart"])
at [Source: {"id":{"upperPart":-7327168959427448358,"lowerPart":4189391734336269051,"bytes":"OiO1lgYSKvuaUKqnBA7x2g=="},"parallelism":4,"subtasks":[{"subtask":0,"attempt":0,"host":"host-0","user-accumulators":[{"name":"test name1","type":"test type1","value":"test value1"},{"name":"test name2","type":"test type2","value":"test value2"}]},{"subtask":1,"attempt":1,"host":"host-1","user-accumulators":[{"name":"test name1","type":"test type1","value":"test value1"},{"name":"test name2","type":"test type2","value":"test value2"}]},{"subtask":2,"attempt":2,"host":"host-2","user-accumulators":[{"name":"test name1","type":"test type1","value":"test value1"},{"name":"test name2","type":"test type2","value":"test value2"}]}]}; line: 1, column: 82] (through reference chain: org.apache.flink.runtime.rest.messages.job.SubtasksAllAccumulatorsInfo["id"]->org.apache.flink.runtime.jobgraph.JobVertexID["bytes"])

zentol · 2018-04-23T13:02:11Z

You have to use the JobVertexIDSerializer for that field.

zentol · 2018-04-23T13:20:34Z

see JobVertexDetailsInfo on how to use it.

yew1eb · 2018-04-23T13:55:48Z

@zentol thanks you. I updated the code. 😆

zentol

2 minor comments, otherwise +1.

zentol · 2018-04-24T07:44:41Z

...c/main/java/org/apache/flink/runtime/rest/handler/legacy/SubtasksAllAccumulatorsHandler.java

 		for (AccessExecutionVertex vertex : jobVertex.getTaskVertices()) {

 			TaskManagerLocation location = vertex.getCurrentAssignedResourceLocation();
 			String locationString = location == null ? "(unassigned)" : location.getHostname();

 			gen.writeStartObject();

-			gen.writeNumberField("subtask", num++);
+			gen.writeNumberField("subtask", vertex.getCurrentExecutionAttempt().getParallelSubtaskIndex());


Let's keep this at it is.

zentol · 2018-04-24T07:45:20Z

...me/src/main/java/org/apache/flink/runtime/rest/messages/job/SubtasksAllAccumulatorsInfo.java

 	public static final String FIELD_NAME_PARALLELISM = "parallelism";
-	public static final String FILED_NMAE_SUBTASKS = "subtasks";
+	public static final String FILED_NAME_SUBTASKS = "subtasks";


typo: FILED -> FIELD

zentol · 2018-04-24T11:30:41Z

I'll address the remaining issues myself when merging, thanks for working on this!

…dpoint This closes apache#5893.

…dpoint This closes #5893.

…dpoint This closes apache#5893.

[FLINK-9212][flip6] Port SubtasksAllAccumulatorsHandler to new REST e…

89905ea

…ndpoint

yew1eb force-pushed the FLINK-9212 branch from 94e1626 to 89905ea Compare April 22, 2018 11:15

zentol requested changes Apr 23, 2018

View reviewed changes

yew1eb force-pushed the FLINK-9212 branch 3 times, most recently from 267f294 to 7e8cc4a Compare April 23, 2018 13:48

update

0232439

yew1eb force-pushed the FLINK-9212 branch from 7e8cc4a to 0232439 Compare April 23, 2018 13:49

zentol approved these changes Apr 24, 2018

View reviewed changes

zentol pushed a commit to zentol/flink that referenced this pull request Apr 24, 2018

[FLINK-9212][REST] Port SubtasksAllAccumulatorsHandler to new REST en…

99c26d3

…dpoint This closes apache#5893.

asfgit pushed a commit that referenced this pull request Apr 25, 2018

[FLINK-9212][REST] Port SubtasksAllAccumulatorsHandler to new REST en…

210abee

…dpoint This closes #5893.

asfgit closed this in 512083a Apr 25, 2018

sampathBhat pushed a commit to sampathBhat/flink that referenced this pull request Jul 26, 2018

[FLINK-9212][REST] Port SubtasksAllAccumulatorsHandler to new REST en…

9072f5b

…dpoint This closes apache#5893.

rmetzger added the component=Runtime/REST label Mar 18, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-9212][flip6] Port SubtasksAllAccumulatorsHandler to new REST endpoint #5893

[FLINK-9212][flip6] Port SubtasksAllAccumulatorsHandler to new REST endpoint #5893

yew1eb commented Apr 22, 2018

zentol Apr 23, 2018

zentol Apr 23, 2018

zentol Apr 23, 2018

zentol Apr 23, 2018

zentol Apr 23, 2018

zentol Apr 23, 2018

zentol Apr 23, 2018

yew1eb commented Apr 23, 2018

zentol commented Apr 23, 2018

yew1eb commented Apr 23, 2018 •

edited

zentol commented Apr 23, 2018

zentol commented Apr 23, 2018

yew1eb commented Apr 23, 2018

zentol left a comment

zentol Apr 24, 2018

zentol Apr 24, 2018

zentol commented Apr 24, 2018


		final List<SubtasksAllAccumulatorsInfo.SubtaskAccumulatorsInfo> subtaskAccumulatorsInfos = new ArrayList<>();

		int index = 0;

		import java.util.Collection;
		import java.util.Objects;

[FLINK-9212][flip6] Port SubtasksAllAccumulatorsHandler to new REST endpoint #5893

[FLINK-9212][flip6] Port SubtasksAllAccumulatorsHandler to new REST endpoint #5893

Conversation

yew1eb commented Apr 22, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yew1eb commented Apr 23, 2018

zentol commented Apr 23, 2018

yew1eb commented Apr 23, 2018 • edited

zentol commented Apr 23, 2018

zentol commented Apr 23, 2018

yew1eb commented Apr 23, 2018

zentol left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zentol commented Apr 24, 2018

yew1eb commented Apr 23, 2018 •

edited