[
https://issues.apache.org/jira/browse/FLINK-40403?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Tim Steinbach updated FLINK-40403:
----------------------------------
Description:
*Problem*
MetricStore.TaskMetricStore#retainSubtasks and #isTransientMetric identify the
per-subtask key prefix (e.g. "0.numRecordsIn") with String#matches(String):
{{// retainSubtasks}}
{{{}return index.matches("\\{}}}{{{}d+") &&
!activeSubtasks.contains(Integer.parseInt(index));{}}}
{{// isTransientMetric}}
{{{}return name.matches("^\\d+\\{}}}{{{}..*") &&
super.isTransientMetric(name);{}}}
String#matches recompiles the regex Pattern on every call. Both methods run
once per metric key, and retainSubtasks is invoked once per vertex on every
MetricFetcher refresh from MetricStore#updateCurrentExecutionAttempts, which is
"public synchronized" — i.e. the O(#metric-keys) work, including a fresh
Pattern.compile per key, executes while the single global MetricStore monitor
is held.
*Impact*
On jobs with a large number of subtasks the per-refresh Pattern compilation
dominates the time the MetricStore monitor is held. Because the JobManager's
REST handler thread pool (rest.server.numThreads, default 4) is shared between
the MetricFetcher and the REST endpoints, the whole pool stalls on that monitor:
* GET /jobs/<jid> — which aggregates per-subtask IO metrics via
MutableIOMetrics#addIOMetrics and takes the same monitor once per subtask;
becomes very slow (observed 38-150 s, sometimes timing out)
* Endpoints that never touch MetricStore (GET /jobs/overview,
/jobs/<jid>/checkpoints) queue behind the same saturated pool. These are on the
Kubernetes Operator's observation path
Observed on a production job with ~2,233 subtasks (main operator chain at
parallelism 900). A thread dump taken while GET /jobs/<jid> was hanging shows
the entire REST pool contending on the MetricStore monitor: one thread RUNNABLE
inside java.util.regex.Pattern$...match / String#matches while holding the
monitor (MetricStore.TaskMetricStore#retainSubtasks -> keySet().removeIf), the
others BLOCKED on it. CPU is otherwise near-idle — this is lock-hold time, not
compute.
*Proposed fix*
Hoist the two regexes to static final Pattern fields on TaskMetricStore and
match via matcher(...).matches(). This is behaviour-preserving —
String#matches(regex) is specified as
Pattern.compile(regex).matcher(s).matches(), and the regex strings are
unchanged — and removes the per-key compilation from inside the lock. Existing
MetricStoreTest coverage (testTaskMetricStoreCleanup,
testSubtaskMetricStoreCleanup, testMalformedNameHandling) exercises both call
sites.
*Reproduction*
Run a job with many subtasks, open the Web UI job page (or poll GET
/jobs/<jid>), and observe REST latency plus a thread dump of the JobManager
showing REST threads BLOCKED on the MetricStore monitor with the holder in
String#matches / Pattern compilation under retainSubtasks.
was:
*Problem*
MetricStore.TaskMetricStore#retainSubtasks and #isTransientMetric identify the
per-subtask key prefix (e.g. "0.numRecordsIn") with String#matches(String):
{{// retainSubtasks}}
{{return index.matches("\\d+") &&
!activeSubtasks.contains(Integer.parseInt(index));}}
{{// isTransientMetric}}
{{return name.matches("^\\d+\\..*") && super.isTransientMetric(name);}}
String#matches recompiles the regex Pattern on every call. Both methods run
once per metric key, and retainSubtasks is invoked once per vertex on every
MetricFetcher refresh from MetricStore#updateCurrentExecutionAttempts, which is
"public synchronized" — i.e. the O(#metric-keys) work, including a fresh
Pattern.compile per key, executes while the single global MetricStore monitor
is held.
*Impact*
On jobs with a large number of subtasks the per-refresh Pattern compilation
dominates the time the MetricStore monitor is held. Because the JobManager's
REST handler thread pool (rest.server.numThreads, default 4) is shared between
the MetricFetcher and the REST endpoints, the whole pool stalls on that monitor:
* GET /jobs/<jid> — which aggregates per-subtask IO metrics via
MutableIOMetrics#addIOMetrics and takes the same monitor once per subtask;
becomes very slow (observed 38-150 s, sometimes timing out)
* Endpoints that never touch MetricStore (GET /jobs/overview,
/jobs/<jid>/checkpoints) queue behind the same saturated pool. These are on the
Kubernetes Operator's observation path
Observed on a production job with ~2,233 subtasks (main operator chain at
parallelism 900). A thread dump taken while GET /jobs/<jid> was hanging shows
the entire REST pool contending on the MetricStore monitor: one thread RUNNABLE
inside java.util.regex.Pattern$...match / String#matches while holding the
monitor (MetricStore.TaskMetricStore#retainSubtasks -> keySet().removeIf), the
others BLOCKED on it. CPU is otherwise near-idle — this is lock-hold time, not
compute.
*Proposed fix*
Hoist the two regexes to static final Pattern fields on TaskMetricStore and
match via matcher(...).matches(). This is behaviour-preserving —
String#matches(regex) is specified as
Pattern.compile(regex).matcher(s).matches(), and the regex strings are
unchanged — and removes the per-key compilation from inside the lock. Existing
MetricStoreTest coverage (testTaskMetricStoreCleanup,
testSubtaskMetricStoreCleanup, testMalformedNameHandling) exercises both call
sites.
*Reproduction*
Run a job with many subtasks, open the Web UI job page (or poll GET
/jobs/<jid>), and observe REST latency plus a thread dump of the JobManager
showing REST threads BLOCKED on the MetricStore monitor with the holder in
String#matches / Pattern compilation under retainSubtasks.
> MetricStore.TaskMetricStore recompiles subtask-index regexes on every metric
> fetch, starving the REST thread pool on high-parallelism jobs
> ------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40403
> URL: https://issues.apache.org/jira/browse/FLINK-40403
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / REST
> Affects Versions: 2.3.0
> Reporter: Tim Steinbach
> Priority: Minor
>
> *Problem*
> MetricStore.TaskMetricStore#retainSubtasks and #isTransientMetric identify
> the per-subtask key prefix (e.g. "0.numRecordsIn") with
> String#matches(String):
> {{// retainSubtasks}}
> {{{}return index.matches("\\{}}}{{{}d+") &&
> !activeSubtasks.contains(Integer.parseInt(index));{}}}
> {{// isTransientMetric}}
> {{{}return name.matches("^\\d+\\{}}}{{{}..*") &&
> super.isTransientMetric(name);{}}}
> String#matches recompiles the regex Pattern on every call. Both methods run
> once per metric key, and retainSubtasks is invoked once per vertex on every
> MetricFetcher refresh from MetricStore#updateCurrentExecutionAttempts, which
> is "public synchronized" — i.e. the O(#metric-keys) work, including a fresh
> Pattern.compile per key, executes while the single global MetricStore monitor
> is held.
> *Impact*
> On jobs with a large number of subtasks the per-refresh Pattern compilation
> dominates the time the MetricStore monitor is held. Because the JobManager's
> REST handler thread pool (rest.server.numThreads, default 4) is shared
> between the MetricFetcher and the REST endpoints, the whole pool stalls on
> that monitor:
> * GET /jobs/<jid> — which aggregates per-subtask IO metrics via
> MutableIOMetrics#addIOMetrics and takes the same monitor once per subtask;
> becomes very slow (observed 38-150 s, sometimes timing out)
> * Endpoints that never touch MetricStore (GET /jobs/overview,
> /jobs/<jid>/checkpoints) queue behind the same saturated pool. These are on
> the Kubernetes Operator's observation path
>
> Observed on a production job with ~2,233 subtasks (main operator chain at
> parallelism 900). A thread dump taken while GET /jobs/<jid> was hanging shows
> the entire REST pool contending on the MetricStore monitor: one thread
> RUNNABLE inside java.util.regex.Pattern$...match / String#matches while
> holding the monitor (MetricStore.TaskMetricStore#retainSubtasks ->
> keySet().removeIf), the others BLOCKED on it. CPU is otherwise near-idle —
> this is lock-hold time, not compute.
> *Proposed fix*
> Hoist the two regexes to static final Pattern fields on TaskMetricStore and
> match via matcher(...).matches(). This is behaviour-preserving —
> String#matches(regex) is specified as
> Pattern.compile(regex).matcher(s).matches(), and the regex strings are
> unchanged — and removes the per-key compilation from inside the lock.
> Existing MetricStoreTest coverage (testTaskMetricStoreCleanup,
> testSubtaskMetricStoreCleanup, testMalformedNameHandling) exercises both call
> sites.
> *Reproduction*
> Run a job with many subtasks, open the Web UI job page (or poll GET
> /jobs/<jid>), and observe REST latency plus a thread dump of the JobManager
> showing REST threads BLOCKED on the MetricStore monitor with the holder in
> String#matches / Pattern compilation under retainSubtasks.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
