Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
luchunliang merged 1 commit into from
Aug 26, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions ...mmon/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/sink/PulsarSinkConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
4 changes: 4 additions & 0 deletions ...andalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/cls/ClsCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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.
Expand Down Expand Up @@ -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();
Expand Down
66 changes: 52 additions & 14 deletions .../java/org/apache/inlong/sort/standalone/sink/pulsar/DefaultEvent2PulsarRecordHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

/**
*
Expand All @@ -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<byte[]> 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<String, ?> 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<byte[]> parseByTransform(PulsarFederationSinkContext context, ProfileEvent event,
TransformProcessor<String, ?> processor) throws IOException {
// extParams
Map<String, Object> 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<byte[]> 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();
Expand All @@ -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();
}

/**
Expand Down
17 changes: 10 additions & 7 deletions ...c/main/java/org/apache/inlong/sort/standalone/sink/pulsar/IEvent2PulsarRecordHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.apache.inlong.sort.standalone.channel.ProfileEvent;

import java.io.IOException;
import java.util.List;

/**
*
Expand All @@ -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<byte[]> parse(PulsarFederationSinkContext context, ProfileEvent event, PulsarIdConfig idConfig)
throws IOException;
}
118 changes: 113 additions & 5 deletions .../main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarFederationSinkContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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);
Expand All @@ -53,6 +66,9 @@ public class PulsarFederationSinkContext extends SinkContext {
private PulsarNodeConfig pulsarNodeConfig;
private CacheClusterConfig cacheClusterConfig;

// Map<threadId, Map<dataFlowId,TransformProcess>>
protected Map<Long, Map<String, TransformProcessor<String, ?>>> transformMap = new ConcurrentHashMap<>();

public PulsarFederationSinkContext(String sinkName, Context context, Channel channel) {
super(sinkName, context, channel);
}
Expand All @@ -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;
}
Expand All @@ -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);
Expand All @@ -91,7 +107,12 @@ public void reload() {
Map<String, PulsarIdConfig> fromTaskConfig = fromTaskConfig(taskConfig);
Map<String, PulsarIdConfig> 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);
}
Expand Down Expand Up @@ -248,4 +269,91 @@ public IEvent2PulsarRecordHandler createEventHandler() {
}
return null;
}

public TransformProcessor<String, ?> getTransformProcessor(String dataFlowId) {
Long threadId = Thread.currentThread().getId();
Map<String, TransformProcessor<String, ?>> transformProcessors = this.transformMap.get(threadId);
if (transformProcessors == null) {
transformProcessors = reloadTransform(taskConfig);
this.transformMap.put(threadId, transformProcessors);
}
return transformProcessors.get(dataFlowId);
}

private Map<String, TransformProcessor<String, ?>> reloadTransform(TaskConfig taskConfig) {
ImmutableMap.Builder<String, TransformProcessor<String, ?>> builder = new ImmutableMap.Builder<>();

taskConfig.getClusterTagConfigs()
.stream()
.map(ClusterTagConfig::getDataFlowConfigs)
.flatMap(Collection::stream)
.forEach(flow -> {
if (StringUtils.isEmpty(flow.getTransformSql())) {
return;
}
TransformProcessor<String, ?> transformProcessor = createTransform(flow);
if (transformProcessor == null) {
return;
}
builder.put(flow.getDataflowId(),
transformProcessor);
});

return builder.build();
}

private TransformProcessor<String, ?> 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<FieldInfo> 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);
}
}
}
3 changes: 3 additions & 0 deletions ...ne-source/src/main/java/org/apache/inlong/sort/standalone/sink/pulsar/PulsarIdConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ public class PulsarIdConfig extends IdConfig {
private String topic;
private DataTypeEnum dataType;
private DataFlowConfig dataFlowConfig;
private String dataFlowId;

public PulsarIdConfig(Map<String, String> idParam) {
this.inlongGroupId = idParam.get(Constants.INLONG_GROUP_ID);
Expand All @@ -64,6 +65,7 @@ public PulsarIdConfig(Map<String, String> 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) {
Expand Down Expand Up @@ -97,6 +99,7 @@ public static PulsarIdConfig create(DataFlowConfig dataFlowConfig) {
.dataType(dataType)
.separator(separator)
.dataFlowConfig(dataFlowConfig)
.dataFlowId(dataFlowConfig.getDataflowId())
.build();

}
Expand Down
Loading
Loading