refactor implementation by Jark

Author: wuchong

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -1735,14 +1735,6 @@ public class ConfigOptions {
                                     + "the CoordinatorServer) it is advisable to use a port range "
                                     + "like 9250-9260.");
 
-    public static final ConfigOption<Duration> METRICS_MANAGER_INACTIVE_EXPIRATION_TIME =
-            key("metrics.manager.inactive-expiration-time")
-                    .durationType()
-                    .defaultValue(Duration.ofHours(1))
-                    .withDescription(
-                            "The time to wait an inactive metric to be expired."
-                                    + "This is not effective for permanent metric but for temporary metric. Mostly user level metric.");
-
     // ------------------------------------------------------------------------
     //  ConfigOptions for jmx reporter
     // ------------------------------------------------------------------------

fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java

@@ -77,8 +77,8 @@ public class MetricNames {
     // --------------------------------------------------------------------------------------------
     // metrics for user
     // --------------------------------------------------------------------------------------------
-    public static final String BYTES_IN_COUNT = "bytesInCount";
-    public static final String BYTES_OUT_COUNT = "bytesOutCount";
+    public static final String BYTES_IN = "bytesIn";
+    public static final String BYTES_OUT = "bytesOut";
 
     // --------------------------------------------------------------------------------------------
     // metrics for table

fluss-common/src/main/java/org/apache/fluss/metrics/groups/AbstractMetricGroup.java

@@ -73,9 +73,6 @@ public abstract class AbstractMetricGroup implements MetricGroup {
     /** The registry that this metrics group belongs to. */
     protected final MetricRegistry registry;
 
-    /** The last record time for the group. */
-    protected volatile long lastRecordTime;
-
     /** All metrics that are directly contained in this group. */
     private final Map<String, Metric> metrics = new HashMap<>();
 
@@ -104,7 +101,6 @@ public AbstractMetricGroup(
         this.scopeComponents = checkNotNull(scope);
         this.parent = parent;
         this.logicalScopeStrings = new String[registry.getNumberReporters()];
-        this.lastRecordTime = System.currentTimeMillis();
     }
 
     @Override
@@ -199,10 +195,6 @@ protected void putVariables(Map<String, String> variables) {}
      */
     protected abstract String getGroupName(CharacterFilter filter);
 
-    public long getLastRecordTime() {
-        return lastRecordTime;
-    }
-
     // ------------------------------------------------------------------------
     //  Closing
     // ------------------------------------------------------------------------

fluss-common/src/main/java/org/apache/fluss/metrics/utils/MetricGroupUtils.java

@@ -19,8 +19,6 @@
 
 import org.apache.fluss.metrics.groups.MetricGroup;
 
-import java.util.Arrays;
-
 /** Utils for {@link MetricGroup}. */
 public class MetricGroupUtils {
 
@@ -32,8 +30,4 @@ public static String[] makeScope(MetricGroup parentGroup, String... scopes) {
         System.arraycopy(scopes, 0, parts, parentComponents.length, scopes.length);
         return parts;
     }
-
-    public static String getScopeName(MetricGroup parentGroup, String... scopes) {
-        return Arrays.toString(makeScope(parentGroup, scopes));
-    }
 }

fluss-server/src/main/java/org/apache/fluss/server/entity/UserContext.java

@@ -20,12 +20,14 @@
 
 import org.apache.fluss.security.acl.FlussPrincipal;
 
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
 /** The context information of user who writes or reads table. */
 public class UserContext {
     private final FlussPrincipal principal;
 
     public UserContext(FlussPrincipal principal) {
-        this.principal = principal;
+        this.principal = checkNotNull(principal);
     }
 
     public FlussPrincipal getPrincipal() {

fluss-server/src/main/java/org/apache/fluss/server/metrics/MetricManager.java

@@ -84,11 +84,9 @@ public static TabletServerMetricGroup createTabletServerGroup(
             String clusterId,
             @Nullable String rack,
             String hostname,
-            int serverId,
-            MetricManager metricManager) {
+            int serverId) {
         TabletServerMetricGroup tabletServerMetricGroup =
-                new TabletServerMetricGroup(
-                        registry, clusterId, rack, hostname, serverId, metricManager);
+                new TabletServerMetricGroup(registry, clusterId, rack, hostname, serverId);
         createAndInitializeStatusMetricGroup(tabletServerMetricGroup);
         return tabletServerMetricGroup;
     }