From ac0c5a7d08c0a197ab622a82eae94e83a0164410 Mon Sep 17 00:00:00 2001 From: ChunLiang Lu <946240095@qq.com> Date: Wed, 26 Aug 2026 11:32:39 +0800 Subject: [PATCH] [INLONG-12191][Sort] Support InLong Transform SDK on the Pulsar sink pipeline --- .../sort/dataflow/sink/PulsarSinkConfig.java | 12 ++ .../sort/standalone/sink/cls/ClsCallback.java | 4 + .../DefaultEvent2PulsarRecordHandler.java | 66 +++++++--- .../pulsar/IEvent2PulsarRecordHandler.java | 17 +-- .../pulsar/PulsarFederationSinkContext.java | 118 +++++++++++++++++- .../sink/pulsar/PulsarIdConfig.java | 3 + .../sink/pulsar/PulsarProducerCluster.java | 58 +++++---- 7 files changed, 230 insertions(+), 48 deletions(-) diff --git a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/sink/PulsarSinkConfig.java b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/sink/PulsarSinkConfig.java index 9142019d205..478e74fe6cc 100644 --- a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/sink/PulsarSinkConfig.java +++ b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/sink/PulsarSinkConfig.java @@ -24,8 +24,20 @@ @Data public class PulsarSinkConfig extends SinkConfig { + 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 pulsarTenant; private String namespace; private String topic; private Integer partitionNum; + private String messageType; + private Character delimiter; + private Character escapeChar; + private Character entrySplitter; + private Character kvSplitter; } diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsCallback.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsCallback.java index cef41438769..827b063f4f9 100644 --- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsCallback.java +++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsCallback.java @@ -20,6 +20,7 @@ import org.apache.inlong.sort.standalone.channel.ProfileEvent; import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory; +import com.google.gson.Gson; import com.tencentcloudapi.cls.producer.Callback; import com.tencentcloudapi.cls.producer.Result; import com.tencentcloudapi.cls.producer.common.Attempt; @@ -40,6 +41,7 @@ public class ClsCallback implements Callback { private final ClsSinkContext context; private final ProfileEvent event; private final String topicId; + private Gson gson = new Gson(); /** * Constructor. @@ -80,6 +82,8 @@ private void onSuccess() { * @param result Send result. */ private void onFailed(Result result) { + LOG.error("onFail,groupId:{},data:{},result:{}", event.getInlongGroupId(), new String(event.getBody()), + gson.toJson(result)); if (isRetryable(result.getReservedAttempts())) { tx.rollback(); tx.close(); diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/DefaultEvent2PulsarRecordHandler.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/DefaultEvent2PulsarRecordHandler.java index f8016f25781..229b5c05b79 100644 --- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/DefaultEvent2PulsarRecordHandler.java +++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/DefaultEvent2PulsarRecordHandler.java @@ -18,15 +18,22 @@ package org.apache.inlong.sort.standalone.sink.pulsar; import org.apache.inlong.sdk.commons.protocol.EventConstants; +import org.apache.inlong.sdk.transform.process.TransformProcessor; import org.apache.inlong.sort.standalone.channel.ProfileEvent; import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory; +import com.google.gson.Gson; import org.slf4j.Logger; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.text.SimpleDateFormat; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; /** * @@ -40,25 +47,57 @@ public class DefaultEvent2PulsarRecordHandler implements IEvent2PulsarRecordHand protected final ByteArrayOutputStream outMsg = new ByteArrayOutputStream(); protected final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"); protected final Date currentDate = new Date(); + protected final Gson gson = new Gson(); /** * parse - * - * @param context - * @param event - * @return byte array - * @throws IOException + * + * @param context pulsar federation sink context + * @param event raw profile event + * @param idConfig id config resolved from event uid + * @return a list of message payload byte arrays + * @throws IOException on any IO error */ @Override - public byte[] parse(PulsarFederationSinkContext context, ProfileEvent event) + public List parse(PulsarFederationSinkContext context, ProfileEvent event, PulsarIdConfig idConfig) throws IOException { - String uid = event.getUid(); - PulsarIdConfig idConfig = context.getIdConfig(uid); - if (idConfig == null) { - context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis()); - LOG.error("Can not find the id config:{}", uid); - return null; + TransformProcessor processor = context.getTransformProcessor(idConfig.getDataFlowId()); + if (processor != null) { + return this.parseByTransform(context, event, processor); + } else { + byte[] record = this.parseByBytes(event, idConfig); + return Arrays.asList(record); + } + } + + public List parseByTransform(PulsarFederationSinkContext context, ProfileEvent event, + TransformProcessor processor) throws IOException { + // extParams + Map extParams = new ConcurrentHashMap<>(); + extParams.putAll(context.getSinkContext().getParameters()); + event.getHeaders().forEach((k, v) -> extParams.put(k, v)); + // transform + List results = processor.transformForBytes(event.getBody(), extParams); + if (results == null) { + return new ArrayList<>(); + } + // build + List records = new ArrayList<>(results.size()); + for (Object result : results) { + byte[] msgContent; + if (result instanceof String) { + msgContent = result.toString().getBytes(); + } else if (result instanceof byte[]) { + msgContent = (byte[]) result; + } else { + msgContent = gson.toJson(result).getBytes(); + } + records.add(msgContent); } + return records; + } + + public byte[] parseByBytes(ProfileEvent event, PulsarIdConfig idConfig) throws IOException { String delimiter = idConfig.getSeparator(); byte separator = (byte) delimiter.charAt(0); outMsg.reset(); @@ -80,8 +119,7 @@ public byte[] parse(PulsarFederationSinkContext context, ProfileEvent event) break; } outMsg.write(event.getBody()); - byte[] msgContent = outMsg.toByteArray(); - return msgContent; + return outMsg.toByteArray(); } /** diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/IEvent2PulsarRecordHandler.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/IEvent2PulsarRecordHandler.java index 694b0d757f7..329d466dd99 100644 --- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/IEvent2PulsarRecordHandler.java +++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/IEvent2PulsarRecordHandler.java @@ -20,6 +20,7 @@ import org.apache.inlong.sort.standalone.channel.ProfileEvent; import java.io.IOException; +import java.util.List; /** * @@ -28,12 +29,14 @@ public interface IEvent2PulsarRecordHandler { /** - * parse - * - * @param context - * @param event - * @return ProducerRecord - * @throws IOException + * parse the event into one or more pulsar message payloads. + * + * @param context pulsar federation sink context + * @param event raw profile event + * @param idConfig id config resolved from event uid + * @return a list of message payload byte arrays; empty/null means filtered + * @throws IOException on any IO error */ - byte[] parse(PulsarFederationSinkContext context, ProfileEvent event) throws IOException; + List parse(PulsarFederationSinkContext context, ProfileEvent event, PulsarIdConfig idConfig) + throws IOException; } diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java index d17139335e0..b8e88c41de9 100644 --- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java +++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java @@ -19,8 +19,18 @@ import org.apache.inlong.common.pojo.sort.ClusterTagConfig; import org.apache.inlong.common.pojo.sort.TaskConfig; +import org.apache.inlong.common.pojo.sort.dataflow.DataFlowConfig; +import org.apache.inlong.common.pojo.sort.dataflow.sink.PulsarSinkConfig; +import org.apache.inlong.common.pojo.sort.dataflow.sink.SinkConfig; import org.apache.inlong.common.pojo.sort.node.PulsarNodeConfig; import org.apache.inlong.common.pojo.sortstandalone.SortTaskConfig; +import org.apache.inlong.sdk.transform.encode.SinkEncoder; +import org.apache.inlong.sdk.transform.encode.SinkEncoderFactory; +import org.apache.inlong.sdk.transform.pojo.CsvSinkInfo; +import org.apache.inlong.sdk.transform.pojo.FieldInfo; +import org.apache.inlong.sdk.transform.pojo.KvSinkInfo; +import org.apache.inlong.sdk.transform.pojo.MapSinkInfo; +import org.apache.inlong.sdk.transform.process.TransformProcessor; import org.apache.inlong.sort.standalone.channel.ProfileEvent; import org.apache.inlong.sort.standalone.config.holder.CommonPropertiesHolder; import org.apache.inlong.sort.standalone.config.holder.SortClusterConfigHolder; @@ -33,7 +43,9 @@ import org.apache.inlong.sort.standalone.sink.SinkContext; import org.apache.inlong.sort.standalone.utils.InlongLoggerFactory; +import com.google.common.collect.ImmutableMap; import org.apache.commons.lang3.ClassUtils; +import org.apache.commons.lang3.StringUtils; import org.apache.flume.Channel; import org.apache.flume.Context; import org.slf4j.Logger; @@ -45,6 +57,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +@SuppressWarnings("deprecation") public class PulsarFederationSinkContext extends SinkContext { public static final Logger LOG = InlongLoggerFactory.getLogger(PulsarFederationSinkContext.class); @@ -53,6 +66,9 @@ public class PulsarFederationSinkContext extends SinkContext { private PulsarNodeConfig pulsarNodeConfig; private CacheClusterConfig cacheClusterConfig; + // Map> + protected Map>> transformMap = new ConcurrentHashMap<>(); + public PulsarFederationSinkContext(String sinkName, Context context, Channel channel) { super(sinkName, context, channel); } @@ -65,8 +81,9 @@ public void reload() { LOG.error("newSortTaskConfig is null."); return; } - if ((this.taskConfig != null && this.taskConfig.equals(newTaskConfig)) - && (this.sortTaskConfig != null && this.sortTaskConfig.equals(newSortTaskConfig))) { + if ((newTaskConfig == null || StringUtils.equals(this.taskConfigJson, gson.toJson(newTaskConfig))) + && (newSortTaskConfig == null + || StringUtils.equals(this.sortTaskConfigJson, gson.toJson(newSortTaskConfig)))) { LOG.info("Same sortTaskConfig, do nothing."); return; } @@ -78,8 +95,7 @@ public void reload() { } } - this.taskConfig = newTaskConfig; - this.sortTaskConfig = newSortTaskConfig; + this.replaceConfig(newTaskConfig, newSortTaskConfig); CacheClusterConfig clusterConfig = new CacheClusterConfig(); clusterConfig.setClusterName(this.taskName); @@ -91,7 +107,12 @@ public void reload() { Map fromTaskConfig = fromTaskConfig(taskConfig); Map fromSortTaskConfig = fromSortTaskConfig(sortTaskConfig); SortConfigMetricReporter.reportClusterDiff(clusterId, taskName, fromTaskConfig, fromSortTaskConfig); - idConfigMap = unifiedConfiguration ? fromTaskConfig : fromSortTaskConfig; + if (unifiedConfiguration) { + this.idConfigMap = fromTaskConfig; + this.transformMap.clear(); + } else { + this.idConfigMap = fromSortTaskConfig; + } } catch (Throwable e) { LOG.error(e.getMessage(), e); } @@ -248,4 +269,91 @@ public IEvent2PulsarRecordHandler createEventHandler() { } return null; } + + public TransformProcessor getTransformProcessor(String dataFlowId) { + Long threadId = Thread.currentThread().getId(); + Map> transformProcessors = this.transformMap.get(threadId); + if (transformProcessors == null) { + transformProcessors = reloadTransform(taskConfig); + this.transformMap.put(threadId, transformProcessors); + } + return transformProcessors.get(dataFlowId); + } + + private Map> reloadTransform(TaskConfig taskConfig) { + ImmutableMap.Builder> builder = new ImmutableMap.Builder<>(); + + taskConfig.getClusterTagConfigs() + .stream() + .map(ClusterTagConfig::getDataFlowConfigs) + .flatMap(Collection::stream) + .forEach(flow -> { + if (StringUtils.isEmpty(flow.getTransformSql())) { + return; + } + TransformProcessor transformProcessor = createTransform(flow); + if (transformProcessor == null) { + return; + } + builder.put(flow.getDataflowId(), + transformProcessor); + }); + + return builder.build(); + } + + private TransformProcessor createTransform(DataFlowConfig dataFlowConfig) { + try { + LOG.info("try to create transform:{}", dataFlowConfig.toString()); + return TransformProcessor.create( + createTransformConfig(dataFlowConfig), + createSourceDecoder(dataFlowConfig.getSourceConfig()), + createSinkEncoder(dataFlowConfig.getSinkConfig())); + } catch (Exception e) { + LOG.error("failed to reload transform of dataflow={}, ex={}", dataFlowConfig.getDataflowId(), + e.getMessage(), e); + return null; + } + } + + private SinkEncoder createSinkEncoder(SinkConfig sinkConfig) { + if (!(sinkConfig instanceof PulsarSinkConfig)) { + throw new IllegalArgumentException("sinkInfo must be an instance of PulsarSinkConfig"); + } + PulsarSinkConfig sSinkConfig = (PulsarSinkConfig) sinkConfig; + List fieldInfos = sSinkConfig.getFieldConfigs() + .stream() + .map(config -> new FieldInfo(config.getName(), deriveTypeConverter(config.getFormatInfo()))) + .collect(Collectors.toList()); + + if (StringUtils.equalsIgnoreCase(PulsarSinkConfig.MESSAGE_TYPE_CSV, sSinkConfig.getMessageType())) { + Character delimiter = sSinkConfig.getDelimiter(); + if (delimiter == null) { + delimiter = PulsarSinkConfig.CSV_DEFAULT_DELIMITER; + } + CsvSinkInfo sinkInfo = new CsvSinkInfo(sinkConfig.getEncodingType(), delimiter, sSinkConfig.getEscapeChar(), + fieldInfos); + return SinkEncoderFactory.createCsvEncoder(sinkInfo); + } else if (StringUtils.equalsIgnoreCase(PulsarSinkConfig.MESSAGE_TYPE_KV, sSinkConfig.getMessageType())) { + Character entrySplitter = sSinkConfig.getEntrySplitter(); + if (entrySplitter == null) { + entrySplitter = PulsarSinkConfig.KV_DEFAULT_ENTRYSPLITTER; + } + Character kvSplitter = sSinkConfig.getKvSplitter(); + if (kvSplitter == null) { + kvSplitter = PulsarSinkConfig.KV_DEFAULT_KVSPLITTER; + } + KvSinkInfo sinkInfo = new KvSinkInfo(sinkConfig.getEncodingType(), fieldInfos); + sinkInfo.setEntryDelimiter(entrySplitter); + sinkInfo.setKvDelimiter(kvSplitter); + return SinkEncoderFactory.createKvEncoder(sinkInfo); + } else if (StringUtils.equalsIgnoreCase(PulsarSinkConfig.MESSAGE_TYPE_JSON, sSinkConfig.getMessageType())) { + MapSinkInfo sinkInfo = new MapSinkInfo(sinkConfig.getEncodingType(), fieldInfos); + return SinkEncoderFactory.createMapEncoder(sinkInfo); + } else { + CsvSinkInfo sinkInfo = new CsvSinkInfo(sinkConfig.getEncodingType(), PulsarSinkConfig.CSV_DEFAULT_DELIMITER, + null, fieldInfos); + return SinkEncoderFactory.createCsvEncoder(sinkInfo); + } + } } diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java index 6dc48d25117..79a23bea96e 100644 --- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java +++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java @@ -55,6 +55,7 @@ public class PulsarIdConfig extends IdConfig { private String topic; private DataTypeEnum dataType; private DataFlowConfig dataFlowConfig; + private String dataFlowId; public PulsarIdConfig(Map idParam) { this.inlongGroupId = idParam.get(Constants.INLONG_GROUP_ID); @@ -64,6 +65,7 @@ public PulsarIdConfig(Map idParam) { this.topic = idParam.getOrDefault(Constants.TOPIC, uid); this.dataType = DataTypeEnum .convert(idParam.getOrDefault(PulsarIdConfig.KEY_DATA_TYPE, DataTypeEnum.TEXT.getType())); + this.dataFlowId = this.uid; } public static PulsarIdConfig create(DataFlowConfig dataFlowConfig) { @@ -97,6 +99,7 @@ public static PulsarIdConfig create(DataFlowConfig dataFlowConfig) { .dataType(dataType) .separator(separator) .dataFlowConfig(dataFlowConfig) + .dataFlowId(dataFlowConfig.getDataflowId()) .build(); } diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java index 4f2f80e0ab5..f4f60d42071 100644 --- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java +++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarProducerCluster.java @@ -43,11 +43,14 @@ import java.io.IOException; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; /** * @@ -276,35 +279,46 @@ public boolean send(ProfileEvent profileEvent, Transaction tx) throws IOExceptio throw new IllegalStateException(); } - // sendAsync - byte[] sendBytes = this.handler.parse(sinkContext, profileEvent); + // resolve idConfig and parse into one or more message payloads + String uid = profileEvent.getUid(); + PulsarIdConfig idConfig = sinkContext.getIdConfig(uid); + List records = this.handler.parse(sinkContext, profileEvent, idConfig); // check - if (sendBytes == null) { + if (records == null || records.isEmpty()) { tx.commit(); profileEvent.ack(); tx.close(); return true; } long sendTime = System.currentTimeMillis(); - CompletableFuture future = producer.newMessage() - .properties(headers) - .value(sendBytes) - .sendAsync(); - // callback - future.whenCompleteAsync((msgId, ex) -> { - if (ex != null) { - LOG.error("Send fail:{}", ex.getMessage()); - LOG.error(ex.getMessage(), ex); - tx.rollback(); - tx.close(); - sinkContext.addSendResultMetric(profileEvent, topic, false, sendTime); - } else { - tx.commit(); - tx.close(); - sinkContext.addSendResultMetric(profileEvent, topic, true, sendTime); - profileEvent.ack(); - } - }); + // send all records, commit tx after all completed + final int total = records.size(); + final AtomicInteger remaining = new AtomicInteger(total); + final AtomicBoolean failed = new AtomicBoolean(false); + for (byte[] sendBytes : records) { + CompletableFuture future = producer.newMessage() + .properties(headers) + .value(sendBytes) + .sendAsync(); + future.whenCompleteAsync((msgId, ex) -> { + if (ex != null) { + LOG.error("Send fail, uid:{}, topic:{}, error:{}", uid, topic, ex.getMessage(), ex); + failed.compareAndSet(false, true); + sinkContext.addSendResultMetric(profileEvent, topic, false, sendTime); + } else { + sinkContext.addSendResultMetric(profileEvent, topic, true, sendTime); + } + if (remaining.decrementAndGet() == 0) { + if (failed.get()) { + tx.rollback(); + } else { + tx.commit(); + profileEvent.ack(); + } + tx.close(); + } + }); + } return true; }