Skip to content

Commit aa7b275

Browse files
luoluoyuyucursoragentCaideyipi
authored
Pipe: Optimize pipe logging with shared PipePeriodicalLogReducer (#17887)
* Optimize pipe logging with shared PipePeriodicalLogReducer Move PipePeriodicalLogReducer to node-commons and route ConfigNode and DataNode pipe logs through PipeLogger to rate-limit repetitive messages. * Hot-reload PipePeriodicalLogReducer settings on ConfigNode Apply pipe log reducer cache updates during ConfigNode configuration reload so pipe_periodical_log_min_interval_seconds and pipe_logger_cache_max_size_in_bytes take effect without restart. Co-authored-by: Cursor <cursoragent@cursor.com> * Remove redundant log when pipeMetaFromAgent is null in heartbeat parser. Co-authored-by: Cursor <cursoragent@cursor.com> * Fix missing i18n for pipe log messages on ConfigNode. Move hardcoded heartbeat and extractor-close logs into ManagerMessages with zh translations. Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
1 parent a440038 commit aa7b275

21 files changed

Lines changed: 287 additions & 182 deletions

File tree

‎iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,10 +63,16 @@ public final class ManagerMessages {
6363
public static final String DECREASE_REFERENCE_COUNT_FOR_SNAPSHOT_ERROR =
6464
"Decrease reference count for snapshot {} error.";
6565
public static final String DELETING_REGIONS_COSTS_MS = "Deleting regions costs {}ms";
66+
public static final String DETECTED_HISTORICAL_PIPE_COMPLETION_REPORT_FROM_DATANODE =
67+
"Detected historical pipe completion report from DataNode {} for pipe {}. remainingEventCount: {}, remainingTime: {}, completedDataNodes: {}";
6668
public static final String DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT =
6769
"Detected completion of pipe {}, static meta: {}, remove it.";
70+
public static final String ALL_DATANODES_REPORTED_HISTORICAL_PIPE_COMPLETED =
71+
"All DataNodes reported historical pipe {} completed. globalRemainingEventCount: {}, globalRemainingTime: {}, staticMeta: {}";
6872
public static final String DETECT_PIPERUNTIMECRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
6973
"Detect PipeRuntimeCriticalException {} from agent, stop pipe {}.";
74+
public static final String DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
75+
"Detect PipeRuntimeSinkCriticalException {} from agent, stop pipe {}.";
7076
public static final String ENABLE_SEPARATION_OF_POWERS_IS_NOT_SUPPORTED =
7177
"Enable separation of powers is not supported";
7278
public static final String ENDEXECUTECQ_TIME_RANGE_IS_CURRENT_TIME_IS =
@@ -113,7 +119,7 @@ public final class ManagerMessages {
113119
public static final String FAILED_TO_CLOSE_CONSUMER_IN_CONSUMER_GROUP_RESULT_STATUS =
114120
"Failed to close consumer {} in consumer group {}. Result status: {}.";
115121
public static final String FAILED_TO_CLOSE_EXTRACTOR_AFTER_FAILED_TO_INITIALIZE_EXTRACTOR =
116-
"Failed to close extractor after failed to initialize extractor. ";
122+
"Failed to close extractor after failed to initialize extractor. Ignore this exception.";
117123
public static final String FAILED_TO_CLOSE_SINK_AFTER_FAILED_TO_INITIALIZE_IT_IGNORE =
118124
"Failed to close sink after failed to initialize it. Ignore this exception.";
119125
public static final String FAILED_TO_COLLECT_COMMITCREATETABLEPLAN =

‎iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java‎

Lines changed: 10 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -63,10 +63,16 @@ public final class ManagerMessages {
6363
public static final String DECREASE_REFERENCE_COUNT_FOR_SNAPSHOT_ERROR =
6464
"Decrease reference count for snapshot {} error.";
6565
public static final String DELETING_REGIONS_COSTS_MS = "Deleting regions costs {}ms";
66+
public static final String DETECTED_HISTORICAL_PIPE_COMPLETION_REPORT_FROM_DATANODE =
67+
"检测到来自 DataNode {} 的历史 pipe 完成上报，pipe {}。remainingEventCount: {}, remainingTime: {}, completedDataNodes: {}";
6668
public static final String DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT =
67-
"Detected completion of pipe {}, static meta: {}, remove it.";
69+
"检测到 pipe {} 已完成，static meta: {}，将其移除。";
70+
public static final String ALL_DATANODES_REPORTED_HISTORICAL_PIPE_COMPLETED =
71+
"所有 DataNode 均已上报历史 pipe {} 完成。globalRemainingEventCount: {}, globalRemainingTime: {}, staticMeta: {}";
6872
public static final String DETECT_PIPERUNTIMECRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
69-
"Detect PipeRuntimeCriticalException {} from agent, stop pipe {}.";
73+
"检测到 agent 上报 PipeRuntimeCriticalException {}，停止 pipe {}。";
74+
public static final String DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE =
75+
"检测到 agent 上报 PipeRuntimeSinkCriticalException {}，停止 pipe {}。";
7076
public static final String ENABLE_SEPARATION_OF_POWERS_IS_NOT_SUPPORTED = "不支持启用权力分离";
7177
public static final String ENDEXECUTECQ_TIME_RANGE_IS_CURRENT_TIME_IS =
7278
"[EndExecuteCQ] {}, time range is [{}, {}), current time is {}";
@@ -112,9 +118,9 @@ public final class ManagerMessages {
112118
public static final String FAILED_TO_CLOSE_CONSUMER_IN_CONSUMER_GROUP_RESULT_STATUS =
113119
"Failed to close consumer {} in consumer group {}. Result status: {}.";
114120
public static final String FAILED_TO_CLOSE_EXTRACTOR_AFTER_FAILED_TO_INITIALIZE_EXTRACTOR =
115-
"Failed to close extractor after failed to initialize extractor. ";
121+
"初始化 extractor 失败后关闭 extractor 失败，忽略此异常。";
116122
public static final String FAILED_TO_CLOSE_SINK_AFTER_FAILED_TO_INITIALIZE_IT_IGNORE =
117-
"Failed to close sink after failed to initialize it. Ignore this exception.";
123+
"初始化 sink 失败后关闭 sink 失败，忽略此异常。";
118124
public static final String FAILED_TO_COLLECT_COMMITCREATETABLEPLAN =
119125
"Failed to collect CommitCreateTablePlan";
120126
public static final String FAILED_TO_COLLECT_PIPE_META_LIST_FROM_CONFIG_NODE_TASK =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525
import org.apache.iotdb.commons.conf.IoTDBConstant;
2626
import org.apache.iotdb.commons.conf.TrimProperties;
2727
import org.apache.iotdb.commons.exception.BadNodeUrlException;
28+
import org.apache.iotdb.commons.pipe.config.PipeDescriptor;
29+
import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
2830
import org.apache.iotdb.commons.schema.SchemaConstant;
2931
import org.apache.iotdb.commons.utils.NodeUrlUtils;
3032
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
@@ -826,12 +828,18 @@ public boolean isSeedConfigNode() {
826828
}
827829
}
828830

829-
public void loadHotModifiedProps(TrimProperties properties) {
831+
public void loadHotModifiedProps(TrimProperties properties) throws IOException {
830832
ConfigurationFileUtils.updateAppliedProperties(properties, true);
831833
Optional.ofNullable(properties.getProperty(IoTDBConstant.CLUSTER_NAME))
832834
.ifPresent(conf::setClusterName);
833835
Optional.ofNullable(properties.getProperty("enable_topology_probing"))
834836
.ifPresent(v -> conf.setEnableTopologyProbing(Boolean.parseBoolean(v)));
837+
loadPipeHotModifiedProp(properties);
838+
}
839+
840+
private void loadPipeHotModifiedProp(TrimProperties properties) throws IOException {
841+
PipeDescriptor.loadPipeProps(commonDescriptor.getConfig(), properties, true);
842+
PipePeriodicalLogReducer.update();
835843
}
836844

837845
public static ConfigNodeDescriptor getInstance() {

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/runtime/PipeConfigNodeRuntimeAgent.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
2727
import org.apache.iotdb.commons.pipe.config.PipeConfig;
2828
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
29+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
30+
import org.apache.iotdb.commons.pipe.resource.log.PipePeriodicalLogReducer;
2931
import org.apache.iotdb.commons.service.IService;
3032
import org.apache.iotdb.commons.service.ServiceType;
3133
import org.apache.iotdb.confignode.i18n.ManagerMessages;
@@ -56,6 +58,7 @@ public class PipeConfigNodeRuntimeAgent implements IService {
5658
@Override
5759
public synchronized void start() {
5860
PipeConfig.getInstance().printAllConfigs();
61+
PipeLogger.setLogger(PipePeriodicalLogReducer::log);
5962

6063
// PipeTasks will not be started here and will be started by "HandleLeaderChange"
6164
// procedure when the consensus layer notify leader ready
@@ -142,19 +145,21 @@ public void report(final EnrichedEvent event, final PipeRuntimeException pipeRun
142145
if (event.getPipeTaskMeta() != null) {
143146
report(event.getPipeTaskMeta(), pipeRuntimeException);
144147
} else {
145-
LOGGER.warn(
146-
ManagerMessages.ATTEMPT_TO_REPORT_PIPE_EXCEPTION_TO_A_NULL_PIPETASKMETA,
147-
pipeRuntimeException);
148+
PipeLogger.log(
149+
LOGGER::warn,
150+
pipeRuntimeException,
151+
ManagerMessages.ATTEMPT_TO_REPORT_PIPE_EXCEPTION_TO_A_NULL_PIPETASKMETA);
148152
}
149153
}
150154

151155
private void report(
152156
final PipeTaskMeta pipeTaskMeta, final PipeRuntimeException pipeRuntimeException) {
153-
LOGGER.warn(
157+
PipeLogger.log(
158+
LOGGER::warn,
159+
pipeRuntimeException,
154160
ManagerMessages.REPORT_PIPERUNTIMEEXCEPTION_TO_LOCAL_PIPETASKMETA_EXCEPTION_MESSAGE,
155161
pipeTaskMeta,
156-
pipeRuntimeException.getMessage(),
157-
pipeRuntimeException);
162+
pipeRuntimeException.getMessage());
158163

159164
pipeTaskMeta.trackExceptionMessage(pipeRuntimeException);
160165

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/agent/task/PipeConfigNodeSubtask.java‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
3333
import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
3434
import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
35+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
3536
import org.apache.iotdb.confignode.i18n.ManagerMessages;
3637
import org.apache.iotdb.confignode.manager.pipe.agent.PipeConfigNodeAgent;
3738
import org.apache.iotdb.confignode.manager.pipe.metric.sink.PipeConfigRegionSinkMetrics;
@@ -105,10 +106,10 @@ private void initSource(final Map<String, String> sourceAttributes) throws Excep
105106
try {
106107
source.close();
107108
} catch (Exception closeException) {
108-
LOGGER.warn(
109-
ManagerMessages.FAILED_TO_CLOSE_EXTRACTOR_AFTER_FAILED_TO_INITIALIZE_EXTRACTOR
110-
+ "Ignore this exception.",
111-
closeException);
109+
PipeLogger.log(
110+
LOGGER::warn,
111+
closeException,
112+
ManagerMessages.FAILED_TO_CLOSE_EXTRACTOR_AFTER_FAILED_TO_INITIALIZE_EXTRACTOR);
112113
}
113114
throw e;
114115
}
@@ -154,9 +155,11 @@ private void initSink(final Map<String, String> sinkAttributes) throws Exception
154155
try {
155156
outputPipeSink.close();
156157
} catch (final Exception closeException) {
157-
LOGGER.warn(
158+
PipeLogger.log(
159+
LOGGER::warn,
160+
closeException,
158161
ManagerMessages.FAILED_TO_CLOSE_SINK_AFTER_FAILED_TO_INITIALIZE_IT_IGNORE,
159-
closeException);
162+
closeException.getMessage());
160163
}
161164
throw e;
162165
}
@@ -208,19 +211,19 @@ public void close() {
208211
try {
209212
source.close();
210213
} catch (final Exception e) {
211-
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEEXTRACTOR, e);
214+
PipeLogger.log(LOGGER::info, e, ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEEXTRACTOR);
212215
}
213216

214217
try {
215218
processor.close();
216219
} catch (final Exception e) {
217-
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEPROCESSOR, e);
220+
PipeLogger.log(LOGGER::info, e, ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPEPROCESSOR);
218221
}
219222

220223
try {
221224
outputPipeSink.close();
222225
} catch (final Exception e) {
223-
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPECONNECTOR, e);
226+
PipeLogger.log(LOGGER::info, e, ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPECONNECTOR);
224227
} finally {
225228
// Should be after connector.close()
226229
super.close();

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/PipeMetaSyncer.java‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.apache.iotdb.commons.concurrent.ThreadName;
2525
import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
2626
import org.apache.iotdb.commons.pipe.config.PipeConfig;
27+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
2728
import org.apache.iotdb.confignode.i18n.ManagerMessages;
2829
import org.apache.iotdb.confignode.manager.ConfigManager;
2930
import org.apache.iotdb.confignode.manager.ProcedureManager;
@@ -95,7 +96,8 @@ private synchronized void sync() {
9596
isLastPipeSyncSuccessful = false;
9697

9798
if (configManager.getPipeManager().getPipeTaskCoordinator().isLocked()) {
98-
LOGGER.warn(
99+
PipeLogger.log(
100+
LOGGER::warn,
99101
ManagerMessages.PIPETASKCOORDINATORLOCK_IS_HELD_BY_ANOTHER_THREAD_SKIP_THIS_ROUND_OF_2);
100102
return;
101103
}
@@ -109,7 +111,9 @@ private synchronized void sync() {
109111
== PipeConfig.getInstance().getPipeMetaSyncerAutoRestartPipeCheckIntervalRound()) {
110112
somePipesNeedRestarting = autoRestartWithLock();
111113
if (somePipesNeedRestarting) {
112-
LOGGER.info(ManagerMessages.SOME_PIPES_NEED_RESTARTING_WILL_RESTART_THEM_AFTER_THIS_SYNC);
114+
PipeLogger.log(
115+
LOGGER::info,
116+
ManagerMessages.SOME_PIPES_NEED_RESTARTING_WILL_RESTART_THEM_AFTER_THIS_SYNC);
113117
}
114118
pipeAutoRestartRoundCounter.set(0);
115119
}
@@ -130,19 +134,22 @@ private synchronized void sync() {
130134
if (handleMetaChangeStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
131135
successfulSync = true;
132136
} else {
133-
LOGGER.warn(
137+
PipeLogger.log(
138+
LOGGER::warn,
134139
ManagerMessages.FAILED_TO_HANDLE_PIPE_META_CHANGE_RESULT_STATUS,
135140
handleMetaChangeStatus);
136141
}
137142
}
138143

139144
if (successfulSync) {
140-
LOGGER.info(
145+
PipeLogger.log(
146+
LOGGER::info,
141147
ManagerMessages.AFTER_THIS_SUCCESSFUL_SYNC_IF_PIPETASKINFO_IS_EMPTY_DURING_THIS);
142148
isLastPipeSyncSuccessful = true;
143149
}
144150
} else {
145-
LOGGER.warn(ManagerMessages.FAILED_TO_SYNC_PIPE_META_RESULT_STATUS, metaSyncStatus);
151+
PipeLogger.log(
152+
LOGGER::warn, ManagerMessages.FAILED_TO_SYNC_PIPE_META_RESULT_STATUS, metaSyncStatus);
146153
}
147154
}
148155

@@ -158,7 +165,8 @@ private boolean autoRestartWithLock() {
158165
final AtomicReference<PipeTaskInfo> pipeTaskInfo =
159166
configManager.getPipeManager().getPipeTaskCoordinator().tryLock();
160167
if (pipeTaskInfo == null) {
161-
LOGGER.warn(ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_AUTO_RESTART_PIPE_TASK);
168+
PipeLogger.log(
169+
LOGGER::warn, ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_AUTO_RESTART_PIPE_TASK);
162170
return false;
163171
}
164172
try {
@@ -176,15 +184,17 @@ private void checkAndRepairConsensusPipes() {
176184
.getRegionMaintainHandler()
177185
.checkAndRepairConsensusPipes();
178186
} catch (Exception e) {
179-
LOGGER.warn(ManagerMessages.FAILED_TO_CHECK_AND_REPAIR_CONSENSUS_PIPES, e);
187+
PipeLogger.log(LOGGER::warn, e, ManagerMessages.FAILED_TO_CHECK_AND_REPAIR_CONSENSUS_PIPES);
180188
}
181189
}
182190

183191
private boolean handleSuccessfulRestartWithLock() {
184192
final AtomicReference<PipeTaskInfo> pipeTaskInfo =
185193
configManager.getPipeManager().getPipeTaskCoordinator().tryLock();
186194
if (pipeTaskInfo == null) {
187-
LOGGER.warn(ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_HANDLING_SUCCESSFUL_RESTART);
195+
PipeLogger.log(
196+
LOGGER::warn,
197+
ManagerMessages.FAILED_TO_ACQUIRE_PIPE_LOCK_FOR_HANDLING_SUCCESSFUL_RESTART);
188198
return false;
189199
}
190200
try {

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java‎

Lines changed: 25 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
3131
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTemporaryMetaInCoordinator;
3232
import org.apache.iotdb.commons.pipe.config.PipeConfig;
33+
import org.apache.iotdb.commons.pipe.resource.log.PipeLogger;
3334
import org.apache.iotdb.confignode.consensus.response.pipe.task.PipeTableResp;
3435
import org.apache.iotdb.confignode.i18n.ManagerMessages;
3536
import org.apache.iotdb.confignode.manager.ConfigManager;
@@ -93,7 +94,8 @@ synchronized void parseHeartbeat(final int nodeId, final PipeHeartbeat pipeHeart
9394
final AtomicReference<PipeTaskInfo> pipeTaskInfo =
9495
configManager.getPipeManager().getPipeTaskCoordinator().tryLock();
9596
if (pipeTaskInfo == null) {
96-
LOGGER.warn(
97+
PipeLogger.log(
98+
LOGGER::warn,
9799
ManagerMessages.FAILED_TO_ACQUIRE_LOCK_WHEN_PARSEHEARTBEAT_FROM_NODE_ID,
98100
nodeId);
99101
return;
@@ -127,8 +129,10 @@ private int getExpectedHeartbeatNodeCount() {
127129
configManager.getNodeManager().getRegisteredDataNodeCount()
128130
+ (PipeConfig.getInstance().isSeperatedPipeHeartbeatEnabled() ? 1 : 0);
129131
if (expectedNodeCount <= 0) {
130-
LOGGER.warn(
131-
ManagerMessages.EXPECTED_PIPE_HEARTBEAT_NODE_COUNT_IS_FALLBACK_TO_1, expectedNodeCount);
132+
PipeLogger.log(
133+
LOGGER::warn,
134+
ManagerMessages.EXPECTED_PIPE_HEARTBEAT_NODE_COUNT_IS_FALLBACK_TO_1,
135+
expectedNodeCount);
132136
return 1;
133137
}
134138
return expectedNodeCount;
@@ -142,10 +146,6 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
142146
final PipeStaticMeta staticMeta = pipeMetaFromCoordinator.getStaticMeta();
143147
final PipeMeta pipeMetaFromAgent = pipeHeartbeat.getPipeMeta(staticMeta);
144148
if (pipeMetaFromAgent == null) {
145-
LOGGER.info(
146-
ManagerMessages.PIPERUNTIMECOORDINATOR_MEETS_ERROR_IN_UPDATING_PIPEMETAKEEPER
147-
+ "pipeMetaFromAgent is null, pipeMetaFromCoordinator: {}",
148-
pipeMetaFromCoordinator);
149149
continue;
150150
}
151151

@@ -157,8 +157,9 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
157157
if (Boolean.TRUE.equals(isPipeCompletedFromAgent)) {
158158

159159
temporaryMeta.markDataNodeCompleted(nodeId);
160-
LOGGER.info(
161-
"Detected historical pipe completion report from DataNode {} for pipe {}. remainingEventCount: {}, remainingTime: {}, completedDataNodes: {}",
160+
PipeLogger.log(
161+
LOGGER::info,
162+
ManagerMessages.DETECTED_HISTORICAL_PIPE_COMPLETION_REPORT_FROM_DATANODE,
162163
nodeId,
163164
staticMeta.getPipeName(),
164165
pipeHeartbeat.getRemainingEventCount(staticMeta),
@@ -169,14 +170,16 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
169170
configManager.getNodeManager().getRegisteredDataNodeLocations().keySet();
170171
uncompletedDataNodeIds.removeAll(temporaryMeta.getCompletedDataNodeIds());
171172
if (uncompletedDataNodeIds.isEmpty()) {
172-
LOGGER.info(
173-
"All DataNodes reported historical pipe {} completed. globalRemainingEventCount: {}, globalRemainingTime: {}, staticMeta: {}",
173+
PipeLogger.log(
174+
LOGGER::info,
175+
ManagerMessages.ALL_DATANODES_REPORTED_HISTORICAL_PIPE_COMPLETED,
174176
staticMeta.getPipeName(),
175177
temporaryMeta.getGlobalRemainingEvents(),
176178
temporaryMeta.getGlobalRemainingTime(),
177179
staticMeta);
178180
pipeTaskInfo.get().removePipeMeta(staticMeta.getPipeName());
179-
LOGGER.info(
181+
PipeLogger.log(
182+
LOGGER::info,
180183
ManagerMessages.DETECTED_COMPLETION_OF_PIPE_STATIC_META_REMOVE_IT,
181184
staticMeta.getPipeName(),
182185
staticMeta);
@@ -267,7 +270,9 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
267270
needWriteConsensusOnConfigNodes.set(true);
268271
needPushPipeMetaToDataNodes.set(false);
269272

270-
LOGGER.warn(
273+
PipeLogger.log(
274+
LOGGER::warn,
275+
exception,
271276
ManagerMessages.DETECT_PIPERUNTIMECRITICALEXCEPTION_FROM_AGENT_STOP_PIPE,
272277
exception,
273278
pipeName);
@@ -296,11 +301,13 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
296301
needWriteConsensusOnConfigNodes.set(true);
297302
needPushPipeMetaToDataNodes.set(false);
298303

299-
LOGGER.warn(
300-
String.format(
301-
"Detect PipeRuntimeConnectorCriticalException %s "
302-
+ "from agent, stop pipe %s.",
303-
exception, pipeName));
304+
PipeLogger.log(
305+
LOGGER::warn,
306+
exception,
307+
ManagerMessages
308+
.DETECT_PIPERUNTIMESINKCRITICALEXCEPTION_FROM_AGENT_STOP_PIPE,
309+
exception,
310+
pipeName);
304311
});
305312
}
306313
}

0 commit comments

Comments
 (0)