This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 3ba11c7188 [Feature][Connector-V2] Support Avro format for Pulsar
connector (#11230)
3ba11c7188 is described below
commit 3ba11c718874e30b8342e3ea5ee021448b2ccca3
Author: Dream95 <[email protected]>
AuthorDate: Wed Jul 8 20:21:36 2026 +0800
[Feature][Connector-V2] Support Avro format for Pulsar connector (#11230)
---
docs/en/connectors/sink/Pulsar.md | 5 +-
docs/en/connectors/source/Pulsar.md | 6 +-
docs/zh/connectors/sink/Pulsar.md | 4 +-
docs/zh/connectors/source/Pulsar.md | 6 +-
seatunnel-connectors-v2/connector-pulsar/pom.xml | 6 +
.../seatunnel/pulsar/config/PulsarBaseOptions.java | 6 +-
.../pulsar/config/PulsarMultiTableConfig.java | 34 ++-
.../seatunnel/pulsar/sink/PulsarSinkWriter.java | 3 +
.../seatunnel/pulsar/source/PulsarSource.java | 3 +
.../pulsar/config/PulsarMultiTableConfigTest.java | 92 +++++++
.../e2e/connector/pulsar/PulsarFormatIT.java | 283 +++++++++++++++++++++
.../test/resources/avro/fake_to_pulsar_avro.conf | 63 +++++
.../test/resources/avro/pulsar_avro_to_assert.conf | 101 ++++++++
13 files changed, 595 insertions(+), 17 deletions(-)
diff --git a/docs/en/connectors/sink/Pulsar.md
b/docs/en/connectors/sink/Pulsar.md
index 7e70db1774..af97ca3275 100644
--- a/docs/en/connectors/sink/Pulsar.md
+++ b/docs/en/connectors/sink/Pulsar.md
@@ -35,7 +35,7 @@ The Pulsar sink writes SeaTunnel rows to Apache Pulsar
topics. It can write to o
| admin.service-url | String | Yes | - | Pulsar
admin HTTP URL, for example `http://localhost:8080`.
|
| auth.plugin-class | String | No | - | Pulsar
authentication plugin class name.
|
| auth.params | String | No | - |
Parameters for the authentication plugin. Configure it together with
`auth.plugin-class`. |
-| format | String | No | json | Data
format. Supports `json` and `text`.
|
+| format | String | No | json | Data
format. The default format is json. Optional text and avro format.
|
| field_delimiter | String | No | , | Field
delimiter used when `format = "text"`.
|
| semantics | Enum | No | AT_LEAST_ONCE | Write
consistency. Valid values: `NON`, `AT_LEAST_ONCE`, `EXACTLY_ONCE`.
|
| transaction_timeout | Int | No | 600 | Pulsar
transaction timeout in seconds. Used by `EXACTLY_ONCE`.
|
@@ -77,7 +77,8 @@ For example, `key1:val1,key2:val2`.
### format [String]
-Data format. The default format is `json`. You can also use `text`. When using
`text`, configure `field_delimiter` if the default comma delimiter is not
suitable.
+Data format. The default format is `json`. Optional text and avro format. You
can also use `text`. When using `text`, configure `field_delimiter` if the
default comma delimiter is not suitable.
+When using avro format, the Avro schema is derived from the upstream row type;
no sink-side `schema` option is required.
### field_delimiter [String]
diff --git a/docs/en/connectors/source/Pulsar.md
b/docs/en/connectors/source/Pulsar.md
index 724e77c6a5..750e4ae44f 100644
--- a/docs/en/connectors/source/Pulsar.md
+++ b/docs/en/connectors/source/Pulsar.md
@@ -41,7 +41,7 @@ Source connector for Apache Pulsar.
| cursor.stop.mode | Enum | No | NEVER | Stop
position mode. Options: `NEVER` (streaming), `LATEST` (batch), `TIMESTAMP`
(batch) |
| cursor.stop.timestamp | Long | No | - | Stop
timestamp (ms) when `cursor.stop.mode=TIMESTAMP`
|
| schema | Config | No | - | Data
structure including field names and types
|
-| format | String | No | json | Data format.
Supports `json` and `canal_json`.
|
+| format | String | No | json | Data format.
Default is json. Supported formats: json, canal_json and avro. **Multi-table
mode only supports JSON, CANAL_JSON and AVRO** |
| common-options | | No | - | Source
plugin common parameters. See [Source Common
Options](../common-options/source-common-options.md) for details |
### topic [String]
@@ -73,7 +73,7 @@ Additional rules:
- `table_path` is required when `topic-pattern` is used.
- `subscription.name` must exist either globally or inside the item.
-- Only `JSON` and `CANAL_JSON` are supported in multi-table mode.
+- Only `JSON`, `CANAL_JSON` and `AVRO` are supported in multi-table mode.
- Explicit `topic` entries must not overlap with any `topic-pattern` entry.
- If multiple `topic-pattern` items can match the same topic, the first
matching item in `tables_configs` wins. Put more specific patterns before
broader ones.
- In batch mode, multi-table configurations must be bounded. If more than one
table is configured and any table uses `cursor.stop.mode = NEVER`, the source
is unbounded and batch jobs are rejected. Single-table mode and single-entry
`tables_configs` keep backward-compatible batch behavior.
@@ -162,7 +162,7 @@ reference to
[Schema-Feature](../../introduction/concepts/schema-feature.md)
### format [String]
-Data format. The default value is `json`. Pulsar source currently supports
`json` and `canal_json`.
+Data format. The default format is json. Supported formats are json,
canal_json and avro. The `schema` option is required when using avro format.
See [formats](../formats) for more details.
### common options
diff --git a/docs/zh/connectors/sink/Pulsar.md
b/docs/zh/connectors/sink/Pulsar.md
index 4e53fab6ce..baee0e785c 100644
--- a/docs/zh/connectors/sink/Pulsar.md
+++ b/docs/zh/connectors/sink/Pulsar.md
@@ -35,7 +35,7 @@ Pulsar Sink 用于将 SeaTunnel 数据写入 Apache Pulsar topic。它既可以
| admin.service-url | String | 是 | - | Pulsar
管理端 HTTP 地址，例如 `http://localhost:8080`。 |
| auth.plugin-class | String | 否 | - | Pulsar
认证插件类名。 |
| auth.params | String | 否 | - |
认证插件参数。需要和 `auth.plugin-class` 一起配置。 |
-| format | String | 否 | json | 数据格式。支持
`json` 和 `text`。 |
+| format | String | 否 | json |
数据格式。默认格式为 json。可选 text 和 avro 格式。 |
| field_delimiter | String | 否 | , | 当
`format = "text"` 时使用的字段分隔符。 |
| semantics | Enum | 否 | AT_LEAST_ONCE |
写入一致性语义。可选值：`NON`、`AT_LEAST_ONCE`、`EXACTLY_ONCE`。 |
| transaction_timeout | Int | 否 | 600 | Pulsar
事务超时时间，单位为秒。用于 `EXACTLY_ONCE`。 |
@@ -77,7 +77,7 @@ Pulsar 管理端 HTTP 地址。
### format [String]
-数据格式。默认值为 `json`，也可以使用 `text`。使用 `text` 时，如果默认逗号分隔符不合适，请配置 `field_delimiter`。
+数据格式。默认格式为 json。可选 text 和 avro
格式。默认字段分隔符为","。如果自定义分隔符，请添加"field_delimiter"选项。使用 avro 格式时，Avro schema 会从上游数据的
row type 自动推导，无需在 sink 侧单独配置 `schema`。
### field_delimiter [String]
diff --git a/docs/zh/connectors/source/Pulsar.md
b/docs/zh/connectors/source/Pulsar.md
index 9d7f20abe8..f905aa562b 100644
--- a/docs/zh/connectors/source/Pulsar.md
+++ b/docs/zh/connectors/source/Pulsar.md
@@ -41,7 +41,7 @@ Apache Pulsar 的源连接器。
| cursor.stop.mode | Enum | 否 | NEVER |
停止位置模式。可选值:`NEVER`(流式)、`LATEST`(批式)、`TIMESTAMP`(批式)
|
| cursor.stop.timestamp | Long | 否 | - | 当
`cursor.stop.mode=TIMESTAMP` 时的停止时间戳(毫秒)
|
| schema | Config | 否 | - | 数据结构,包括字段名称和字段类型
|
-| format | String | 否 | json | 数据格式。支持 `json` 和
`canal_json`。 |
+| format | String | 否 | json | 数据格式。默认为 json。支持
json、canal_json 和 avro 格式。**多表模式仅支持 JSON、CANAL_JSON 和 AVRO**
|
| common-options | | 否 | - | Source 插件通用参数,请参考
[Source Common Options](../common-options/source-common-options.md) 了解详情
|
### topic [String]
@@ -73,7 +73,7 @@ Apache Pulsar 的源连接器。
- 当使用 `topic-pattern` 时，必须显式配置 `table_path`。
- `subscription.name` 必须在全局或 item 内存在。
-- 多表模式当前只支持 `JSON` 和 `CANAL_JSON`。
+- 多表模式当前只支持 `JSON`、`CANAL_JSON` 和 `AVRO`。
- 显式配置的 `topic` 不能与任何 `topic-pattern` 发生重叠。
- 在 batch 模式下，多表配置必须全部是 bounded。只有当配置了多于一张表且任意一张表使用 `cursor.stop.mode = NEVER`
时，整个 source 才会被视为 unbounded，并拒绝在 batch 作业中运行。单表模式和仅包含一个配置项的 `tables_configs`
保持向后兼容的 batch 行为。
@@ -155,7 +155,7 @@ Pulsar 消费者的启动模式，有效值为 `'EARLIEST'`、`'LATEST'`、`'SUB
### format [String]
-数据格式。默认值为 `json`。Pulsar Source 当前支持 `json` 和 `canal_json`。
+数据格式。默认值为 `json`。支持 json、canal_json 和 avro 格式。使用 avro 格式时需要配置
`schema`。更多格式说明参考 [formats](../formats)。
### 通用参数
diff --git a/seatunnel-connectors-v2/connector-pulsar/pom.xml
b/seatunnel-connectors-v2/connector-pulsar/pom.xml
index e213275a72..fe7cc67723 100644
--- a/seatunnel-connectors-v2/connector-pulsar/pom.xml
+++ b/seatunnel-connectors-v2/connector-pulsar/pom.xml
@@ -106,5 +106,11 @@
<artifactId>seatunnel-format-text</artifactId>
<version>${project.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-avro</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
</project>
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarBaseOptions.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarBaseOptions.java
index 7cf5998737..65b8822e7d 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarBaseOptions.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarBaseOptions.java
@@ -62,6 +62,8 @@ public class PulsarBaseOptions extends ConnectorCommonOptions
{
public static final String TEXT_FORMAT = "text";
+ public static final String AVRO_FORMAT = "avro";
+
/** The default field delimiter is “,” */
public static final String DEFAULT_FIELD_DELIMITER = ",";
@@ -70,8 +72,8 @@ public class PulsarBaseOptions extends ConnectorCommonOptions
{
.stringType()
.defaultValue(DEFAULT_FORMAT)
.withDescription(
- "Data format. The default format is json. Optional
text format. "
- + "For multi-table mode, only JSON and
CANAL_JSON are supported. "
+ "Data format. The default format is json. Optional
text and avro format. "
+ + "For multi-table mode, only JSON,
CANAL_JSON and AVRO are supported. "
+ "The default field separator is \", \". "
+ "If you customize the delimiter, add the
\"field_delimiter\" option.");
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfig.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfig.java
index 4b6bc721ab..e17db3ddcd 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfig.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfig.java
@@ -106,7 +106,8 @@ public class PulsarMultiTableConfig implements Serializable
{
stopMode,
stopTimestamp,
resetMode,
- subscriptionName);
+ subscriptionName,
+ config);
PulsarTableConfig tableConfig =
new PulsarTableConfig(
tablePath,
@@ -216,7 +217,8 @@ public class PulsarMultiTableConfig implements Serializable
{
stopMode,
stopTimestamp,
resetMode,
- subscriptionName);
+ subscriptionName,
+ tableConfig);
validateTopicOverlap(
i,
topic,
@@ -264,7 +266,8 @@ public class PulsarMultiTableConfig implements Serializable
{
StopMode stopMode,
Long stopTimestamp,
PulsarSourceOptions.CursorResetStrategy resetMode,
- String subscriptionName) {
+ String subscriptionName,
+ ReadonlyConfig schemaConfig) {
String configPrefix =
index >= 0
? String.format("tables_configs[%d] ('%s')", index,
tablePath)
@@ -285,6 +288,7 @@ public class PulsarMultiTableConfig implements Serializable
{
}
validateFormat(format, index, tablePath);
+ validateAvroSchema(format, index, tablePath, schemaConfig);
if (startMode == StartMode.TIMESTAMP && startTimestamp == null) {
throw new PulsarConnectorException(
@@ -316,15 +320,35 @@ public class PulsarMultiTableConfig implements
Serializable {
? String.format("tables_configs[%d] ('%s')", index,
tablePath)
: "Pulsar source config";
String normalized = format.toUpperCase();
- if (!Objects.equals("JSON", normalized) &&
!Objects.equals("CANAL_JSON", normalized)) {
+ if (!Objects.equals("JSON", normalized)
+ && !Objects.equals("CANAL_JSON", normalized)
+ && !Objects.equals("AVRO", normalized)) {
throw new PulsarConnectorException(
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
String.format(
- "%s uses unsupported format '%s', only JSON and
CANAL_JSON are supported",
+ "%s uses unsupported format '%s', only JSON,
CANAL_JSON and AVRO are supported",
configPrefix, format));
}
}
+ private static void validateAvroSchema(
+ String format, int index, TablePath tablePath, ReadonlyConfig
schemaConfig) {
+ if (!Objects.equals("AVRO", format.toUpperCase())) {
+ return;
+ }
+ if (schemaConfig.getOptional(PulsarSourceOptions.SCHEMA).isPresent()) {
+ return;
+ }
+ String configPrefix =
+ index >= 0
+ ? String.format("tables_configs[%d] ('%s')", index,
tablePath)
+ : "Pulsar source config";
+ throw new PulsarConnectorException(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
+ String.format(
+ "%s uses format 'AVRO' but 'schema' is not
configured", configPrefix));
+ }
+
private static void validateTopicOverlap(
int index,
String topic,
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
index cee685f69e..677ea795cb 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/sink/PulsarSinkWriter.java
@@ -37,6 +37,7 @@ import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnecto
import
org.apache.seatunnel.connectors.seatunnel.pulsar.exception.PulsarConnectorException;
import org.apache.seatunnel.connectors.seatunnel.pulsar.state.PulsarCommitInfo;
import org.apache.seatunnel.connectors.seatunnel.pulsar.state.PulsarSinkState;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
import org.apache.seatunnel.format.json.JsonSerializationSchema;
import org.apache.seatunnel.format.json.exception.SeaTunnelJsonFormatException;
import org.apache.seatunnel.format.text.TextSerializationSchema;
@@ -281,6 +282,8 @@ public class PulsarSinkWriter
.seaTunnelRowType(rowType)
.delimiter(delimiter)
.build();
+ } else if (PulsarSinkOptions.AVRO_FORMAT.equals(format)) {
+ return new AvroSerializationSchema(rowType);
} else {
throw new SeaTunnelJsonFormatException(
CommonErrorCode.UNSUPPORTED_DATA_TYPE, "Unsupported
format: " + format);
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
index 0a72cd9114..f4c28ff6fe 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/main/java/org/apache/seatunnel/connectors/seatunnel/pulsar/source/PulsarSource.java
@@ -52,6 +52,7 @@ import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.enumerator.discov
import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.format.PulsarCanalDecorator;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.reader.PulsarSourceReader;
import
org.apache.seatunnel.connectors.seatunnel.pulsar.source.split.PulsarPartitionSplit;
+import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
import org.apache.seatunnel.format.json.JsonDeserializationSchema;
import org.apache.seatunnel.format.json.canal.CanalJsonDeserializationSchema;
import org.apache.seatunnel.format.json.exception.SeaTunnelJsonFormatException;
@@ -271,6 +272,8 @@ public class PulsarSource
CanalJsonDeserializationSchema.builder(catalogTable)
.setIgnoreParseErrors(true)
.build());
+ case "AVRO":
+ return new AvroDeserializationSchema(catalogTable);
default:
throw new SeaTunnelJsonFormatException(
CommonErrorCode.UNSUPPORTED_DATA_TYPE, "Unsupported
format: " + format);
diff --git
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfigTest.java
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfigTest.java
index 992825e827..0201493603 100644
---
a/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfigTest.java
+++
b/seatunnel-connectors-v2/connector-pulsar/src/test/java/org/apache/seatunnel/connectors/seatunnel/pulsar/config/PulsarMultiTableConfigTest.java
@@ -58,6 +58,21 @@ class PulsarMultiTableConfigTest {
"CANAL_JSON",
multiTableConfig.getTableConfigs().get(1).getFormat());
}
+ @Test
+ void shouldAllowAvroFormatInTablesConfigs() {
+ Map<String, Object> config = createBaseConfig();
+ config.put(
+ "tables_configs",
+ Collections.singletonList(
+ createTableConfig(
+ "db.events",
"persistent://public/default/events", null, "AVRO")));
+
+ PulsarMultiTableConfig multiTableConfig =
+ PulsarMultiTableConfig.of(ReadonlyConfig.fromMap(config));
+
+ Assertions.assertEquals("AVRO",
multiTableConfig.getTableConfigs().get(0).getFormat());
+ }
+
@Test
void shouldRejectOverlappingTopicDeclarations() {
Map<String, Object> config = createBaseConfig();
@@ -171,6 +186,67 @@ class PulsarMultiTableConfigTest {
SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
exception.getSeaTunnelErrorCode());
}
+ @Test
+ void shouldRejectAvroFormatWithoutSchemaInSingleTable() {
+ Map<String, Object> config = createBaseConfig();
+ config.put("topic", "persistent://public/default/events");
+ config.put("format", "AVRO");
+
+ PulsarConnectorException exception =
+ Assertions.assertThrows(
+ PulsarConnectorException.class,
+ () ->
PulsarMultiTableConfig.of(ReadonlyConfig.fromMap(config)));
+
+ Assertions.assertEquals(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
exception.getSeaTunnelErrorCode());
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+ .contains("uses format 'AVRO' but 'schema' is not
configured"));
+ }
+
+ @Test
+ void shouldRejectAvroFormatWithoutSchemaInTablesConfigs() {
+ Map<String, Object> config = createBaseConfig();
+ Map<String, Object> tableConfig =
+ createTableConfigWithoutSchema(
+ "db.events", "persistent://public/default/events",
null, "AVRO");
+ config.put("tables_configs", Collections.singletonList(tableConfig));
+
+ PulsarConnectorException exception =
+ Assertions.assertThrows(
+ PulsarConnectorException.class,
+ () ->
PulsarMultiTableConfig.of(ReadonlyConfig.fromMap(config)));
+
+ Assertions.assertEquals(
+ SeaTunnelAPIErrorCode.CONFIG_VALIDATION_FAILED,
exception.getSeaTunnelErrorCode());
+ Assertions.assertTrue(
+ exception
+ .getMessage()
+ .contains("uses format 'AVRO' but 'schema' is not
configured"));
+ }
+
+ @Test
+ void shouldAllowAvroFormatWithGlobalSchemaInTablesConfigs() {
+ Map<String, Object> config = createBaseConfig();
+ config.put("format", "AVRO");
+ Map<String, Object> schema = new HashMap<>();
+ Map<String, Object> fields = new HashMap<>();
+ fields.put("id", "bigint");
+ schema.put("fields", fields);
+ config.put("schema", schema);
+ config.put(
+ "tables_configs",
+ Collections.singletonList(
+ createTableConfigWithoutSchema(
+ "db.events",
"persistent://public/default/events", null, null)));
+
+ PulsarMultiTableConfig multiTableConfig =
+ PulsarMultiTableConfig.of(ReadonlyConfig.fromMap(config));
+
+ Assertions.assertEquals("AVRO",
multiTableConfig.getTableConfigs().get(0).getFormat());
+ }
+
private Map<String, Object> createBaseConfig() {
Map<String, Object> config = new HashMap<>();
config.put("subscription.name", "seatunnel-sub");
@@ -208,4 +284,20 @@ class PulsarMultiTableConfigTest {
config.put("schema", schema);
return config;
}
+
+ private Map<String, Object> createTableConfigWithoutSchema(
+ String tablePath, String topic, String topicPattern, String
format) {
+ Map<String, Object> config = new HashMap<>();
+ config.put("table_path", tablePath);
+ if (topic != null) {
+ config.put("topic", topic);
+ }
+ if (topicPattern != null) {
+ config.put("topic-pattern", topicPattern);
+ }
+ if (format != null) {
+ config.put("format", format);
+ }
+ return config;
+ }
}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
new file mode 100644
index 0000000000..b6f2ce92bc
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/java/org/apache/seatunnel/e2e/connector/pulsar/PulsarFormatIT.java
@@ -0,0 +1,283 @@
+/*
+ * 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.seatunnel.e2e.connector.pulsar;
+
+import org.apache.seatunnel.api.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.type.ArrayType;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.DecimalType;
+import org.apache.seatunnel.api.table.type.LocalTimeType;
+import org.apache.seatunnel.api.table.type.MapType;
+import org.apache.seatunnel.api.table.type.PrimitiveByteArrayType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.container.TestContainerId;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.format.avro.AvroDeserializationSchema;
+import org.apache.seatunnel.format.avro.AvroSerializationSchema;
+
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.Duration;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
+import static java.time.temporal.ChronoUnit.SECONDS;
+
+@Slf4j
+public class PulsarFormatIT extends TestSuiteBase implements TestResource {
+
+ private static final String PULSAR_IMAGE_NAME =
"apachepulsar/pulsar:2.3.1";
+ public static final String PULSAR_HOST = "pulsar.e2e.format";
+ private static final String FAKE_SOURCE_TOPIC =
"test_avro_topic_fake_source";
+ private static final String SOURCE_TOPIC = "test_avro_topic";
+
+ private static final SeaTunnelRowType FAKE_SOURCE_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {
+ "c_map",
+ "c_array",
+ "c_string",
+ "c_boolean",
+ "c_tinyint",
+ "c_smallint",
+ "c_int",
+ "c_bigint",
+ "c_float",
+ "c_double",
+ "c_bytes",
+ "c_date",
+ "c_decimal",
+ "c_timestamp"
+ },
+ new SeaTunnelDataType[] {
+ new MapType<>(BasicType.STRING_TYPE,
BasicType.STRING_TYPE),
+ ArrayType.INT_ARRAY_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.BOOLEAN_TYPE,
+ BasicType.BYTE_TYPE,
+ BasicType.SHORT_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.LONG_TYPE,
+ BasicType.FLOAT_TYPE,
+ BasicType.DOUBLE_TYPE,
+ PrimitiveByteArrayType.INSTANCE,
+ LocalTimeType.LOCAL_DATE_TYPE,
+ new DecimalType(38, 18),
+ LocalTimeType.LOCAL_DATE_TIME_TYPE
+ });
+
+ private static final SeaTunnelRowType SEATUNNEL_ROW_TYPE =
+ new SeaTunnelRowType(
+ new String[] {
+ "id",
+ "c_map",
+ "c_array",
+ "c_string",
+ "c_boolean",
+ "c_tinyint",
+ "c_smallint",
+ "c_int",
+ "c_bigint",
+ "c_float",
+ "c_double",
+ "c_decimal",
+ "c_bytes",
+ "c_date",
+ "c_timestamp"
+ },
+ new SeaTunnelDataType[] {
+ BasicType.LONG_TYPE,
+ new MapType<>(BasicType.STRING_TYPE,
BasicType.SHORT_TYPE),
+ ArrayType.BYTE_ARRAY_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.BOOLEAN_TYPE,
+ BasicType.BYTE_TYPE,
+ BasicType.SHORT_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.LONG_TYPE,
+ BasicType.FLOAT_TYPE,
+ BasicType.DOUBLE_TYPE,
+ new DecimalType(2, 1),
+ PrimitiveByteArrayType.INSTANCE,
+ LocalTimeType.LOCAL_DATE_TYPE,
+ LocalTimeType.LOCAL_DATE_TIME_TYPE
+ });
+
+ private PulsarContainer pulsarContainer;
+ private PulsarClient client;
+
+ @Override
+ @BeforeAll
+ public void startUp() throws Exception {
+ pulsarContainer =
+ PulsarContainerSupport.startPulsarContainer(
+ dockerClient,
+ PULSAR_IMAGE_NAME,
+ NETWORK,
+ PULSAR_HOST,
+ Duration.of(400, SECONDS));
+ Awaitility.given()
+ .ignoreExceptions()
+ .atLeast(100, TimeUnit.MILLISECONDS)
+ .pollInterval(500, TimeUnit.MILLISECONDS)
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(pulsarContainer.isRunning());
+ if (client == null) {
+ client =
+ PulsarClient.builder()
+
.serviceUrl(pulsarContainer.getPulsarBrokerUrl())
+ .build();
+ }
+ });
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ if (client != null) {
+ client.close();
+ }
+ pulsarContainer.close();
+ }
+
+ @TestTemplate
+ @DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
+ public void testFakeSourceToPulsarAvroFormat(TestContainer container)
+ throws IOException, InterruptedException {
+ Container.ExecResult execResult =
container.executeJob("/avro/fake_to_pulsar_avro.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+
+ CatalogTable catalogTable =
+ CatalogTableUtil.getCatalogTable("", "", "", "test",
FAKE_SOURCE_ROW_TYPE);
+ AvroDeserializationSchema avroDeserializationSchema =
+ new AvroDeserializationSchema(catalogTable);
+ List<SeaTunnelRow> rows = consumeAvroRows(FAKE_SOURCE_TOPIC,
avroDeserializationSchema);
+
+ Assertions.assertEquals(90, rows.size());
+ rows.forEach(
+ row -> {
+ Assertions.assertInstanceOf(Map.class, row.getField(0));
+ Assertions.assertInstanceOf(Integer[].class,
row.getField(1));
+ Assertions.assertInstanceOf(String.class, row.getField(2));
+ Assertions.assertEquals("fake_source_avro",
row.getField(2).toString());
+ Assertions.assertInstanceOf(Boolean.class,
row.getField(3));
+ Assertions.assertInstanceOf(Byte.class, row.getField(4));
+ Assertions.assertInstanceOf(Short.class, row.getField(5));
+ Assertions.assertInstanceOf(Integer.class,
row.getField(6));
+ Assertions.assertInstanceOf(Long.class, row.getField(7));
+ Assertions.assertInstanceOf(Float.class, row.getField(8));
+ Assertions.assertInstanceOf(Double.class, row.getField(9));
+ Assertions.assertInstanceOf(byte[].class,
row.getField(10));
+ Assertions.assertInstanceOf(LocalDate.class,
row.getField(11));
+ Assertions.assertInstanceOf(BigDecimal.class,
row.getField(12));
+ Assertions.assertInstanceOf(LocalDateTime.class,
row.getField(13));
+ });
+ }
+
+ @TestTemplate
+ @DisabledOnContainer(value = {TestContainerId.SPARK_2_4})
+ public void testPulsarAvroToAssert(TestContainer container) throws
Exception {
+ produceAvroTestData(0, 100);
+ Container.ExecResult execResult =
container.executeJob("/avro/pulsar_avro_to_assert.conf");
+ Assertions.assertEquals(0, execResult.getExitCode(),
execResult.getStderr());
+ }
+
+ private void produceAvroTestData(int start, int end) throws Exception {
+ AvroSerializationSchema serializationSchema =
+ new AvroSerializationSchema(SEATUNNEL_ROW_TYPE);
+ try (Producer<byte[]> producer =
+ client.newProducer(Schema.BYTES).topic(SOURCE_TOPIC).create())
{
+ for (int i = start; i < end; i++) {
+ SeaTunnelRow row =
+ new SeaTunnelRow(
+ new Object[] {
+ Long.valueOf(i),
+ Collections.singletonMap("key",
Short.parseShort("1")),
+ new Byte[] {Byte.parseByte("1")},
+ "string",
+ Boolean.FALSE,
+ Byte.parseByte("1"),
+ Short.parseShort("1"),
+ Integer.parseInt("1"),
+ Long.parseLong("1"),
+ Float.parseFloat("1.1"),
+ Double.parseDouble("1.1"),
+ BigDecimal.valueOf(11, 1),
+ "test".getBytes(),
+ LocalDate.of(2024, 1, 1),
+ LocalDateTime.of(2024, 1, 1, 12, 59, 23)
+ });
+ producer.send(serializationSchema.serialize(row));
+ }
+ }
+ }
+
+ private List<SeaTunnelRow> consumeAvroRows(
+ String topic, AvroDeserializationSchema deserializationSchema) {
+ List<SeaTunnelRow> rows = new ArrayList<>();
+ try (Consumer<byte[]> consumer =
+ client.newConsumer(Schema.BYTES)
+ .topic(topic)
+ .subscriptionName("PulsarFormatSubTest" + new
Random().nextInt())
+ .subscriptionType(SubscriptionType.Exclusive)
+
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe()) {
+ for (int i = 0; i < 90; i++) {
+ Message<byte[]> message = consumer.receive(30,
TimeUnit.SECONDS);
+ if (message == null) {
+ break;
+ }
+ rows.add(deserializationSchema.deserialize(message.getData()));
+ consumer.acknowledge(message.getMessageId());
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to consume pulsar avro data",
e);
+ }
+ return rows;
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/fake_to_pulsar_avro.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/fake_to_pulsar_avro.conf
new file mode 100644
index 0000000000..1fe201854c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/fake_to_pulsar_avro.conf
@@ -0,0 +1,63 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+
+ #spark config
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 1
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+}
+
+source {
+ FakeSource {
+ row.num = 90
+ string.template = ["fake_source_avro"]
+ schema = {
+ fields {
+ c_map = "map<string, string>"
+ c_array = "array<int>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_int = int
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_bytes = bytes
+ c_date = date
+ c_decimal = "decimal(38, 18)"
+ c_timestamp = timestamp
+ }
+ }
+ plugin_output = "test_avro_topic_fake_source"
+ }
+}
+
+sink {
+ pulsar {
+ topic = "test_avro_topic_fake_source"
+ client.service-url = "pulsar://pulsar.e2e.format:6650"
+ admin.service-url = "http://pulsar.e2e.format:8080";
+ format = avro
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/pulsar_avro_to_assert.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/pulsar_avro_to_assert.conf
new file mode 100644
index 0000000000..bf74eb190f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-pulsar-e2e/src/test/resources/avro/pulsar_avro_to_assert.conf
@@ -0,0 +1,101 @@
+#
+# 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.
+#
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 1
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+}
+
+source {
+ Pulsar {
+ client.service-url = "pulsar://pulsar.e2e.format:6650"
+ admin.service-url = "http://pulsar.e2e.format:8080";
+ subscription.name = "e2e-avro"
+ topic = "test_avro_topic"
+ cursor.startup.mode = "EARLIEST"
+ cursor.stop.mode = "LATEST"
+ format = avro
+ plugin_output = "pulsar_avro"
+ schema = {
+ fields {
+ id = bigint
+ c_map = "map<string, smallint>"
+ c_array = "array<tinyint>"
+ c_string = string
+ c_boolean = boolean
+ c_tinyint = tinyint
+ c_smallint = smallint
+ c_int = int
+ c_bigint = bigint
+ c_float = float
+ c_double = double
+ c_decimal = "decimal(2, 1)"
+ c_bytes = bytes
+ c_date = date
+ c_timestamp = timestamp
+ }
+ }
+ }
+}
+
+sink {
+ Assert {
+ plugin_input = "pulsar_avro"
+ rules =
+ {
+ field_rules = [
+ {
+ field_name = id
+ field_type = bigint
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN
+ rule_value = 0
+ },
+ {
+ rule_type = MAX
+ rule_value = 99
+ }
+ ]
+ },
+ {
+ field_name = c_string
+ field_type = string
+ field_value = [
+ {
+ rule_type = MIN_LENGTH
+ rule_value = 6
+ },
+ {
+ rule_type = MAX_LENGTH
+ rule_value = 6
+ }
+ ]
+ }
+ ]
+ }
+ }
+}
