This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong-website.git
The following commit(s) were added to refs/heads/master by this push:
new fbfc7f8233 [INLONG-957][Doc] Supplement Agent plugin documentation
(#958)
fbfc7f8233 is described below
commit fbfc7f8233aa29ad58c431d7790943a45a9b76bb
Author: haifxu <[email protected]>
AuthorDate: Tue Jun 25 15:29:37 2024 +0800
[INLONG-957][Doc] Supplement Agent plugin documentation (#958)
---
.../how_to_write_plugin_agent.md | 48 +++++++++++++++------
.../how_to_write_plugin_agent.md | 50 +++++++++++++++-------
2 files changed, 69 insertions(+), 29 deletions(-)
diff --git a/docs/design_and_concept/how_to_write_plugin_agent.md
b/docs/design_and_concept/how_to_write_plugin_agent.md
index 2bb4d1239e..6e85ba501a 100644
--- a/docs/design_and_concept/how_to_write_plugin_agent.md
+++ b/docs/design_and_concept/how_to_write_plugin_agent.md
@@ -18,7 +18,8 @@ Source and Sink are lower-level concepts of Instance. They
can be simply underst
- Add Task: implement logic such as initialization, destruction, configuration
verification, etc.
- Add Instance: implements logic such as node information setting.
- Add Source: implements logic such as initialization, destruction, data
collection, and data provision.
-- Add Sink: implement logic such as initialization, destruction, data input,
data output (this article only focuses on new data sources, Sink will not be
introduced, the default Sink is ProxySink)
+- Add Sink: implement logic such as initialization, destruction, data input,
data output (this article only focuses on new data sources, Sink will not be
introduced, the default Sink is ProxySink).
+- Add TaskPojo: handles the differences between Agent and Manager fields and
binds Task, Source, etc.
### Add Task
Here we need to add a PulsarTask class to org.apache.inlong.agent.plugin.task.
@@ -111,23 +112,42 @@ public class PulsarSource extends AbstractSource {
- readFromSource: actually reads data from the data source, such as consuming
data from Kafka SDK and Pulsar SDK.
- getThreadName: get the worker thread name of the data source.
- isRunnable: returns whether this data source should continue.
-- releaseSource: release the resources of the data source
+- releaseSource: release the resources of the data source.
+
+### Add TaskPojo
+Add the PulsarTask class in `org.apache.inlong.agent.pojo`:
+```
+public class PulsarTask {
+
+ private String topic;
+ private String subscription;
+
+ public static class PulsarTaskConfig {
+
+ private String topic;
+ private String subscription;
+ }
+}
+```
+- The field names in PulsarTaskConfig are the names passed by the Manager and
must be consistent with the field names defined by the Manager.
+- The field names and types in PulsarTask are the ones required by the Agent.
## Task configuration
From the above, we can see that we have created new classes such as Task,
Instance, Source, etc., and task configuration is to connect these classes
together.
+
+Bind Task, Source, etc. to Pulsar in `convertToTaskProfile` in
`org.apache.inlong.agent.pojo.TaskProfileDto` class:
```
-{
- "task.id": "74",
- "task.groupId": "test_group_pulsar",
- "task.streamId": "test_stream_pulsar",
- "task.source": "org.apache.inlong.agent.plugin.sources.PulsarSource",
- "task.sink": "org.apache.inlong.agent.plugin.sinks.ProxySink",
- "task.taskClass": "org.apache.inlong.agent.plugin.task.PulsarTask"
-}
+case PULSAR:
+ task.setTaskClass(DEFAULT_PULSAR_TASK);
+ PulsarTask pulsarTask = getPulsarTask(dataConfig);
+ task.setPulsarTask(pulsarTask);
+ task.setSource(PULSAR_SOURCE);
+ profileDto.setTask(task);
+ break;
```
-- task.source: Source class specified
-- task.sink: Sink class specified
-- task.taskClass: specifies the Task class
+- task.source: Source class specified.
+- task.sink: Sink class specified.
+- task.taskClass: specifies the Task class.
## Offset control
```
@@ -144,7 +164,7 @@ From the above, we can see that we have created new classes
such as Task, Instan
```
We can see that when the Source reads data, each piece of data will record its
corresponding Offset. This Offset will be automatically recorded by the Agent
after the Sink is successfully written.
When Source is initialized, its corresponding Offset will be automatically
read and stored in the member variable offsetProfile of AbstractSource. You can
use offsetProfile.getOffset() to
-Get its Offset for initializing the data source.
+get its Offset for initializing the data source.
```
protected void initOffset() {
offsetProfile = OffsetManager.getInstance().getOffset(taskId,
instanceId);
diff --git
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/design_and_concept/how_to_write_plugin_agent.md
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/design_and_concept/how_to_write_plugin_agent.md
index c995ceead7..691b625c6b 100644
---
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/design_and_concept/how_to_write_plugin_agent.md
+++
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/design_and_concept/how_to_write_plugin_agent.md
@@ -18,7 +18,8 @@ Source 和 Sink 属于 Instance 下一级的概念，可以简单理解为每个
- 新增 Task：实现初始化、销毁、配置校验等逻辑。
- 新增 Instance：实现节点信息设置等逻辑。
- 新增 Source：实现初始化、销毁、采集数据、提供数据等逻辑。
-- 新增 Sink：实现初始化、销毁、数据输入、数据输出等逻辑（本文只针对新增数据源，Sink 不做介绍，默认 Sink 是 ProxySink）
+- 新增 Sink：实现初始化、销毁、数据输入、数据输出等逻辑（本文只针对新增数据源，Sink 不做介绍，默认 Sink 是 ProxySink）。
+- 新增 TaskPojo：处理 Agent 与 Manager 字段上的差异以及绑定 Task、Source 等。
### 新增 Task
这里就是要在 org.apache.inlong.agent.plugin.task 新增一个 PulsarTask 类。
@@ -111,23 +112,42 @@ public class PulsarSource extends AbstractSource {
- readFromSource：真正从数据源读取数据，例如从 Kafka SDK、Pulsar SDK 消费数据。
- getThreadName：获取该数据源的工作线程名。
- isRunnable：返回该数据源是否应该继续。
-- releaseSource：释放该数据源的资源
+- releaseSource：释放该数据源的资源。
-## 任务配置
-从上面看我们新建了 Task、Instance、Source 等类，而任务配置就是将这些了类串联起来
-```
-{
- "task.id": "74",
- "task.groupId": "test_group_pulsar",
- "task.streamId": "test_stream_pulsar",
- "task.source": "org.apache.inlong.agent.plugin.sources.PulsarSource",
- "task.sink": "org.apache.inlong.agent.plugin.sinks.ProxySink",
- "task.taskClass": "org.apache.inlong.agent.plugin.task.PulsarTask"
+### 新增 TaskPojo
+在 `org.apache.inlong.agent.pojo 增加 PulsarTask` 类：
+```
+public class PulsarTask {
+
+ private String topic;
+ private String subscription;
+
+ public static class PulsarTaskConfig {
+
+ private String topic;
+ private String subscription;
+ }
}
```
-- task.source：指定了 Source 类
-- task.sink：指定了 Sink 类
-- task.taskClass：指定了 Task 类
+- PulsarTaskConfig 中的字段名称为 Manager 传递的名称，必须与 Manager 定义的字段名称保持一致。
+- PulsarTask 中的字段名称以及类型为 Agent 所需。
+
+## 任务配置
+从上面看我们新建了 Task、Instance、Source 等类，而任务配置就是将这些了类串联起来。
+
+在`org.apache.inlong.agent.pojo.TaskProfileDto` 类中的 `convertToTaskProfile` 中为
Pulsar 绑定 Task、Source 等：
+```
+case PULSAR:
+ task.setTaskClass(DEFAULT_PULSAR_TASK);
+ PulsarTask pulsarTask = getPulsarTask(dataConfig);
+ task.setPulsarTask(pulsarTask);
+ task.setSource(PULSAR_SOURCE);
+ profileDto.setTask(task);
+ break;
+```
+- task.source：指定了 Source 类。
+- task.sink：指定了 Sink 类。
+- task.taskClass：指定了 Task 类。
## 位点控制
```
