apache · dockerzhang · Jul 3, 2024 · Jul 2, 2024
diff --git a/...src/main/java/org/apache/inlong/sort/standalone/config/holder/CommonPropertiesHolder.java b/...src/main/java/org/apache/inlong/sort/standalone/config/holder/CommonPropertiesHolder.java
@@ -179,6 +179,14 @@ public static Integer getInteger(String key, Integer defaultValue) {
  return defaultValue;
  }
 
+ public static Boolean getBoolean(String key, Boolean defaultValue) {
+ String value = get().get(key);
+ if (value != null) {
+ return Boolean.valueOf(value.trim());
+ }
+ return defaultValue;
+ }
+
  /**
  * Gets value mapped to key, returning null if unmapped.
  * <p>
@@ -220,4 +228,8 @@ public static AckPolicy getAckPolicy() {
  return ackPolicy;
  }
 
+ public static boolean useUnifiedConfiguration() {
+ return getBoolean(KEY_USE_UNIFIED_CONFIGURATION, false);
+ }
+
 }
diff --git a/...in/java/org/apache/inlong/sort/standalone/config/loader/SortConfigQueryConsumeConfig.java b/...in/java/org/apache/inlong/sort/standalone/config/loader/SortConfigQueryConsumeConfig.java
@@ -41,12 +41,6 @@
 @Slf4j
 public class SortConfigQueryConsumeConfig implements QueryConsumeConfig {
 
- private List<InLongTopic> subscribedTopic = new ArrayList<>();
-
- public void reload() {
-
- }
-
  @Override
  public ConsumeConfig queryCurrentConsumeConfig(String sortTaskId) {
  TaskConfig taskConfig = SortConfigHolder.getTaskConfig(sortTaskId);

diff --git a/...andalone-common/src/main/java/org/apache/inlong/sort/standalone/config/pojo/IdConfig.java b/...andalone-common/src/main/java/org/apache/inlong/sort/standalone/config/pojo/IdConfig.java
@@ -0,0 +1,33 @@
+/*
+ * 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.inlong.sort.standalone.config.pojo;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+@NoArgsConstructor
+@AllArgsConstructor
+@Data
+@SuperBuilder
+public abstract class IdConfig {
+
+ protected String inlongGroupId;
+ protected String inlongStreamId;
+}
diff --git a/...mon/src/main/java/org/apache/inlong/sort/standalone/metrics/SortConfigMetricListener.java b/...mon/src/main/java/org/apache/inlong/sort/standalone/metrics/SortConfigMetricListener.java
@@ -21,8 +21,6 @@
 
 import org.apache.flume.conf.Configurable;
 
-import java.util.Collection;
-
 public interface SortConfigMetricListener extends Configurable {
 
  void reportOffline(SortConfig sortConfig);
@@ -34,6 +32,11 @@ public interface SortConfigMetricListener extends Configurable {
  void reportCheckFail();
  void reportRequestNoUpdate();
  void reportRequestUpdate();
- void reportMissInSortClusterConfig(Collection<String> dataflows);
- void reportMissInSortConfig(Collection<String> dataflows);
+ void reportMissInSortClusterConfig(String cluster, String task, String group, String stream);
+ void reportMissInSortConfig(String cluster, String task, String group, String stream);
+ void reportClusterDiff(String cluster, String task, String group, String stream);
+ void reportSourceDiff(String sortClusterName, String sortTaskName, String topic, String mqClusterName);
+ void reportSourceMissInSortClusterConfig(String sortClusterName, String sortTaskName,
+ String topic, String mqClusterName);
+ void reportSourceMissInSortConfig(String sortClusterName, String sortTaskName, String topic, String mqClusterName);
 }
diff --git a/...mon/src/main/java/org/apache/inlong/sort/standalone/metrics/SortConfigMetricReporter.java b/...mon/src/main/java/org/apache/inlong/sort/standalone/metrics/SortConfigMetricReporter.java
@@ -19,8 +19,10 @@
 
 import org.apache.inlong.common.pojo.sort.SortConfig;
 import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.ClassUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flume.Context;
@@ -120,11 +122,57 @@ public static void reportRequestUpdate() {
  listeners.forEach(SortConfigMetricListener::reportRequestUpdate);
  }
 
- public static void reportMissInSortClusterConfig(Collection<String> dataflows) {
- listeners.forEach(listener -> listener.reportMissInSortClusterConfig(dataflows));
+ public static void reportClusterDiff(
+ String sortClusterName,
+ String sortTaskName,
+ Map<String, ? extends IdConfig> fromTaskConfig,
+ Map<String, ? extends IdConfig> fromSortTaskConfig) {
+ Collection<String> intersection = CollectionUtils.intersection(fromTaskConfig.keySet(),
+ fromSortTaskConfig.keySet());
+ List<IdConfig> diff = intersection.stream()
+ .filter(k -> !fromTaskConfig.get(k).equals(fromSortTaskConfig.get(k)))
+ .map(fromSortTaskConfig::get)
+ .collect(Collectors.toList());
+ // report diff
+ diff.forEach(idConfig -> {
+ listeners.forEach(listener -> listener.reportClusterDiff(sortClusterName, sortTaskName,
+ idConfig.getInlongGroupId(), idConfig.getInlongStreamId()));
+ });
+
+ // report miss in sort cluster config
+ fromTaskConfig.forEach((k, v) -> {
+ if (!intersection.contains(k)) {
+ listeners.forEach(listener -> listener.reportMissInSortClusterConfig(sortClusterName, sortTaskName,
+ v.getInlongGroupId(), v.getInlongStreamId()));
+ }
+ });
+
+ // report miss in sort config
+ fromSortTaskConfig.forEach((k, v) -> {
+ if (!intersection.contains(k)) {
+ listeners.forEach(listener -> listener.reportMissInSortConfig(sortClusterName, sortTaskName,
+ v.getInlongGroupId(), v.getInlongStreamId()));
+ }
+ });
+ }
+
+ public static void reportSourceDiff(
+ String sortClusterName, String sortTaskName,
+ String topic, String mqClusterName) {
+ listeners.forEach(listener -> listener.reportSourceDiff(sortClusterName, sortTaskName, topic, mqClusterName));
+ }
+
+ public static void reportSourceMissInSortClusterConfig(
+ String sortClusterName, String sortTaskName,
+ String topic, String mqClusterName) {
+ listeners.forEach(listener -> listener.reportSourceMissInSortClusterConfig(sortClusterName, sortTaskName, topic,
+ mqClusterName));
  }
 
- public static void reportMissInSortConfig(Collection<String> dataflows) {
- listeners.forEach(listener -> listener.reportMissInSortConfig(dataflows));
+ public static void reportSourceMissInSortConfig(
+ String sortClusterName, String sortTaskName,
+ String topic, String mqClusterName) {
+ listeners.forEach(
+ listener -> listener.reportSourceMissInSortConfig(sortClusterName, sortTaskName, topic, mqClusterName));
  }
 }
diff --git a/...t-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/SinkContext.java b/...t-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/SinkContext.java
@@ -73,8 +73,7 @@ public SinkContext(String sinkName, Context context, Channel channel) {
  this.processInterval = sinkContext.getInteger(KEY_PROCESSINTERVAL, 100);
  this.reloadInterval = sinkContext.getLong(KEY_RELOADINTERVAL, 60000L);
  this.metricItemSet = new SortMetricItemSet(sinkName);
- this.unifiedConfiguration = sinkContext.getBoolean(CommonPropertiesHolder.KEY_USE_UNIFIED_CONFIGURATION,
- false);
+ this.unifiedConfiguration = CommonPropertiesHolder.useUnifiedConfiguration();
  MetricRegister.register(this.metricItemSet);
  }
 

diff --git a/...andalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsIdConfig.java b/...andalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsIdConfig.java
@@ -21,28 +21,27 @@
 import org.apache.inlong.common.pojo.sort.dataflow.field.FieldConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.sink.ClsSinkConfig;
 import org.apache.inlong.common.pojo.sort.node.ClsNodeConfig;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
 
 import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 import java.util.List;
 import java.util.stream.Collectors;
 
 /**
  * Cls config of each uid.
  */
+@EqualsAndHashCode(callSuper = true)
 @Data
-@Builder
 @NoArgsConstructor
 @AllArgsConstructor
-@EqualsAndHashCode
-public class ClsIdConfig {
+@SuperBuilder
+public class ClsIdConfig extends IdConfig {
 
- private String inlongGroupId;
- private String inlongStreamId;
  private String separator = "|";
  private String endpoint;
  private String secretId;

diff --git a/...alone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsSinkContext.java b/...alone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsSinkContext.java
@@ -26,6 +26,7 @@
 import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
 import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
 import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
 import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
 import org.apache.inlong.sort.standalone.sink.SinkContext;
@@ -117,6 +118,7 @@ public void reload() {
 
  Map<String, ClsIdConfig> fromTaskConfig = reloadIdParamsFromTaskConfig(taskConfig, clsNodeConfig);
  Map<String, ClsIdConfig> fromSortTaskConfig = reloadIdParamsFromSortTaskConfig(sortTaskConfig);
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
  idConfigMap = unifiedConfiguration ? fromTaskConfig : fromSortTaskConfig;
  this.reloadClients(idConfigMap);
  this.reloadHandler();

diff --git a/...source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/EsIdConfig.java b/...source/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/EsIdConfig.java
@@ -20,22 +20,25 @@
 import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.field.FieldConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.sink.EsSinkConfig;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
 
 import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.List;
 import java.util.stream.Collectors;
 
+@EqualsAndHashCode(callSuper = true)
 @Data
-@Builder
 @NoArgsConstructor
 @AllArgsConstructor
-public class EsIdConfig {
+@SuperBuilder
+public class EsIdConfig extends IdConfig {
 
  public static final String PATTERN_DAY = "{yyyyMMdd}";
  public static final String PATTERN_HOUR = "{yyyyMMddHH}";
@@ -62,8 +65,6 @@ protected SimpleDateFormat initialValue() {
  }
  };
 
- private String inlongGroupId;
- private String inlongStreamId;
  private String separator = "|";
  private String indexNamePattern;
  private String fieldNames;

diff --git a/...rce/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/EsSinkContext.java b/...rce/src/main/java/org/apache/inlong/sort/standalone/sink/elasticsearch/EsSinkContext.java
@@ -26,6 +26,7 @@
 import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
 import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
 import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
 import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
 import org.apache.inlong.sort.standalone.sink.SinkContext;
@@ -159,6 +160,7 @@ public void reload() {
  idConfigMap = fromSortTaskConfig;
  reloadClientsFromSortTaskConfig(sortTaskConfig);
  }
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
  // log
  LOG.info("end to get SortTaskConfig:taskName:{}:newIdConfigMap:{}", taskName,
  objectMapper.writeValueAsString(idConfigMap));

diff --git a/...rc/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaFederationSinkContext.java b/...rc/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaFederationSinkContext.java
@@ -26,6 +26,7 @@
 import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
 import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
 import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
 import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
 import org.apache.inlong.sort.standalone.sink.SinkContext;
@@ -83,6 +84,7 @@ public void reload() {
 
  Map<String, KafkaIdConfig> fromTaskConfig = fromTaskConfig(taskConfig);
  Map<String, KafkaIdConfig> fromSortTaskConfig = fromSortTaskConfig(sortTaskConfig);
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
  this.idConfigMap = unifiedConfiguration ? fromTaskConfig : fromSortTaskConfig;
  } catch (Throwable e) {
  LOG.error(e.getMessage(), e);

diff --git a/...lone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java b/...lone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/KafkaIdConfig.java
@@ -20,28 +20,29 @@
 import org.apache.inlong.common.enums.DataTypeEnum;
 import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.sink.KafkaSinkConfig;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
 import org.apache.inlong.sort.standalone.utils.Constants;
 
 import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 import java.util.Map;
 
+@EqualsAndHashCode(callSuper = true)
 @Data
-@Builder
 @NoArgsConstructor
 @AllArgsConstructor
-public class KafkaIdConfig {
+@SuperBuilder
+public class KafkaIdConfig extends IdConfig {
 
  public static final String KEY_DATA_TYPE = "dataType";
  public static final String KEY_SEPARATOR = "separator";
  public static final String DEFAULT_SEPARATOR = "|";
 
- private String inlongGroupId;
- private String inlongStreamId;
  private String uid;
  private String separator = "|";
  private String topic;

diff --git a/.../main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java b/.../main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java
@@ -26,6 +26,7 @@
 import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder;
 import org.apache.inlong.sort.standalone.config.holder.v2.SortConfigHolder;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
+import org.apache.inlong.sort.standalone.metrics.SortConfigMetricReporter;
 import org.apache.inlong.sort.standalone.metrics.SortMetricItem;
 import org.apache.inlong.sort.standalone.metrics.audit.AuditUtils;
 import org.apache.inlong.sort.standalone.sink.SinkContext;
@@ -77,6 +78,7 @@ public void reload() {
 
  Map<String, PulsarIdConfig> fromTaskConfig = fromTaskConfig(taskConfig);
  Map<String, PulsarIdConfig> fromSortTaskConfig = fromSortTaskConfig(sortTaskConfig);
+ SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig);
  idConfigMap = unifiedConfiguration ? fromTaskConfig : fromSortTaskConfig;
  } catch (Throwable e) {
  LOG.error(e.getMessage(), e);

diff --git a/...ne-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java b/...ne-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java
@@ -20,30 +20,31 @@
 import org.apache.inlong.common.enums.DataTypeEnum;
 import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig;
 import org.apache.inlong.common.pojo.sort.dataflow.sink.PulsarSinkConfig;
+import org.apache.inlong.sort.standalone.config.pojo.IdConfig;
 import org.apache.inlong.sort.standalone.config.pojo.InlongId;
 import org.apache.inlong.sort.standalone.utils.Constants;
 
 import lombok.AllArgsConstructor;
-import lombok.Builder;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
 
 import java.util.Map;
 
+@EqualsAndHashCode(callSuper = true)
 @Data
-@Builder
 @NoArgsConstructor
 @AllArgsConstructor
-public class PulsarIdConfig {
+@SuperBuilder
+public class PulsarIdConfig extends IdConfig {
 
  public static final String KEY_DATA_TYPE = "dataType";
  public static final String KEY_SEPARATOR = "separator";
  public static final String DEFAULT_SEPARATOR = "|";
 
  private static final String DEFAULT_INLONG_STREAM = "1";
 
- private String inlongGroupId;
- private String inlongStreamId;
  private String uid;
  private String separator = "|";
  private String topic;