luchunliang opened a new pull request, #12192:
URL: https://github.com/apache/inlong/pull/12192
Fixes #12191
### Motivation
- **On-the-fly re-encoding** — decode an incoming line (CSV / KV / PB /
JSON), apply SQL projection / filter / function calls (e.g.
`STR_TO_MAP(URL_DECODE(event_value), '&', '=')['HY50']`), then emit CSV / KV /
JSON to Pulsar.
- **Row explosion** — transforms like `$childIndex` + child-array unpacking
can turn one event into multiple downstream messages; the sink pipeline must
support that.
- **Filtering** — `WHERE` clauses can filter events entirely; the sink must
ack the event without emitting anything, without leaking transactions.
- **Consistency with Kafka side** — reduce operational surprise for users:
same `transformSql`, same encoding switch, same filter/explode semantics
regardless of the underlying MQ.
### Modifications
### Proposed change
Four files changed. All changes are additive; existing Pulsar deployments
without `transformSql` are unaffected.
#### 1) `inlong-common` : `PulsarSinkConfig`
Add encoding hints and defaults (mirroring `KafkaSinkConfig`):
```java
public static final String MESSAGE_TYPE_CSV = "csv";
public static final Character CSV_DEFAULT_DELIMITER = '|';
public static final String MESSAGE_TYPE_KV = "kv";
public static final Character KV_DEFAULT_ENTRYSPLITTER = '&';
public static final Character KV_DEFAULT_KVSPLITTER = '=';
public static final String MESSAGE_TYPE_JSON = "json";
private String messageType;
private Character delimiter;
private Character escapeChar;
private Character entrySplitter;
private Character kvSplitter;
```
The pre-existing fields (`pulsarTenant / namespace / topic / partitionNum`)
are kept intact.
#### 2) `PulsarIdConfig`
Add `dataFlowId` (aligned with `KafkaIdConfig`) so the handler can look up
the correct `TransformProcessor`:
- `Map<String,String>` constructor initializes `dataFlowId = uid`
(back-compat).
- `create(DataFlowConfig)` builder sets `dataFlowId =
dataFlowConfig.getDataflowId()`.
#### 3) `PulsarFederationSinkContext`
Introduce transform caching and encoder wiring:
```java
// Map<threadId, Map<dataFlowId, TransformProcessor>>
protected Map<Long, Map<String, TransformProcessor<String, ?>>> transformMap
= new ConcurrentHashMap<>();
public TransformProcessor<String, ?> getTransformProcessor(String
dataFlowId) { ... }
private Map<String, TransformProcessor<String, ?>>
reloadTransform(TaskConfig taskConfig) { ... }
private TransformProcessor<String, ?> createTransform(DataFlowConfig
dataFlowConfig) { ... }
private SinkEncoder<?> createSinkEncoder(SinkConfig sinkConfig) { ... }
```
Semantics:
- `reload()` uses `taskConfigJson` / `sortTaskConfigJson` for change
detection (via `replaceConfig(...)` in the base `SinkContext`).
- When `unifiedConfiguration` is on and config changes,
`transformMap.clear()` so stale processors on worker threads are dropped.
- `createTransform` builds a `TransformProcessor` using the base class
helpers `createTransformConfig` + `createSourceDecoder` + our new
`createSinkEncoder`.
- `createSinkEncoder` dispatches on `PulsarSinkConfig.messageType`:
- `csv` → `CsvSinkInfo(encodingType, delimiter [default '|'], escapeChar,
fieldInfos)`
- `kv` → `KvSinkInfo(encodingType, fieldInfos)` + `entrySplitter [default
'&']` + `kvSplitter [default '=']`
- `json` → `MapSinkInfo(encodingType, fieldInfos)`
- otherwise → default CSV encoder with `'|'`.
Flows without `transformSql` skip `TransformProcessor` construction entirely.
#### 4) `IEvent2PulsarRecordHandler` (**breaking** — see notes)
Signature updated to align with the Kafka handler and to allow 0/1/N outputs:
```java
public interface IEvent2PulsarRecordHandler {
List<byte[]> parse(PulsarFederationSinkContext context, ProfileEvent
event, PulsarIdConfig idConfig)
throws IOException;
}
```
#### 5) `DefaultEvent2PulsarRecordHandler`
Two branches, matching the Kafka side 1:1:
```java
@Override
public List<byte[]> parse(PulsarFederationSinkContext context, ProfileEvent
event, PulsarIdConfig idConfig)
throws IOException {
TransformProcessor<String, ?> processor =
context.getTransformProcessor(idConfig.getDataFlowId());
if (processor != null) {
return parseByTransform(context, event, processor);
}
return Arrays.asList(parseByBytes(event, idConfig));
}
```
- **`parseByTransform`** — builds `extParams` from
`context.getSinkContext().getParameters()` + `event.getHeaders()`, calls
`processor.transformForBytes(event.getBody(), extParams)`, converts each result:
- `String` → `getBytes()`
- `byte[]` → as-is
- other → `gson.toJson(...).getBytes()`
- **`parseByBytes`** — the original behavior is preserved: for `TEXT`
prepend `ftime + separator + extinfo + separator`, then append
`event.getBody()`; for `PB / JCE / UNKNOWN` just emit `event.getBody()`.
#### 6) `PulsarProducerCluster#send`
Adapted to the new `List<byte[]>` contract while keeping the "one
transaction per event" model:
- Resolve `PulsarIdConfig` from `event.getUid()`.
- Call `handler.parse(sinkContext, event, idConfig)`.
- If empty / null → `tx.commit(); event.ack(); tx.close();` (filter case).
- Otherwise send N messages in parallel and aggregate via `AtomicInteger
remaining` + `AtomicBoolean failed`:
- Every `sendAsync` callback records a per-message metric.
- When the last message's callback fires: if any failed → `tx.rollback()`,
else `tx.commit()` + `event.ack()`; finally `tx.close()`.
This preserves atomicity — an event either fully lands or fully rolls back —
even when it fans out to multiple Pulsar messages.
### Behavior matrix
| Config | `transformSql` present? | `messageType` | Result |
| --- | --- | --- | --- |
| Pulsar sink (existing users) | ❌ | any / unset | Same as today:
`parseByBytes` → 1 message per event |
| Pulsar sink | ✅ | `csv` | Transform → CSV encoder (custom delimiter /
escape) → N messages |
| Pulsar sink | ✅ | `kv` | Transform → KV encoder (custom entry / kv
splitter) → N messages |
| Pulsar sink | ✅ | `json` | Transform → MAP (JSON) encoder → N messages |
| Pulsar sink | ✅ | unset / unknown | Transform → default CSV encoder with
`'|'` |
### Backward compatibility
- **Wire config**: `PulsarSinkConfig` only adds fields/constants; old JSON
that doesn't set `messageType / delimiter / escapeChar / entrySplitter /
kvSplitter` deserializes fine.
- **Runtime behavior for existing flows**: when `transformSql` is empty,
`TransformProcessor` is not built, `parseByTransform` is not taken,
`parseByBytes` returns exactly one payload — semantically identical to today.
- **API break — internal SPI only**: `IEvent2PulsarRecordHandler#parse`
return type changes from `byte[]` to `List<byte[]>` and gains `PulsarIdConfig`.
This interface is a sort-standalone internal SPI (not published to
`inlong-common`), and the only known implementation
`DefaultEvent2PulsarRecordHandler` is updated in the same change. Users who
plugged in a custom handler via the `eventHandler` common property will need a
small adaptation:
```java
// before
byte[] out = doStuff(event);
// after
return Arrays.asList(doStuff(event));
```
- No public API on `SinkContext` / `PulsarFederationSinkContext` is removed
or renamed; only additions.
### Risks / Notes
- **Custom `eventHandler` implementations must adapt** to the new interface
signature. Because a wrong signature will surface as a `NoSuchMethodError` at
boot, this failure is loud (not silent).
- **Per-message vs per-event metrics**: for a fanout event,
`addSendResultMetric` is now invoked once per outbound message, which slightly
changes metric cardinality (more accurate, but different from previous
behavior). Existing dashboards that count "sink send success = event count" may
need to switch to "sink send success = message count".
- **Transactional atomicity in fanout**: if any of the N sub-sends fails,
the whole transaction rolls back (all N messages are considered unsent from the
sort perspective, even the ones the broker already acked). This mirrors the
current single-message semantics and keeps ack behavior predictable at the cost
of possible duplicate delivery under partial failures — same trade-off the
Kafka side already takes.
- **JSON encoding path** uses `MapSinkInfo`
(`SinkEncoderFactory.createMapEncoder`) to keep parity with Kafka. If you need
strict record-JSON in the future (schema-aware), that is a separate follow-up.
### Files changed
-
`inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/sink/PulsarSinkConfig.java`
-
`inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java`
-
`inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java`
-
`inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/IEvent2PulsarRecordHandler.java`
-
`inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/DefaultEvent2PulsarRecordHandler.java`
-
`inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java`
### Checklist
- [x] Behavior of existing Pulsar sink pipelines (no `transformSql`) is
preserved
- [x] `PulsarSinkConfig` is JSON back-compat (only new fields/constants
added)
- [x] `PulsarFederationSinkContext` caches `TransformProcessor` per worker
thread and clears it on config change
- [x] `DefaultEvent2PulsarRecordHandler` supports 0 / 1 / N output rows
- [x] `PulsarProducerCluster` sends N messages under one transaction and
commits/rolls back atomically
- [x] Parity with `KafkaFederationSinkContext` +
`DefaultEvent2KafkaRecordHandler` on the transform code path
### Verifying this change
*(Please pick either of the following options)*
- [ ] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [ ] This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads
(10MB)*
- *Extended integration test for recovery after broker failure*
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs
/ not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a follow-up
issue for adding the documentation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
