[INLONG-12191][Sort] Support InLong Transform SDK on the Pulsar sink pipeline

[luchunliang]

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):

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:

// 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:

public interface IEvent2PulsarRecordHandler {
    List<byte[]> parse(PulsarFederationSinkContext context, ProfileEvent event, PulsarIdConfig idConfig)
            throws IOException;
}

5) DefaultEvent2PulsarRecordHandler

Two branches, matching the Kafka side 1:1:

@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:

  // 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

• Behavior of existing Pulsar sink pipelines (no transformSql) is preserved
• PulsarSinkConfig is JSON back-compat (only new fields/constants added)
• PulsarFederationSinkContext caches TransformProcessor per worker thread and clears it on config change
• DefaultEvent2PulsarRecordHandler supports 0 / 1 / N output rows
• PulsarProducerCluster sends N messages under one transaction and commits/rolls back atomically
• 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