You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2022/02/23 04:34:19 UTC
[incubator-inlong] branch master updated: [INLONG-2382][Sort-sdk] Support Pulsar consumer of PB compression cache message protocol (#2575)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new e575c43 [INLONG-2382][Sort-sdk] Support Pulsar consumer of PB compression cache message protocol (#2575)
e575c43 is described below
commit e575c434603328f80ae87d889b37e35ade01ddc5
Author: wardli <95...@users.noreply.github.com>
AuthorDate: Wed Feb 23 12:34:12 2022 +0800
[INLONG-2382][Sort-sdk] Support Pulsar consumer of PB compression cache message protocol (#2575)
---
inlong-sdk/sort-sdk/pom.xml | 48 ++--
.../apache/inlong/sdk/sort/api/Deserializer.java | 33 +++
.../inlong/sdk/sort/api/InLongTopicFetcher.java | 4 +
.../inlong/sdk/sort/entity/InLongMessage.java | 48 +++-
.../inlong/sdk/sort/entity/MessageRecord.java | 2 -
.../sdk/sort/impl/InLongTopicManagerImpl.java | 10 +-
.../sdk/sort/impl/decode/MessageDeserializer.java | 147 ++++++++++++
.../sort/impl/pulsar/InLongPulsarFetcherImpl.java | 9 +-
.../sdk/sort/stat/SortClientStateCounter.java | 12 +-
.../org/apache/inlong/sdk/sort/util/Utils.java | 265 +++++++++++++++++++++
.../sort/impl/decode/MessageDeserializerTest.java | 171 +++++++++++++
.../standalone/source/sortsdk/FetchCallback.java | 4 +-
12 files changed, 701 insertions(+), 52 deletions(-)
diff --git a/inlong-sdk/sort-sdk/pom.xml b/inlong-sdk/sort-sdk/pom.xml
index 475e550..ea5d8b2 100644
--- a/inlong-sdk/sort-sdk/pom.xml
+++ b/inlong-sdk/sort-sdk/pom.xml
@@ -76,52 +76,46 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<plugin.compile.version>3.8.1</plugin.compile.version>
+ <pulsar.version>2.8.1</pulsar.version>
+ <httpclient.version>4.5.13</httpclient.version>
+ <snappy-java.version>1.1.7.3</snappy-java.version>
</properties>
<dependencies>
<dependency>
- <groupId>org.apache.pulsar</groupId>
- <artifactId>pulsar-client</artifactId>
- <version>2.7.2</version>
- </dependency>
-
- <dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- <version>4.5.13</version>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>inlong-common</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
- <artifactId>powermock-module-junit4</artifactId>
- <groupId>org.powermock</groupId>
- <scope>test</scope>
- <version>2.0.2</version>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>tubemq-client</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
- <artifactId>powermock-api-mockito2</artifactId>
- <groupId>org.powermock</groupId>
- <scope>test</scope>
- <version>2.0.2</version>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-client</artifactId>
+ <version>${pulsar.version}</version>
</dependency>
<dependency>
- <artifactId>mockito-core</artifactId>
- <groupId>org.mockito</groupId>
- <scope>test</scope>
- <version>2.23.0</version>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>sdk-common</artifactId>
+ <version>${project.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.inlong</groupId>
- <artifactId>inlong-common</artifactId>
- <version>${project.version}</version>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.inlong</groupId>
- <artifactId>tubemq-client</artifactId>
- <version>${project.version}</version>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ <version>${snappy-java.version}</version>
</dependency>
</dependencies>
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/Deserializer.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/Deserializer.java
new file mode 100644
index 0000000..fd1d245
--- /dev/null
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/Deserializer.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.inlong.sdk.sort.api;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import org.apache.inlong.sdk.sort.entity.InLongMessage;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+
+public interface Deserializer extends Serializable {
+
+ List<InLongMessage> deserialize(ClientContext context, InLongTopic inLongTopic, Map<String, String> headers,
+ byte[] data)
+ throws Exception;
+}
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
index b113a82..68af40e 100644
--- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/api/InLongTopicFetcher.java
@@ -18,11 +18,14 @@
package org.apache.inlong.sdk.sort.api;
import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.impl.decode.MessageDeserializer;
public abstract class InLongTopicFetcher {
protected InLongTopic inLongTopic;
protected ClientContext context;
+ protected Deserializer deserializer;
+ protected volatile Thread fetchThread;
protected volatile boolean closed = false;
protected volatile boolean isStopConsume = false;
// use for empty topic to sleep
@@ -32,6 +35,7 @@ public abstract class InLongTopicFetcher {
public InLongTopicFetcher(InLongTopic inLongTopic, ClientContext context) {
this.inLongTopic = inLongTopic;
this.context = context;
+ this.deserializer = new MessageDeserializer();
}
public abstract boolean init(Object client);
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java
index f332812..e7bd3b3 100644
--- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/InLongMessage.java
@@ -21,23 +21,49 @@ import java.util.Map;
public class InLongMessage {
- private final Map<String, String> msgHeader;
- private byte[] data;
+ private String inlongGroupId;
+ private String inlongStreamId;
+ private long msgTime; //message generation time, milliseconds
+ private String sourceIp; // agent ip of message generation
+ private final Map<String, String> params;
+ private final byte[] body;
- public InLongMessage(byte[] data, Map<String, String> msgHeader) {
- this.data = data;
- this.msgHeader = msgHeader;
+ public InLongMessage(byte[] body, Map<String, String> params) {
+ this.body = body;
+ this.params = params;
}
- public byte[] getData() {
- return data;
+ public InLongMessage(String inlongGroupId, String inlongStreamId, long msgTime, String sourceIp, byte[] body,
+ Map<String, String> params) {
+ this.inlongGroupId = inlongGroupId;
+ this.inlongStreamId = inlongStreamId;
+ this.msgTime = msgTime;
+ this.sourceIp = sourceIp;
+ this.body = body;
+ this.params = params;
}
- public void setData(byte[] data) {
- this.data = data;
+ public long getMsgTime() {
+ return msgTime;
}
- public Map<String, String> getMsgHeader() {
- return msgHeader;
+ public String getSourceIp() {
+ return sourceIp;
+ }
+
+ public byte[] getBody() {
+ return body;
+ }
+
+ public Map<String, String> getParams() {
+ return params;
+ }
+
+ public String getInlongGroupId() {
+ return inlongGroupId;
+ }
+
+ public String getInlongStreamId() {
+ return inlongStreamId;
}
}
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
index 6917426..104ddf1 100644
--- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/entity/MessageRecord.java
@@ -22,9 +22,7 @@ import java.util.List;
public class MessageRecord {
private final String msgKey;
-
private final List<InLongMessage> msgs;
-
private final String offset;
private final long recTime;
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
index 2f310e5..312283e 100644
--- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/InLongTopicManagerImpl.java
@@ -350,9 +350,10 @@ public class InLongTopicManagerImpl extends InLongTopicManager {
.authentication(AuthenticationFactory.token(inLongTopic.getInLongCluster().getToken()))
.build();
pulsarClients.put(inLongTopic.getInLongCluster().getClusterId(), pulsarClient);
- logger.info("create pulsar client succ {} {} {}", inLongTopic.getInLongCluster().getClusterId(),
+ logger.info("create pulsar client succ {} {} {}",
+ new String[]{inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getInLongCluster().getBootstraps(),
- inLongTopic.getInLongCluster().getToken());
+ inLongTopic.getInLongCluster().getToken()});
} catch (Exception e) {
logger.error("create pulsar client error {}", inLongTopic);
logger.error(e.getMessage(), e);
@@ -377,9 +378,10 @@ public class InLongTopicManagerImpl extends InLongTopicManager {
TubeConsumerCreater tubeConsumerCreater = new TubeConsumerCreater(messageSessionFactory,
tubeConfig);
tubeFactories.put(inLongTopic.getInLongCluster().getClusterId(), tubeConsumerCreater);
- logger.info("create tube client succ {} {} {}", inLongTopic.getInLongCluster().getClusterId(),
+ logger.info("create tube client succ {} {} {}",
+ new String[]{inLongTopic.getInLongCluster().getClusterId(),
inLongTopic.getInLongCluster().getBootstraps(),
- inLongTopic.getInLongCluster().getToken());
+ inLongTopic.getInLongCluster().getToken()});
} catch (Exception e) {
logger.error("create tube client error {}", inLongTopic);
logger.error(e.getMessage(), e);
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java
new file mode 100644
index 0000000..86e9ecd
--- /dev/null
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializer.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.inlong.sdk.sort.impl.decode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MapFieldEntry;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessageObj;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessageObjs;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.Deserializer;
+import org.apache.inlong.sdk.sort.entity.InLongMessage;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.util.Utils;
+
+public class MessageDeserializer implements Deserializer {
+
+ private static final int MESSAGE_VERSION_NONE = 0;
+ private static final int MESSAGE_VERSION_PB = 1;
+ private static final int COMPRESS_TYPE_NONE = 0;
+ private static final int COMPRESS_TYPE_GZIP = 1;
+ private static final int COMPRESS_TYPE_SNAPPY = 2;
+ private static final String VERSION_KEY = "version";
+ private static final String COMPRESS_TYPE_KEY = "compressType";
+ private static final String MSG_TIME_KEY = "msgTime";
+ private static final String SOURCE_IP_KEY = "sourceIp";
+ private static final String INLONG_GROUPID_KEY = "inlongGroupId";
+ private static final String INLONG_STREAMID_KEY = "inlongStreamId";
+
+ public MessageDeserializer() {
+ }
+
+ @Override
+ public List<InLongMessage> deserialize(ClientContext context, InLongTopic inLongTopic, Map<String, String> headers,
+ byte[] data) throws Exception {
+
+ //1. version
+ int version = Integer.parseInt(headers.getOrDefault(VERSION_KEY, "0"));
+ switch (version) {
+ case MESSAGE_VERSION_NONE: {
+ return decode(context, inLongTopic, data, headers);
+ }
+ case MESSAGE_VERSION_PB: {
+ return decodePB(context, inLongTopic, data, headers);
+ }
+ default:
+ throw new IllegalArgumentException("Unknown version type:" + version);
+ }
+ }
+
+ private List<InLongMessage> decode(ClientContext context, InLongTopic inLongTopic, byte[] msgBytes,
+ Map<String, String> headers) {
+ long msgTime = Long.parseLong(headers.getOrDefault(MSG_TIME_KEY, "0"));
+ String sourceIp = headers.getOrDefault(SOURCE_IP_KEY, "");
+ String inlongGroupId = headers.getOrDefault(INLONG_GROUPID_KEY, "");
+ String inlongStreamId = headers.getOrDefault(INLONG_STREAMID_KEY, "");
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addDecompressionConsumeSize(msgBytes.length);
+ return Collections
+ .singletonList(new InLongMessage(inlongGroupId, inlongStreamId, msgTime, sourceIp, msgBytes, headers));
+ }
+
+ /**
+ * uncompress and decode byte[]
+ *
+ * @param msgBytes byte[]
+ * @return {@link MessageObjs}
+ */
+ private List<InLongMessage> decodePB(ClientContext context, InLongTopic inLongTopic, byte[] msgBytes,
+ Map<String, String> headers) throws IOException {
+ int compressType = Integer.parseInt(headers.getOrDefault(COMPRESS_TYPE_KEY, "0"));
+ String inlongGroupId = headers.getOrDefault(INLONG_GROUPID_KEY, "");
+ String inlongStreamId = headers.getOrDefault(INLONG_STREAMID_KEY, "");
+ switch (compressType) {
+ case COMPRESS_TYPE_NONE: {
+ return transformMessageObjs(context, inLongTopic, MessageObjs.parseFrom(msgBytes), inlongGroupId,
+ inlongStreamId);
+ }
+ case COMPRESS_TYPE_SNAPPY: {
+ byte[] values = Utils.snappyDecompress(msgBytes, 0, msgBytes.length);
+ return transformMessageObjs(context, inLongTopic, MessageObjs.parseFrom(values), inlongGroupId,
+ inlongStreamId);
+ }
+ case COMPRESS_TYPE_GZIP: {
+ byte[] values = Utils.gzipDecompress(msgBytes, 0, msgBytes.length);
+ return transformMessageObjs(context, inLongTopic, MessageObjs.parseFrom(values), inlongGroupId,
+ inlongStreamId);
+ }
+ default:
+ throw new IllegalArgumentException("Unknown compress type:" + compressType);
+ }
+ }
+
+ /**
+ * transform MessageObjs to SortSdkMessage
+ *
+ * @param messageObjs {@link MessageObjs}
+ * @return {@link List<InLongMessage>}
+ */
+ private List<InLongMessage> transformMessageObjs(ClientContext context, InLongTopic inLongTopic,
+ MessageObjs messageObjs, String inlongGroupId,
+ String inlongStreamId) {
+ if (null == messageObjs) {
+ return null;
+ }
+ List<InLongMessage> inLongMessages = new ArrayList<>();
+ for (MessageObj messageObj : messageObjs.getMsgsList()) {
+ List<MapFieldEntry> mapFieldEntries = messageObj.getParamsList();
+ Map<String, String> headers = new HashMap<>();
+ for (MapFieldEntry mapFieldEntry : mapFieldEntries) {
+ headers.put(mapFieldEntry.getKey(), mapFieldEntry.getValue());
+ }
+ InLongMessage inLongMessage = new InLongMessage(inlongGroupId, inlongStreamId, messageObj.getMsgTime(),
+ messageObj.getSourceIp(),
+ messageObj.getBody().toByteArray(), headers);
+ inLongMessages.add(inLongMessage);
+ context.getStatManager()
+ .getStatistics(context.getConfig().getSortTaskId(),
+ inLongTopic.getInLongCluster().getClusterId(), inLongTopic.getTopic())
+ .addDecompressionConsumeSize(inLongMessage.getBody().length);
+ }
+ return inLongMessages;
+ }
+}
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java
index 9803936..c179756 100644
--- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/impl/pulsar/InLongPulsarFetcherImpl.java
@@ -19,7 +19,6 @@ package org.apache.inlong.sdk.sort.impl.pulsar;
import java.util.ArrayList;
import java.util.Base64;
-import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
@@ -48,11 +47,8 @@ public class InLongPulsarFetcherImpl extends InLongTopicFetcher {
private final Logger logger = LoggerFactory.getLogger(InLongPulsarFetcherImpl.class);
private final ReentrantReadWriteLock mainLock = new ReentrantReadWriteLock(true);
private final ConcurrentHashMap<String, MessageId> offsetCache = new ConcurrentHashMap<>();
-
private Consumer<byte[]> consumer;
- private volatile Thread fetchThread;
-
public InLongPulsarFetcherImpl(InLongTopic inLongTopic,
ClientContext context) {
super(inLongTopic, context);
@@ -291,8 +287,11 @@ public class InLongPulsarFetcherImpl extends InLongTopicFetcher {
String offsetKey = getOffset(msg.getMessageId());
offsetCache.put(offsetKey, msg.getMessageId());
+ List<InLongMessage> inLongMessages = deserializer
+ .deserialize(context, inLongTopic, msg.getProperties(), msg.getData());
+
msgs.add(new MessageRecord(inLongTopic.getTopicKey(),
- Collections.singletonList(new InLongMessage(msg.getData(), msg.getProperties())),
+ inLongMessages,
offsetKey, System.currentTimeMillis()));
context.getStatManager()
.getStatistics(context.getConfig().getSortTaskId(),
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
index e9352b6..ee3c226 100644
--- a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/stat/SortClientStateCounter.java
@@ -21,7 +21,7 @@ import java.util.concurrent.atomic.AtomicLongArray;
public class SortClientStateCounter {
- private final AtomicLongArray count = new AtomicLongArray(20);
+ private final AtomicLongArray count = new AtomicLongArray(21);
public String sortTaskId;
public String cacheClusterId;
public String topic;
@@ -287,4 +287,14 @@ public class SortClientStateCounter {
count.getAndAdd(19, num);
return this;
}
+
+ /**
+ * count decompression consume size
+ * @param num long
+ * @return {@link SortClientStateCounter}
+ */
+ public SortClientStateCounter addDecompressionConsumeSize(long num) {
+ count.getAndAdd(20, num);
+ return this;
+ }
}
diff --git a/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/util/Utils.java b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/util/Utils.java
new file mode 100644
index 0000000..56700ea
--- /dev/null
+++ b/inlong-sdk/sort-sdk/src/main/java/org/apache/inlong/sdk/sort/util/Utils.java
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.inlong.sdk.sort.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.nio.ByteBuffer;
+import java.text.NumberFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+import org.xerial.snappy.Snappy;
+
+public class Utils {
+
+ public static final long MINUTE_IN_MILLIS = 60 * 1000;
+
+ public static final long HOUR_IN_MILLIS = 60 * 60 * 1000;
+
+ public static final long DAY_IN_MILLIS = 24 * 60 * 60 * 1000;
+
+
+ private static ConcurrentHashMap<String, AtomicLong> timeMap = new ConcurrentHashMap<String, AtomicLong>();
+
+ public static double toKB(long bytes) {
+ return bytes / 1024d;
+ }
+
+ public static double toMB(long bytes) {
+ return bytes / 1024d / 1024d;
+ }
+
+ public static double toGB(long bytes) {
+ return bytes / 1024d / 1024d / 1024d;
+ }
+
+ public static String ipInt2String(int ipInt) {
+ byte[] bytes = new byte[4];
+ bytes[0] = (byte) ((ipInt >>> 24) & 0xFF);
+ bytes[1] = (byte) ((ipInt >>> 16) & 0xFF);
+ bytes[2] = (byte) ((ipInt >>> 8) & 0xFF);
+ bytes[3] = (byte) (ipInt & 0xFF);
+ return "" + (bytes[0] & 0x00FF) + '.' + (bytes[1] & 0x00FF) + '.' + (bytes[2] & 0x00FF)
+ + '.' + (bytes[3] & 0x00FF);
+ }
+
+ public static String bytesToStr(byte[] value, String charsetName, String defaultValue) {
+ if (value == null || value.length == 0) {
+ return defaultValue;
+ }
+ try {
+ String str = new String(value, charsetName);
+ return str;
+ } catch (UnsupportedEncodingException e) {
+ return defaultValue;
+ }
+ }
+
+ public static <T> Set<T> toSet(T... values) {
+ Set<T> result = new HashSet<T>();
+ for (T v : values) {
+ result.add(v);
+ }
+ return result;
+ }
+
+ public static <T> List<T> toList(T... values) {
+ List<T> result = new ArrayList<T>();
+ for (T v : values) {
+ result.add(v);
+ }
+ return result;
+ }
+
+ public static String exceptionMsg(Throwable exception) {
+ if (exception == null) {
+ return "";
+ }
+ try {
+ StringBuilder sb = new StringBuilder();
+ sb.append(exception.getClass().getSimpleName() + ",");
+ sb.append(exception.getMessage());
+
+ Throwable cause = exception.getCause();
+ if (cause != null) {
+ sb.append("," + cause.getMessage());
+ }
+ return sb.toString();
+ } catch (Throwable e) {
+ return "";
+ }
+ }
+
+ public static String doubleToStr(double value, int fractionDigits) {
+ NumberFormat format = NumberFormat.getInstance();
+ format.setMinimumFractionDigits(fractionDigits);
+ String str = format.format(value);
+ return str;
+ }
+
+ public static String getUUID() {
+ UUID uuid = UUID.randomUUID();
+ return uuid.toString().replaceAll("-", "");
+
+ }
+
+ public static String toStr(String splitChar, Object... objs) {
+ StringBuilder sb = new StringBuilder();
+ int size = objs.length;
+ int lastIndex = size - 1;
+ for (int i = 0; i < size; i++) {
+ sb.append(objs[i]);
+ if (i < lastIndex) {
+ sb.append(splitChar);
+ }
+ }
+ return sb.toString();
+ }
+
+ public static <E> List<List<E>> splitList(List<E> list, int splitSize) {
+ List<List<E>> result = new ArrayList<List<E>>();
+ if (list == null || list.size() <= splitSize) {
+ return result;
+ }
+
+ int size = 0;
+ List<E> subList = new ArrayList<E>();
+ for (E e : list) {
+ if ((size++) > splitSize) {
+ result.add(subList);
+
+ size = 0;
+ subList = new ArrayList<E>();
+ } else {
+ subList.add(e);
+ }
+ }
+
+ return result;
+ }
+
+ public static byte[] compressGZip(byte[] data) {
+ if (data == null || data.length == 0) {
+ return null;
+ }
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ GZIPOutputStream gzip;
+ try {
+ gzip = new GZIPOutputStream(out);
+ gzip.write(data);
+ gzip.close();
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return out.toByteArray();
+ }
+
+ public static byte[] gzipDecompress(byte[] data, int startOffset, int dataLength)
+ throws IOException {
+ GZIPInputStream gzip = null;
+ int initBufSize = 8192;
+ try {
+ //int inputLength=(data.length-(startOffset==0? 0:startOffset));
+ ByteArrayInputStream in = new ByteArrayInputStream(data, startOffset, dataLength);
+ gzip = new GZIPInputStream(in, 8192);
+
+ byte[] buf = new byte[initBufSize];
+ int offset = 0;
+ do {
+ int readNum = gzip.read(buf, offset, 8192);
+ if (readNum == -1) {
+ break;
+ } else {
+ offset += readNum;
+ if (offset + 8192 > (buf.length - 1)) {
+ buf = Arrays.copyOf(buf, buf.length * 2);
+ }
+ }
+ } while (true);
+
+ return Arrays.copyOf(buf, offset);
+ } finally {
+ if (gzip != null) {
+ gzip.close();
+ }
+ }
+ }
+
+ public static byte[] snappyCompress(byte[] data) throws IOException {
+ if (data == null || data.length == 0) {
+ return null;
+ }
+
+ return Snappy.compress(data);
+ }
+
+ public static byte[] snappyDecompress(byte[] data, int startOffset, int dataLength)
+ throws IOException {
+ int needSize = Snappy.uncompressedLength(data, startOffset, dataLength);
+ byte[] result = new byte[needSize];
+
+ Snappy.uncompress(data, startOffset, dataLength, result, 0);
+ return result;
+ }
+
+ public static byte[] toBytes(ByteBuffer buffer) {
+ if (buffer == null || buffer.remaining() == 0) {
+ return new byte[0];
+ }
+ int remainBytes = buffer.remaining();
+ byte[] value = new byte[buffer.remaining()];
+
+ if (buffer.hasArray()) {
+ System.arraycopy(buffer.array(), buffer.arrayOffset(), value, 0, remainBytes);
+ } else {
+ buffer.get(value, 0, remainBytes);
+ }
+ return value;
+ }
+
+ public static double div(double d1, double d2) {
+ return (d2 != 0 ? (d1 / d2) : 0);
+ }
+
+ public static String nowToStr() {
+ return StringUtil.formatDate(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
+ }
+
+ public static <K, V> V getMapValue(Map<K, V> map, K k, V defaultV) {
+ if (map == null) {
+ return defaultV;
+ }
+ V v = map.get(k);
+ return (v != null ? v : defaultV);
+ }
+
+}
diff --git a/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializerTest.java b/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializerTest.java
new file mode 100644
index 0000000..55f3ffd
--- /dev/null
+++ b/inlong-sdk/sort-sdk/src/test/java/org/apache/inlong/sdk/sort/impl/decode/MessageDeserializerTest.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.sdk.sort.impl.decode;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.powermock.api.mockito.PowerMockito.when;
+
+import com.google.protobuf.ByteString;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MapFieldEntry;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessageObj;
+import org.apache.inlong.sdk.commons.protocol.ProxySdk.MessageObjs;
+import org.apache.inlong.sdk.sort.api.ClientContext;
+import org.apache.inlong.sdk.sort.api.SortClientConfig;
+import org.apache.inlong.sdk.sort.entity.CacheZoneCluster;
+import org.apache.inlong.sdk.sort.entity.InLongMessage;
+import org.apache.inlong.sdk.sort.entity.InLongTopic;
+import org.apache.inlong.sdk.sort.stat.SortClientStateCounter;
+import org.apache.inlong.sdk.sort.stat.StatManager;
+import org.apache.inlong.sdk.sort.util.Utils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore("javax.management.*")
+public class MessageDeserializerTest {
+
+ private MessageDeserializer messageDeserializer;
+ private Map<String, String> headers;
+ private ClientContext context;
+ private SortClientConfig sortClientConfig;
+ private StatManager statManager;
+ private SortClientStateCounter sortClientStateCounter;
+ private InLongTopic inLongTopic;
+ private String testData;
+ private MessageObjs messageObjs;
+
+ /**
+ * setUp
+ */
+ @Before
+ public void setUp() throws Exception {
+ messageDeserializer = new MessageDeserializer();
+ headers = new HashMap<>();
+ context = PowerMockito.mock(ClientContext.class);
+ sortClientConfig = PowerMockito.mock(SortClientConfig.class);
+ statManager = PowerMockito.mock(StatManager.class);
+ inLongTopic = new InLongTopic();
+ inLongTopic.setTopic("testTopic");
+
+ CacheZoneCluster cacheZoneCluster = new CacheZoneCluster("clusterId", "bootstraps", "token");
+ inLongTopic.setInLongCluster(cacheZoneCluster);
+ when(context.getConfig()).thenReturn(sortClientConfig);
+ when(context.getStatManager()).thenReturn(statManager);
+ sortClientStateCounter = new SortClientStateCounter("sortTaskId", cacheZoneCluster.getClusterId(),
+ inLongTopic.getTopic(), 0);
+ when(statManager.getStatistics(anyString(), anyString(), anyString())).thenReturn(sortClientStateCounter);
+ when(sortClientConfig.getSortTaskId()).thenReturn("sortTaskId");
+ }
+
+ @Test
+ public void testDeserializeVersion0() {
+ try {
+ // test version == 0
+ headers.put("version", "0");
+ testData = "test data";
+ List<InLongMessage> deserialize = messageDeserializer
+ .deserialize(context, inLongTopic, headers, testData.getBytes());
+ Assert.assertEquals(1, deserialize.size());
+ Assert.assertEquals(testData, new String(deserialize.get(0).getBody()));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void testDeserializeVersion1CompressionType0() {
+ try {
+ // test version == 1
+ prepareTestMessageObjs();
+ // non compression
+ headers.put("compressType", "0");
+
+ List<InLongMessage> deserialize = messageDeserializer
+ .deserialize(context, inLongTopic, headers, messageObjs.toByteArray());
+ Assert.assertEquals(2, deserialize.size());
+ Assert.assertEquals(testData, new String(deserialize.get(0).getBody()));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void testDeserializeVersion1CompressionType1() {
+ try {
+ // test version == 1
+ prepareTestMessageObjs();
+ // compression gzip
+ headers.put("compressType", "1");
+
+ byte[] testDataByteArray = Utils.compressGZip(messageObjs.toByteArray());
+
+ List<InLongMessage> deserialize = messageDeserializer
+ .deserialize(context, inLongTopic, headers, testDataByteArray);
+ Assert.assertEquals(2, deserialize.size());
+ Assert.assertEquals(testData, new String(deserialize.get(0).getBody()));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ @Test
+ public void testDeserializeVersion1CompressionType2() {
+ try {
+ // test version == 1
+ prepareTestMessageObjs();
+ // compression snappy
+ headers.put("compressType", "2");
+
+ byte[] testDataByteArray = Utils.snappyCompress(messageObjs.toByteArray());
+
+ List<InLongMessage> deserialize = messageDeserializer
+ .deserialize(context, inLongTopic, headers, testDataByteArray);
+ Assert.assertEquals(2, deserialize.size());
+ Assert.assertEquals(testData, new String(deserialize.get(0).getBody()));
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+
+ private void prepareTestMessageObjs() {
+ headers.put("version", "1");
+ testData = "test data";
+ long messageTime = System.currentTimeMillis();
+ MapFieldEntry mapFieldEntry = MapFieldEntry.newBuilder().setKey("key").setValue("val").build();
+ MessageObj messageObj1 = MessageObj.newBuilder().setBody(ByteString.copyFrom(testData.getBytes()))
+ .setMsgTime(messageTime)
+ .setSourceIp("ip1")
+ .addParams(mapFieldEntry)
+ .build();
+ MessageObj messageObj2 = MessageObj.newBuilder().setBody(ByteString.copyFrom(testData.getBytes()))
+ .setMsgTime(messageTime)
+ .setSourceIp("ip2")
+ .addParams(mapFieldEntry)
+ .build();
+
+ messageObjs = MessageObjs.newBuilder().addMsgs(messageObj1).addMsgs(messageObj2).build();
+ }
+}
\ No newline at end of file
diff --git a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
index 53e96dd..52a8dcb 100644
--- a/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
+++ b/inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java
@@ -106,8 +106,8 @@ public class FetchCallback implements ReadCallback {
//TODO fix here
final SubscribeFetchResult result = SubscribeFetchResult.Factory
.create(sortId, messageRecord.getMsgKey(), messageRecord.getOffset(),
- inLongMessage.getMsgHeader(), messageRecord.getRecTime(),
- inLongMessage.getData());
+ inLongMessage.getParams(), messageRecord.getRecTime(),
+ inLongMessage.getBody());
final ProfileEvent profileEvent = new ProfileEvent(result.getBody(), result.getHeaders());
channelProcessor.processEvent(profileEvent);
context.reportToMetric(profileEvent, sortId, "-", SortSdkSourceContext.FetchResult.SUCCESS);