ferenc-csaky commented on code in PR #27540:
URL: https://github.com/apache/flink/pull/27540#discussion_r3009552775
##########
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java:
##########
@@ -850,6 +852,25 @@ public CompletableFuture<ArchivedExecutionGraph>
getArchivedExecutionGraph(JobID
.thenApply(ExecutionGraphInfo::getArchivedExecutionGraph));
}
+ public CompletableFuture<ExecutionGraphInfo> getExecutionGraphInfo(JobID
jobId) {
+ return runDispatcherCommand(
+ dispatcherGateway ->
+ dispatcherGateway.requestExecutionGraphInfo(jobId,
rpcTimeout));
+ }
+
+ public CompletableFuture<Acknowledge> updateJobResourceRequirements(
+ JobID jobId, JobResourceRequirements resourceRequirements) {
+ if
(miniClusterConfiguration.getConfiguration().get(JobManagerOptions.SCHEDULER)
+ != JobManagerOptions.SchedulerType.Adaptive) {
+ throw new UnsupportedOperationException(
+ "Unsupported updateJobResourceRequirements when scheduler
type is adaptive");
Review Comment:
The statement should be the opposite, this is only supported for adaptive
scheduler:
```
"updateJobResourceRequirements is only supported for adaptive scheduler"
```
##########
flink-core/src/main/java/org/apache/flink/configuration/WebOptions.java:
##########
@@ -141,6 +141,20 @@ public class WebOptions {
.withDescription(
"The maximum number of failures collected by the
exception history per job.");
+ /** The maximum number of the adaptive scheduler rescale history. */
+ @Documentation.Section(Documentation.Sections.ALL_JOB_MANAGER)
+ public static final ConfigOption<Integer>
MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE =
+ key("web.adaptive-scheduler.rescale-history.size")
+ .intType()
+ .defaultValue(0)
+ .withDescription(
+ Description.builder()
+ .text(
+ "The maximum number of the rescale
records per job whose scheduler is %s. "
Review Comment:
I'd rather state something like "`jobmanager.scheduler` is set to
`adaptive`" in the description, cause that's how the user-facing configuration
looks like. `AdaptiveScheduler` as an implementation is pretty much internal
IMO.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionGraphInfo.java:
##########
@@ -50,31 +51,44 @@ public class ExecutionGraphInfo implements Serializable {
*/
@Nullable private final JobRescaleConfigInfo jobRescaleConfigInfo;
+ /**
+ * The field is non-null when the {@link
+ *
org.apache.flink.configuration.WebOptions#MAX_ADAPTIVE_SCHEDULER_RESCALE_HISTORY_SIZE}
is
+ * enabled.
+ */
+ @Nullable private final RescalesStatsSnapshot rescalesStatsSnapshot;
+
public ExecutionGraphInfo(ArchivedExecutionGraph executionGraph) {
this(
executionGraph,
executionGraph.getFailureInfo() != null
? Collections.singleton(
RootExceptionHistoryEntry.fromGlobalFailure(
executionGraph.getFailureInfo()))
- : Collections.emptyList());
+ : Collections.emptyList(),
Review Comment:
Since we touch the ctor, i'd replace this with `List.of()`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/timeline/DefaultRescaleTimeline.java:
##########
@@ -122,6 +124,14 @@ Rescale currentRescale() {
return currentRescale;
}
+ @Override
+ public RescalesStatsSnapshot createSnapshot() {
+ List<Rescale> rescales = rescaleHistory.toArrayList();
+ Collections.reverse(rescales);
+ return new RescalesStatsSnapshot(
+ Collections.unmodifiableList(rescales),
rescalesSummary.createSnapshot());
Review Comment:
nit: `List.copyOf` instead of `Collections.unmodifiableList`. We finally
ditched JDK8, we can start use newer features. :)
##########
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionGraphInfo.java:
##########
@@ -50,31 +51,44 @@ public class ExecutionGraphInfo implements Serializable {
*/
@Nullable private final JobRescaleConfigInfo jobRescaleConfigInfo;
+ /**
+ * The field is non-null when the {@link
Review Comment:
I'd revert the phrasing and state `This field is null when ... is not a
positive number` to match with javadoc on `JobRescaleConfigInfo`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
