This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 168800087 [AMORO-1766] Implement table optimizing metrics collection.
(#2674)
168800087 is described below
commit 168800087886c17df3d6081ec8d4cb5522b26174
Author: baiyangtx <[email protected]>
AuthorDate: Wed Apr 24 15:36:43 2024 +0800
[AMORO-1766] Implement table optimizing metrics collection. (#2674)
* table optimizing metrics
* review comments
* review comments
* fix reviews
* fix reviews
* fix reviews
* fix merge conflict
---------
Co-authored-by: ZhouJinsong <[email protected]>
---
.../amoro/server/optimizing/OptimizingQueue.java | 3 +
.../apache/amoro/server/table/TableMetrics.java | 180 -----------
.../amoro/server/table/TableOptimizingMetrics.java | 353 +++++++++++++++++++++
.../apache/amoro/server/table/TableRuntime.java | 22 +-
.../server/optimizing/TestOptimizingQueue.java | 4 -
5 files changed, 368 insertions(+), 194 deletions(-)
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
index eaa7bb316..a1c1842a9 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/optimizing/OptimizingQueue.java
@@ -414,6 +414,9 @@ public class OptimizingQueue extends PersistentBase {
public void close() {
lock.lock();
try {
+ if (this.status != Status.RUNNING) {
+ return;
+ }
this.status = OptimizingProcess.Status.CLOSED;
this.endTime = System.currentTimeMillis();
persistProcessCompleted(false);
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableMetrics.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableMetrics.java
deleted file mode 100644
index e547159ec..000000000
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableMetrics.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.amoro.server.table;
-
-import static org.apache.amoro.api.metrics.MetricDefine.defineGauge;
-
-import org.apache.amoro.api.ServerTableIdentifier;
-import org.apache.amoro.api.metrics.Gauge;
-import org.apache.amoro.api.metrics.Metric;
-import org.apache.amoro.api.metrics.MetricDefine;
-import org.apache.amoro.api.metrics.MetricKey;
-import org.apache.amoro.server.metrics.MetricRegistry;
-import org.apache.amoro.server.optimizing.OptimizingStatus;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-
-import java.util.List;
-
-public class TableMetrics {
- /** Table is no need optimizing. */
- public static final String STATE_IDLE = "idle";
-
- /** Table is need optimizing, but waiting for resource */
- public static final String STATE_PENDING = "pending";
-
- /** Table is doing optimizing process planing. */
- public static final String STATE_PLANING = "planing";
-
- /** Table is executing optimizing process */
- public static final String STATE_EXECUTING = "executing";
-
- /** All optimizing process task is done, and process is committing. */
- public static final String STATE_COMMITTING = "committing";
-
- // table optimizing status duration metrics
- public static final MetricDefine TABLE_OPTIMIZING_STATE_IDLE_DURATION =
- defineGauge("table_optimizing_status_idle_duration_mills")
- .withDescription("Duration in seconds after table be in idle state")
- .withTags("catalog", "database", "table")
- .build();
-
- public static final MetricDefine TABLE_OPTIMIZING_STATE_PENDING_DURATION =
- defineGauge("table_optimizing_status_pending_duration_mills")
- .withDescription("Duration in seconds after table be in pending
state")
- .withTags("catalog", "database", "table")
- .build();
-
- public static final MetricDefine TABLE_OPTIMIZING_STATE_PLANNING_DURATION =
- defineGauge("table_optimizing_status_planning_duration_mills")
- .withDescription("Duration in seconds after table be in planning
state")
- .withTags("catalog", "database", "table")
- .build();
-
- public static final MetricDefine TABLE_OPTIMIZING_STATE_EXECUTING_DURATION =
- defineGauge("table_optimizing_status_executing_duration_mills")
- .withDescription("Duration in seconds after table be in executing
state")
- .withTags("catalog", "database", "table")
- .build();
-
- public static final MetricDefine TABLE_OPTIMIZING_STATE_COMMITTING_DURATION =
- defineGauge("table_optimizing_status_committing_duration_mills")
- .withDescription("Duration in seconds after table be in committing
state")
- .withTags("catalog", "database", "table")
- .build();
-
- private final ServerTableIdentifier identifier;
-
- private OptimizingStatus optimizingStatus;
- private long stateSetTimestamp = System.currentTimeMillis();
- private final List<MetricKey> registeredMetricKeys = Lists.newArrayList();
- private MetricRegistry globalRegistry;
-
- public TableMetrics(ServerTableIdentifier identifier) {
- this.identifier = identifier;
- }
-
- private void registerMetric(MetricRegistry registry, MetricDefine define,
Metric metric) {
- MetricKey key =
- registry.register(
- define,
- ImmutableMap.of(
- "catalog",
- identifier.getCatalog(),
- "database",
- identifier.getDatabase(),
- "table",
- identifier.getTableName()),
- metric);
- registeredMetricKeys.add(key);
- }
-
- public void register(MetricRegistry registry) {
- if (globalRegistry == null) {
- registerMetric(
- registry, TABLE_OPTIMIZING_STATE_IDLE_DURATION, new
StateDurationGauge(STATE_IDLE));
- registerMetric(
- registry, TABLE_OPTIMIZING_STATE_PENDING_DURATION, new
StateDurationGauge(STATE_PENDING));
- registerMetric(
- registry,
- TABLE_OPTIMIZING_STATE_PLANNING_DURATION,
- new StateDurationGauge(STATE_PLANING));
- registerMetric(
- registry,
- TABLE_OPTIMIZING_STATE_EXECUTING_DURATION,
- new StateDurationGauge(STATE_EXECUTING));
- registerMetric(
- registry,
- TABLE_OPTIMIZING_STATE_COMMITTING_DURATION,
- new StateDurationGauge(STATE_COMMITTING));
- globalRegistry = registry;
- }
- }
-
- public void unregister() {
- registeredMetricKeys.forEach(globalRegistry::unregister);
- registeredMetricKeys.clear();
- globalRegistry = null;
- }
-
- public void stateChanged(OptimizingStatus optimizingStatus, long
stateSetTimestamp) {
- this.optimizingStatus = optimizingStatus;
- this.stateSetTimestamp = stateSetTimestamp;
- }
-
- class StateDurationGauge implements Gauge<Long> {
- final String targetState;
-
- StateDurationGauge(String targetState) {
- this.targetState = targetState;
- }
-
- @Override
- public Long getValue() {
- String state = optimizingStatusToMetricState(optimizingStatus);
- if (targetState.equals(state)) {
- return stateDuration();
- }
- return 0L;
- }
-
- private String optimizingStatusToMetricState(OptimizingStatus status) {
- switch (status) {
- case IDLE:
- return STATE_IDLE;
- case PENDING:
- return STATE_PENDING;
- case PLANNING:
- return STATE_PLANING;
- case FULL_OPTIMIZING:
- case MAJOR_OPTIMIZING:
- case MINOR_OPTIMIZING:
- return STATE_EXECUTING;
- case COMMITTING:
- return STATE_COMMITTING;
- default:
- return status.name();
- }
- }
-
- private Long stateDuration() {
- return System.currentTimeMillis() - stateSetTimestamp;
- }
- }
-}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
new file mode 100644
index 000000000..6ee04c5cc
--- /dev/null
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableOptimizingMetrics.java
@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.table;
+
+import static org.apache.amoro.api.metrics.MetricDefine.defineCounter;
+import static org.apache.amoro.api.metrics.MetricDefine.defineGauge;
+
+import org.apache.amoro.api.ServerTableIdentifier;
+import org.apache.amoro.api.metrics.Counter;
+import org.apache.amoro.api.metrics.Gauge;
+import org.apache.amoro.api.metrics.Metric;
+import org.apache.amoro.api.metrics.MetricDefine;
+import org.apache.amoro.api.metrics.MetricKey;
+import org.apache.amoro.server.metrics.MetricRegistry;
+import org.apache.amoro.server.optimizing.OptimizingStatus;
+import org.apache.amoro.server.optimizing.OptimizingType;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+import java.util.List;
+
+/** Table self optimizing metrics */
+public class TableOptimizingMetrics {
+ /** Table is no need optimizing. */
+ public static final String STATUS_IDLE = "idle";
+
+ /** Table is need optimizing, but waiting for resource */
+ public static final String STATUS_PENDING = "pending";
+
+ /** Table is doing optimizing process planing. */
+ public static final String STATUS_PLANING = "planing";
+
+ /** Table is executing optimizing process */
+ public static final String STATUS_EXECUTING = "executing";
+
+ /** All optimizing process task is done, and process is committing. */
+ public static final String STATUS_COMMITTING = "committing";
+
+ // table optimizing status duration metrics
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_IDLE_DURATION =
+ defineGauge("table_optimizing_status_idle_duration_mills")
+ .withDescription("Duration in seconds after table be in idle status")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_PENDING_DURATION =
+ defineGauge("table_optimizing_status_pending_duration_mills")
+ .withDescription("Duration in seconds after table be in pending
status")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_PLANNING_DURATION =
+ defineGauge("table_optimizing_status_planning_duration_mills")
+ .withDescription("Duration in seconds after table be in planning
status")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_EXECUTING_DURATION =
+ defineGauge("table_optimizing_status_executing_duration_mills")
+ .withDescription("Duration in seconds after table be in executing
status")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_COMMITTING_DURATION
=
+ defineGauge("table_optimizing_status_committing_duration_mills")
+ .withDescription("Duration in seconds after table be in committing
status")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ // table optimizing process count metrics
+ public static final MetricDefine TABLE_OPTIMIZING_PROCESS_TOTAL_COUNT =
+ defineCounter("table_optimizing_process_total_count")
+ .withDescription("Count of all process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_PROCESS_FAILED_COUNT =
+ defineCounter("table_optimizing_process_failed_count")
+ .withDescription("Count of failed process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_MINOR_TOTAL_COUNT =
+ defineCounter("table_optimizing_minor_total_count")
+ .withDescription("Count of minor process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_MINOR_FAILED_COUNT =
+ defineCounter("table_optimizing_minor_failed_count")
+ .withDescription("Count of failed minor process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_MAJOR_TOTAL_COUNT =
+ defineCounter("table_optimizing_major_total_count")
+ .withDescription("Count of major process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_MAJOR_FAILED_COUNT =
+ defineCounter("table_optimizing_major_failed_count")
+ .withDescription("Count of failed major process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_FULL_TOTAL_COUNT =
+ defineCounter("table_optimizing_full_total_count")
+ .withDescription("Count of full process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_FULL_FAILED_COUNT =
+ defineCounter("table_optimizing_full_failed_count")
+ .withDescription("Count of failed full process since ams started")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ // table optimizing process status metrics
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_IN_IDLE =
+ defineGauge("table_optimizing_status_in_idle")
+ .withDescription("If currently table is in status idle")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_IN_PENDING =
+ defineGauge("table_optimizing_status_in_pending")
+ .withDescription("If currently table is in status pending")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_IN_PLANNING =
+ defineGauge("table_optimizing_status_in_planning")
+ .withDescription("If currently table is in status planning")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_IN_EXECUTING =
+ defineGauge("table_optimizing_status_in_executing")
+ .withDescription("If currently table is in status executing")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ public static final MetricDefine TABLE_OPTIMIZING_STATUS_IN_COMMITTING =
+ defineGauge("table_optimizing_status_in_committing")
+ .withDescription("If currently table is in status committing")
+ .withTags("catalog", "database", "table")
+ .build();
+
+ private final Counter processTotalCount = new Counter();
+ private final Counter processFailedCount = new Counter();
+ private final Counter minorTotalCount = new Counter();
+ private final Counter minorFailedCount = new Counter();
+ private final Counter majorTotalCount = new Counter();
+ private final Counter majorFailedCount = new Counter();
+ private final Counter fullTotalCount = new Counter();
+ private final Counter fullFailedCount = new Counter();
+
+ private final ServerTableIdentifier identifier;
+
+ private OptimizingStatus optimizingStatus;
+ private long stateSetTimestamp = System.currentTimeMillis();
+ private final List<MetricKey> registeredMetricKeys = Lists.newArrayList();
+ private MetricRegistry globalRegistry;
+
+ public TableOptimizingMetrics(ServerTableIdentifier identifier) {
+ this.identifier = identifier;
+ }
+
+ private void registerMetric(MetricRegistry registry, MetricDefine define,
Metric metric) {
+ MetricKey key =
+ registry.register(
+ define,
+ ImmutableMap.of(
+ "catalog",
+ identifier.getCatalog(),
+ "database",
+ identifier.getDatabase(),
+ "table",
+ identifier.getTableName()),
+ metric);
+ registeredMetricKeys.add(key);
+ }
+
+ public void register(MetricRegistry registry) {
+ if (globalRegistry == null) {
+ // register state duration metrics
+ registerMetric(
+ registry, TABLE_OPTIMIZING_STATUS_IDLE_DURATION, new
StatusDurationGauge(STATUS_IDLE));
+ registerMetric(
+ registry,
+ TABLE_OPTIMIZING_STATUS_PENDING_DURATION,
+ new StatusDurationGauge(STATUS_PENDING));
+ registerMetric(
+ registry,
+ TABLE_OPTIMIZING_STATUS_PLANNING_DURATION,
+ new StatusDurationGauge(STATUS_PLANING));
+ registerMetric(
+ registry,
+ TABLE_OPTIMIZING_STATUS_EXECUTING_DURATION,
+ new StatusDurationGauge(STATUS_EXECUTING));
+ registerMetric(
+ registry,
+ TABLE_OPTIMIZING_STATUS_COMMITTING_DURATION,
+ new StatusDurationGauge(STATUS_COMMITTING));
+
+ // register table in status metrics
+ registerMetric(registry, TABLE_OPTIMIZING_STATUS_IN_IDLE, new
IsInStatusGauge(STATUS_IDLE));
+ registerMetric(
+ registry, TABLE_OPTIMIZING_STATUS_IN_PENDING, new
IsInStatusGauge(STATUS_PENDING));
+ registerMetric(
+ registry, TABLE_OPTIMIZING_STATUS_IN_PLANNING, new
IsInStatusGauge(STATUS_PLANING));
+ registerMetric(
+ registry, TABLE_OPTIMIZING_STATUS_IN_EXECUTING, new
IsInStatusGauge(STATUS_EXECUTING));
+ registerMetric(
+ registry, TABLE_OPTIMIZING_STATUS_IN_COMMITTING, new
IsInStatusGauge(STATUS_COMMITTING));
+
+ // register table process count metrics
+ registerMetric(registry, TABLE_OPTIMIZING_PROCESS_TOTAL_COUNT,
processTotalCount);
+ registerMetric(registry, TABLE_OPTIMIZING_PROCESS_FAILED_COUNT,
processFailedCount);
+ registerMetric(registry, TABLE_OPTIMIZING_MINOR_TOTAL_COUNT,
minorTotalCount);
+ registerMetric(registry, TABLE_OPTIMIZING_MINOR_FAILED_COUNT,
minorFailedCount);
+ registerMetric(registry, TABLE_OPTIMIZING_MAJOR_TOTAL_COUNT,
majorTotalCount);
+ registerMetric(registry, TABLE_OPTIMIZING_MAJOR_FAILED_COUNT,
majorFailedCount);
+ registerMetric(registry, TABLE_OPTIMIZING_FULL_TOTAL_COUNT,
fullTotalCount);
+ registerMetric(registry, TABLE_OPTIMIZING_FULL_FAILED_COUNT,
fullFailedCount);
+
+ globalRegistry = registry;
+ }
+ }
+
+ public void unregister() {
+ registeredMetricKeys.forEach(globalRegistry::unregister);
+ registeredMetricKeys.clear();
+ globalRegistry = null;
+ }
+
+ /**
+ * Handle table self optimizing state change event.
+ *
+ * @param optimizingStatus new optimizing status
+ * @param stateSetTimestamp timestamp of status changed.
+ */
+ public void statusChanged(OptimizingStatus optimizingStatus, long
stateSetTimestamp) {
+ this.optimizingStatus = optimizingStatus;
+ this.stateSetTimestamp = stateSetTimestamp;
+ }
+
+ /**
+ * Handle table self optimizing process completed event.
+ *
+ * @param processType optimizing process type.
+ * @param success is optimizing process success.
+ */
+ public void processComplete(OptimizingType processType, boolean success) {
+ processTotalCount.inc();
+ Counter totalCounter = null;
+ Counter failedCounter = null;
+ switch (processType) {
+ case MINOR:
+ totalCounter = minorTotalCount;
+ failedCounter = minorFailedCount;
+ break;
+ case MAJOR:
+ totalCounter = majorTotalCount;
+ failedCounter = majorFailedCount;
+ break;
+ case FULL:
+ totalCounter = fullTotalCount;
+ failedCounter = fullFailedCount;
+ break;
+ }
+ if (totalCounter != null) {
+ totalCounter.inc();
+ }
+ if (!success && failedCounter != null) {
+ failedCounter.inc();
+ }
+ }
+
+ private String getOptimizingStatusDesc(OptimizingStatus status) {
+ switch (status) {
+ case IDLE:
+ return STATUS_IDLE;
+ case PENDING:
+ return STATUS_PENDING;
+ case PLANNING:
+ return STATUS_PLANING;
+ case FULL_OPTIMIZING:
+ case MAJOR_OPTIMIZING:
+ case MINOR_OPTIMIZING:
+ return STATUS_EXECUTING;
+ case COMMITTING:
+ return STATUS_COMMITTING;
+ default:
+ return status.name();
+ }
+ }
+
+ class StatusDurationGauge implements Gauge<Long> {
+ final String targetStatus;
+
+ StatusDurationGauge(String targetStatus) {
+ this.targetStatus = targetStatus;
+ }
+
+ @Override
+ public Long getValue() {
+ String state = getOptimizingStatusDesc(optimizingStatus);
+ if (targetStatus.equals(state)) {
+ return stateDuration();
+ }
+ return 0L;
+ }
+
+ private Long stateDuration() {
+ return System.currentTimeMillis() - stateSetTimestamp;
+ }
+ }
+
+ class IsInStatusGauge implements Gauge<Long> {
+ final String targetStatus;
+
+ IsInStatusGauge(String targetStatus) {
+ this.targetStatus = targetStatus;
+ }
+
+ @Override
+ public Long getValue() {
+ String state = getOptimizingStatusDesc(optimizingStatus);
+ if (targetStatus.equals(state)) {
+ return 1L;
+ }
+ return 0L;
+ }
+ }
+}
diff --git
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
index 29f902cb9..9e6e3507b 100644
---
a/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
+++
b/amoro-ams/amoro-ams-server/src/main/java/org/apache/amoro/server/table/TableRuntime.java
@@ -94,7 +94,7 @@ public class TableRuntime extends StatedPersistentBase {
@StateField private volatile long processId;
@StateField private volatile OptimizingEvaluator.PendingInput pendingInput;
private volatile long lastPlanTime;
- private final TableMetrics metrics;
+ private final TableOptimizingMetrics optimizingMetrics;
private final ReentrantLock blockerLock = new ReentrantLock();
protected TableRuntime(
@@ -107,7 +107,7 @@ public class TableRuntime extends StatedPersistentBase {
this.tableConfiguration = TableConfiguration.parseConfig(properties);
this.optimizerGroup =
tableConfiguration.getOptimizingConfig().getOptimizerGroup();
persistTableRuntime();
- metrics = new TableMetrics(tableIdentifier);
+ optimizingMetrics = new TableOptimizingMetrics(tableIdentifier);
}
protected TableRuntime(TableRuntimeMeta tableRuntimeMeta,
TableRuntimeHandler tableHandler) {
@@ -136,8 +136,8 @@ public class TableRuntime extends StatedPersistentBase {
? OptimizingStatus.PENDING
: tableRuntimeMeta.getTableStatus();
this.pendingInput = tableRuntimeMeta.getPendingInput();
- metrics = new TableMetrics(tableIdentifier);
- metrics.stateChanged(optimizingStatus, this.currentStatusStartTime);
+ optimizingMetrics = new TableOptimizingMetrics(tableIdentifier);
+ optimizingMetrics.statusChanged(optimizingStatus,
this.currentStatusStartTime);
}
public void recover(OptimizingProcess optimizingProcess) {
@@ -149,7 +149,7 @@ public class TableRuntime extends StatedPersistentBase {
}
public void registerMetric(MetricRegistry metricRegistry) {
- this.metrics.register(metricRegistry);
+ this.optimizingMetrics.register(metricRegistry);
}
public void dispose() {
@@ -162,7 +162,7 @@ public class TableRuntime extends StatedPersistentBase {
TableMetaMapper.class,
mapper ->
mapper.deleteOptimizingRuntime(tableIdentifier.getId())));
});
- metrics.unregister();
+ optimizingMetrics.unregister();
}
public void beginPlanning() {
@@ -273,20 +273,22 @@ public class TableRuntime extends StatedPersistentBase {
invokeConsistency(
() -> {
OptimizingStatus originalStatus = optimizingStatus;
+ OptimizingType processType = optimizingProcess.getOptimizingType();
if (success) {
lastOptimizedSnapshotId = optimizingProcess.getTargetSnapshotId();
lastOptimizedChangeSnapshotId =
optimizingProcess.getTargetChangeSnapshotId();
- if (optimizingProcess.getOptimizingType() == OptimizingType.MINOR)
{
+ if (processType == OptimizingType.MINOR) {
lastMinorOptimizingTime = optimizingProcess.getPlanTime();
- } else if (optimizingProcess.getOptimizingType() ==
OptimizingType.MAJOR) {
+ } else if (processType == OptimizingType.MAJOR) {
lastMajorOptimizingTime = optimizingProcess.getPlanTime();
- } else if (optimizingProcess.getOptimizingType() ==
OptimizingType.FULL) {
+ } else if (processType == OptimizingType.FULL) {
lastFullOptimizingTime = optimizingProcess.getPlanTime();
}
}
updateOptimizingStatus(OptimizingStatus.IDLE);
optimizingProcess = null;
persistUpdatingRuntime();
+ optimizingMetrics.processComplete(processType, success);
tableHandler.handleTableChanged(this, originalStatus);
});
}
@@ -294,7 +296,7 @@ public class TableRuntime extends StatedPersistentBase {
private void updateOptimizingStatus(OptimizingStatus status) {
this.optimizingStatus = status;
this.currentStatusStartTime = System.currentTimeMillis();
- this.metrics.stateChanged(status, currentStatusStartTime);
+ this.optimizingMetrics.statusChanged(status, currentStatusStartTime);
}
private boolean refreshSnapshots(AmoroTable<?> amoroTable) {
diff --git
a/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
b/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
index e63475f65..c24236308 100644
---
a/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
+++
b/amoro-ams/amoro-ams-server/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
@@ -226,10 +226,6 @@ public class TestOptimizingQueue extends AMSTableTestBase {
Assert.assertThrows(IllegalStateException.class,
optimizingProcess::commit);
Assert.assertEquals(OptimizingProcess.Status.SUCCESS,
optimizingProcess.getStatus());
- // 9.close
- optimizingProcess.close();
- Assert.assertEquals(OptimizingProcess.Status.CLOSED,
optimizingProcess.getStatus());
-
Assert.assertEquals(0, queue.collectTasks().size());
queue.dispose();
}
