Skip to content

Commit ed90a3b

Browse files
CaideyipiCaideyipi
committed
Merge branch 'optimize-kernel-log-noise' of https://github.com/Caideyipi/iotdb into optimize-kernel-log-noise
2 parents 072ec5a + 24ea88e commit ed90a3b

29 files changed

Lines changed: 174 additions & 175 deletions

File tree

‎iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeProcessor.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@
3030
/**
3131
* {@link PipeProcessor}
3232
*
33-
* <p>{@link PipeProcessor} is used to filter and transform the {@link Event} formed by the {@link
34-
* PipeExtractor}.
33+
* <p>{@link PipeProcessor} is used to filter and transform the {@link Event} supplied by the {@link
34+
* PipeSource}.
3535
*
3636
* <p>The lifecycle of a {@link PipeProcessor} is as follows:
3737
*
@@ -44,15 +44,14 @@
4444
* to config the runtime behavior of the {@link PipeProcessor}.
4545
* <li>While the collaboration task is in progress:
4646
* <ul>
47-
* <li>{@link PipeExtractor} captures the {@link Event}s and wraps them into three types of
47+
* <li>{@link PipeSource} captures the {@link Event}s and wraps them into three types of
4848
* {@link Event} instances.
4949
* <li>{@link PipeProcessor} processes the {@link Event} and then passes them to the {@link
50-
* PipeConnector}. The following 3 methods will be called: {@link
50+
* PipeSink}. The following 3 methods will be called: {@link
5151
* PipeProcessor#process(TabletInsertionEvent, EventCollector)}, {@link
5252
* PipeProcessor#process(TsFileInsertionEvent, EventCollector)} and {@link
5353
* PipeProcessor#process(Event, EventCollector)}.
54-
* <li>{@link PipeConnector} serializes the {@link Event}s into binaries and send them to
55-
* sinks.
54+
* <li>{@link PipeSink} serializes the {@link Event}s into binaries and sends them to sinks.
5655
* </ul>
5756
* <li>When the collaboration task is cancelled (the `DROP PIPE` command is executed), the {@link
5857
* PipeProcessor#close() } method will be called.

‎iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/PipeSource.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -34,9 +34,10 @@
3434
* <p>The lifecycle of a {@link PipeSource} is as follows:
3535
*
3636
* <ul>
37-
* <li>When a collaboration task is created, the KV pairs of `WITH EXTRACTOR` clause in SQL are
38-
* parsed and the validation method {@link PipeSource#validate(PipeParameterValidator)} will
39-
* be called to validate the {@link PipeParameters}.
37+
* <li>When a collaboration task is created, the KV pairs of `WITH SOURCE` (or the legacy `WITH
38+
* EXTRACTOR`) clause in SQL are parsed and the validation method {@link
39+
* PipeSource#validate(PipeParameterValidator)} will be called to validate the {@link
40+
* PipeParameters}.
4041
* <li>Before the collaboration task starts, the method {@link
4142
* PipeSource#customize(PipeParameters, PipeSourceRuntimeConfiguration)} will be called to
4243
* configure the runtime behavior of the {@link PipeSource}.

‎iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,12 @@
1919

2020
package org.apache.iotdb.pipe.api.customizer.parameter;
2121

22-
import org.apache.iotdb.pipe.api.PipeConnector;
23-
import org.apache.iotdb.pipe.api.PipeExtractor;
2422
import org.apache.iotdb.pipe.api.PipeProcessor;
25-
import org.apache.iotdb.pipe.api.customizer.configuration.PipeConnectorRuntimeConfiguration;
26-
import org.apache.iotdb.pipe.api.customizer.configuration.PipeExtractorRuntimeConfiguration;
23+
import org.apache.iotdb.pipe.api.PipeSink;
24+
import org.apache.iotdb.pipe.api.PipeSource;
2725
import org.apache.iotdb.pipe.api.customizer.configuration.PipeProcessorRuntimeConfiguration;
26+
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSinkRuntimeConfiguration;
27+
import org.apache.iotdb.pipe.api.customizer.configuration.PipeSourceRuntimeConfiguration;
2828

2929
import java.util.HashMap;
3030
import java.util.HashSet;
@@ -38,9 +38,9 @@
3838
import java.util.stream.Collectors;
3939

4040
/**
41-
* Used in {@link PipeExtractor#customize(PipeParameters, PipeExtractorRuntimeConfiguration)} ,
42-
* {@link PipeProcessor#customize(PipeParameters, PipeProcessorRuntimeConfiguration)} and {@link
43-
* PipeConnector#customize(PipeParameters, PipeConnectorRuntimeConfiguration)}.
41+
* Used in {@link PipeSource#customize(PipeParameters, PipeSourceRuntimeConfiguration)}, {@link
42+
* PipeProcessor#customize(PipeParameters, PipeProcessorRuntimeConfiguration)} and {@link
43+
* PipeSink#customize(PipeParameters, PipeSinkRuntimeConfiguration)}.
4444
*
4545
* <p>This class is used to parse the parameters in WITH SOURCE, WITH PROCESSOR and WITH SINK when
4646
* creating a pipe.

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

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ public PipeConfigNodeSubtask(
7171
final Map<String, String> sinkAttributes,
7272
final PipeTaskMeta pipeTaskMeta)
7373
throws Exception {
74-
// We initialize outputPipeConnector by initConnector()
74+
// We initialize outputPipeSink by initSink()
7575
super(pipeName + "_" + creationTime, creationTime, null);
7676
this.pipeName = pipeName;
7777
this.pipeTaskMeta = pipeTaskMeta;
@@ -88,14 +88,14 @@ public PipeConfigNodeSubtask(
8888
private void initSource(final Map<String, String> sourceAttributes) throws Exception {
8989
final PipeParameters sourceParameters = new PipeParameters(sourceAttributes);
9090

91-
// 1. Construct extractor
91+
// 1. Construct source
9292
source = PipeConfigNodeAgent.plugin().reflectSource(sourceParameters);
9393

9494
try {
95-
// 2. Validate extractor parameters
95+
// 2. Validate source parameters
9696
source.validate(new PipeParameterValidator(sourceParameters));
9797

98-
// 3. Customize extractor
98+
// 3. Customize source
9999
final PipeTaskRuntimeConfiguration runtimeConfiguration =
100100
new PipeTaskRuntimeConfiguration(
101101
new PipeTaskSourceRuntimeEnvironment(
@@ -135,14 +135,14 @@ private void initProcessor(final Map<String, String> processorAttributes) {
135135
private void initSink(final Map<String, String> sinkAttributes) throws Exception {
136136
final PipeParameters sinkParameters = new PipeParameters(sinkAttributes);
137137

138-
// 1. Construct connector
138+
// 1. Construct sink
139139
outputPipeSink = PipeConfigNodeAgent.plugin().reflectSink(sinkParameters);
140140

141141
try {
142-
// 2. Validate connector parameters
142+
// 2. Validate sink parameters
143143
outputPipeSink.validate(new PipeParameterValidator(sinkParameters));
144144

145-
// 3. Customize connector
145+
// 3. Customize sink
146146
final PipeTaskRuntimeConfiguration runtimeConfiguration =
147147
new PipeTaskRuntimeConfiguration(
148148
new PipeTaskSinkRuntimeEnvironment(pipeName, creationTime, CONFIG_REGION_ID.getId()));
@@ -222,7 +222,7 @@ public void close() {
222222
} catch (final Exception e) {
223223
LOGGER.info(ManagerMessages.ERROR_OCCURRED_DURING_CLOSING_PIPECONNECTOR, e);
224224
} finally {
225-
// Should be after connector.close()
225+
// Should be after sink.close()
226226
super.close();
227227
}
228228
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -269,7 +269,7 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
269269
runtimeMeta -> !runtimeMeta.getStatus().get().equals(PipeStatus.STOPPED))
270270
.forEach(
271271
runtimeMeta -> {
272-
// Record the connector exception for each pipe affected
272+
// Record the sink exception for each pipe affected
273273
Map<Integer, PipeRuntimeException> exceptionMap =
274274
runtimeMeta.getNodeId2PipeRuntimeExceptionMap();
275275
if (!exceptionMap.containsKey(nodeId)
@@ -285,7 +285,7 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
285285

286286
LOGGER.warn(
287287
String.format(
288-
"Detect PipeRuntimeConnectorCriticalException %s "
288+
"Detect PipeRuntimeSinkCriticalException %s "
289289
+ "from agent, stop pipe %s.",
290290
exception, pipeName));
291291
});

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/util/PipeExternalSourceLoadBalancer.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ public PipeExternalSourceLoadBalancer(final String balanceStrategy) {
7070
* Balances the given number of parallel tasks across available nodes.
7171
*
7272
* @param parallelCount number of external source tasks to distribute
73-
* @param pipeStaticMeta metadata about the pipe extractor
73+
* @param pipeStaticMeta metadata about the pipe source
7474
* @param configManager reference to ConfigManager for cluster information
7575
* @return a mapping from task index to leader node id
7676
*/

‎iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusV2Messages.java‎

Lines changed: 35 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
package org.apache.iotdb.consensus.i18n;
2121

2222
/**
23-
* IoTConsensusV2（基于 pipe 的共识）特有消息。日志消息使用 SLF4J {@code {}} 占位符；异常消息使用 {@code %s}（String.format）或纯字符串。
23+
* IoTConsensusV2（基于 pipe 的共识）特有消息。日志消息使用 SLF4J {@code {}} 占位符；异常消息使用 {@code %s}（String.format）或普通字符串。
2424
*/
2525
public final class IoTConsensusV2Messages {
2626

@@ -31,92 +31,92 @@ private IoTConsensusV2Messages() {}
3131
public static final String RECOVER_TASK_CANCELLED =
3232
"IoTV2 恢复任务已取消";
3333
public static final String RECOVER_FUTURE_EXCEPTION =
34-
"等待恢复 future 完成时发生异常";
34+
"等待恢复 Future 完成时发生异常";
3535
public static final String RECOVER_TASK_INTERRUPTED =
3636
"IoTV2 恢复任务被中断";
3737
public static final String FAILED_RECOVER_CONSENSUS =
38-
"从 {} 恢复共识组 {} 失败，忽略并继续恢复其他组，异步后台检查线程将自动注销该失败共识组的 pipe 副作用。";
38+
"从 {} 恢复共识组 {} 失败，将忽略该组并继续恢复其他共识组；异步后台检查线程会自动清理该失败共识组相关的 pipe 副作用。";
3939
public static final String FAILED_RECOVER_CONSENSUS_READ_DIR =
40-
"从 {} 恢复共识失败，因为读取目录失败";
40+
"从 {} 恢复共识失败，原因：读取目录失败";
4141
public static final String FAILED_RECOVER_CONSENSUS_SHORT =
4242
"从 {} 恢复共识失败";
4343

44-
// ===================== IoTConsensusV2 peer 操作 =====================
44+
// ===================== IoTConsensusV2 节点操作 =====================
4545

4646
public static final String START_DELETE_LOCAL_PEER =
47-
"[{}] 开始删除共识组 {} 的本地 peer";
47+
"[{}] 开始删除共识组 {} 的本地节点";
4848
public static final String FINISH_DELETE_LOCAL_PEER =
49-
"[{}] 完成删除共识组 {} 的本地 peer";
49+
"[{}] 完成删除共识组 {} 的本地节点";
5050
public static final String INACTIVATE_NEW_PEER =
51-
"[{}] 将新 peer 设为非活跃：{}";
51+
"[{}] 停用新节点：{}";
5252
public static final String NOTIFY_CREATE_CONSENSUS_PIPES =
53-
"[{}] 通知当前 peer 创建 consensus pipe...";
53+
"[{}] 通知当前节点创建共识管道...";
5454
public static final String WAIT_PEERS_FINISH_TRANSFER =
55-
"[{}] 等待所有其他 peer 完成传输...";
55+
"[{}] 等待其他节点完成传输...";
5656
public static final String ACTIVATE_NEW_PEER =
57-
"[{}] 激活新 peer...";
57+
"[{}] 激活新节点...";
5858
public static final String ADD_REMOTE_PEER_FAILED_CLEANUP =
59-
"[{}] 添加远程 peer 失败，正在自动清理副作用...";
59+
"[{}] 添加远程节点失败，正在自动清理副作用...";
6060
public static final String FAILED_CLEANUP_SIDE_EFFECTS =
61-
"[{}] 添加远程 peer 失败后清理副作用失败";
61+
"[{}] 添加远程节点失败后清理副作用失败";
6262
public static final String NOTIFY_DROP_CONSENSUS_PIPES =
63-
"[{}] 通知其他 peer 删除 consensus pipe...";
63+
"[{}] 通知其他节点删除共识管道...";
6464
public static final String INACTIVATE_PEER =
65-
"[{}] 将 peer {} 设为非活跃";
65+
"[{}] 停用节点 {}";
6666
public static final String WAIT_TARGET_PEER_COMPLETE_TRANSFER =
67-
"[{}] 等待目标 peer{} 完成传输...";
67+
"[{}] 等待目标节点 {} 完成传输...";
6868
public static final String WAIT_PEER_RELEASE_RESOURCE =
6969
"[{}] 等待 {} 释放所有资源...";
7070
public static final String NOT_SUPPORT_LEADER_TRANSFER =
71-
"%s 不支持 leader 切换";
71+
"%s 不支持主节点切换";
7272

7373
// ===================== IoTConsensusV2ServerImpl =====================
7474

7575
public static final String ERROR_SET_PEER_ACTIVE =
76-
"将 peer %s 设置为活跃状态 %s 时出错。结果状态：%s";
76+
"设置节点 %s 的活跃状态为 %s 时出错。结果状态：%s";
7777
public static final String ERROR_SET_PEER_ACTIVE_SHORT =
78-
"将 peer %s 设置为活跃状态 %s 时出错";
78+
"设置节点 %s 的活跃状态为 %s 时出错";
7979
public static final String TARGET_PEER_MAY_BE_DOWN =
80-
"目标 peer 可能已下线，将 peer {} 设置为活跃状态 {} 时出错";
80+
"目标节点可能已下线，设置节点 {} 的活跃状态为 {} 时出错";
8181
public static final String CANNOT_NOTIFY_PEER_CREATE_PIPE =
82-
"{} 无法通知 peer {} 创建 consensus pipe，该 peer 可能当前未知，请手动检查！";
82+
"{} 无法通知节点 {} 创建共识管道，该节点当前可能未知，请手动检查。";
8383
public static final String CANNOT_CREATE_CONSENSUS_PIPE =
84-
"{} 无法创建到 {} 的 consensus pipe，目标 peer 可能当前未知，请手动检查！";
84+
"{} 无法创建到 {} 的共识管道，目标节点当前可能未知，请手动检查。";
8585
public static final String ERROR_NOTIFY_PEER_CREATE_PIPE =
86-
"通知 peer %s 创建 consensus pipe 时出错";
86+
"通知节点 %s 创建共识管道时出错";
8787
public static final String CANNOT_NOTIFY_PEER_DROP_PIPE =
88-
"{} 无法通知 peer {} 删除 consensus pipe，该 peer 可能当前未知，请手动检查！";
88+
"{} 无法通知节点 {} 删除共识管道，该节点当前可能未知，请手动检查。";
8989
public static final String CANNOT_DROP_CONSENSUS_PIPE =
90-
"{} 无法删除到 {} 的 consensus pipe，目标 peer 可能当前未知，请手动检查！";
90+
"{} 无法删除到 {} 的共识管道，目标节点当前可能未知，请手动检查。";
9191
public static final String ERROR_NOTIFY_PEER_DROP_PIPE =
92-
"通知 peer %s 删除 consensus pipe 时出错";
92+
"通知节点 %s 删除共识管道时出错";
9393
public static final String INTERRUPTED_WAITING_TRANSFER =
9494
"{} 等待传输完成时被中断";
9595
public static final String INTERRUPTED_WAITING_TRANSFER_FMT =
9696
"%s 等待传输完成时被中断";
9797
public static final String CANNOT_CHECK_PIPE_TRANSMISSION =
98-
"{} 无法检查到 peer {} 的 consensus pipe 传输完成状态";
98+
"{} 无法检查到节点 {} 的共识管道是否传输完成";
9999
public static final String ERROR_CHECK_PIPE_TRANSMISSION =
100-
"检查到 peer %s 的 consensus pipe 传输完成状态时出错";
100+
"检查到节点 %s 的共识管道是否传输完成时出错";
101101
public static final String CANNOT_CHECK_PIPE_TRANSMISSION_SHORT =
102-
"{} 无法检查 consensus pipe 传输完成状态";
102+
"{} 无法检查共识管道是否传输完成";
103103

104104
// ===================== IoTConsensusV2RPCServiceProcessor =====================
105105

106106
public static final String UNEXPECTED_GROUP_SET_ACTIVE =
107-
"共识组 ID %s 与设置活跃请求 %s 不匹配";
107+
"共识组 ID %s 与设置活跃状态请求 %s 不匹配";
108108
public static final String UNEXPECTED_GROUP_CREATE_PIPE =
109-
"共识组 ID %s 与创建 consensus pipe 请求 %s 不匹配";
109+
"共识组 ID %s 与创建共识管道请求 %s 不匹配";
110110
public static final String UNEXPECTED_GROUP_DROP_PIPE =
111-
"共识组 ID %s 与删除 consensus pipe 请求 %s 不匹配";
111+
"共识组 ID %s 与删除共识管道请求 %s 不匹配";
112112
public static final String UNEXPECTED_GROUP_CHECK_TRANSFER =
113113
"共识组 ID %s 与检查传输完成请求 %s 不匹配";
114114
public static final String UNEXPECTED_GROUP_WAIT_RELEASE =
115115
"共识组 ID %s 与 TWaitReleaseAllRegionRelatedResourceRes 请求不匹配";
116116
public static final String FAILED_CREATE_CONSENSUS_PIPE =
117-
"创建到目标 peer 的 consensus pipe 失败，请求 {}";
117+
"创建到目标节点的共识管道失败，请求：{}";
118118
public static final String FAILED_DROP_CONSENSUS_PIPE =
119-
"删除到目标 peer 的 consensus pipe 失败，请求 {}";
119+
"删除到目标节点的共识管道失败，请求：{}";
120120
public static final String FAILED_CHECK_CONSENSUS_PIPE =
121-
"检查 consensus pipe 完成状态失败，请求 {}，将完成状态设为 {}";
121+
"检查共识管道传输完成状态失败，请求：{}，将完成状态设为 {}";
122122
}

0 commit comments

Comments
 (0)