luchunliang opened a new issue, #12191: URL: https://github.com/apache/inlong/issues/12191
### Description Sort-standalone already integrates the InLong Transform SDK on the **Kafka** sink side (`KafkaFederationSinkContext` + `DefaultEvent2KafkaRecordHandler`), so a user can attach a `transformSql` on a data flow and have each event decoded → transformed → re-encoded (CSV / KV / JSON) before being emitted to Kafka. The **Pulsar** sink side does not yet have this capability: - `PulsarSinkConfig` only carries basic topic-related fields (`pulsarTenant / namespace / topic / partitionNum`) and has no encoding hints. - `PulsarFederationSinkContext` builds an id-config map only; it does not build/cache any `TransformProcessor`. - `DefaultEvent2PulsarRecordHandler#parse` returns a single `byte[]` and does the "prepend ftime + extinfo + separator" concatenation only. There is no path for a `transformSql` to take effect, and there is no support for 1-input → N-output rows (a very common outcome of `transformForBytes(...)`). - `PulsarProducerCluster#send` is hard-coded around exactly one payload per event and one commit/rollback per event. This issue proposes to bring Pulsar to feature-parity with Kafka on this pipeline, so that: 1. A `transformSql` on a `DataFlowConfig` targeting a Pulsar sink is honored. 2. One inbound `ProfileEvent` can be turned into any number (0 / 1 / N) of Pulsar messages, respecting filter semantics and preserving transactional ack. 3. The output encoding of the transformed payload can be picked via `PulsarSinkConfig.messageType` (`csv` / `kv` / `json`) with sane defaults, aligned with `KafkaSinkConfig`. ### Use case - **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. ### Are you willing to submit PR? - [x] Yes, I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
