Skip to content
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

Jump to bottom

[INLONG-11469][SDK] Optimize the single message processing logic in the EncodeObject class #11470

Merged
merged 1 commit into from
Nov 8, 2024
Merged

[INLONG-11469][SDK] Optimize the single message processing logic in the EncodeObject class #11470

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
1 change: 1 addition & 0 deletions inlong-common/src/main/java/org/apache/inlong/common/msg/AttributeConstants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ public interface AttributeConstants {

String SEPARATOR = "&";
String KEY_VALUE_SEPARATOR = "=";
String LINE_FEED_SEP = "\n";

/**
* group id
Expand Down
68 changes: 30 additions & 38 deletions ...sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/DefaultMessageSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -229,15 +230,6 @@ private String attemptSendMessageIndex(Function<Sender, String> sendOperation) {
return sendIndexResult;
}

@Deprecated
public SendResult sendMessage(byte[] body, String attributes, String msgUUID,
long timeout, TimeUnit timeUnit) {
Function<Sender, SendResult> sendOperation =
(sender) -> sender.syncSendMessage(new EncodeObject(body, attributes, idGenerator.getNextId()), msgUUID,
timeout, timeUnit);
return attemptSendMessage(sendOperation);
}

public SendResult sendMessage(byte[] body, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit) {
return sendMessage(body, groupId, streamId, dt, msgUUID, timeout, timeUnit, false);
Expand Down Expand Up @@ -275,8 +267,9 @@ public SendResult sendMessage(byte[] body, String groupId, String streamId, long
boolean isCompressEnd = (isCompress && (body.length > cpsSize));

if (msgtype == 7 || msgtype == 8) {
EncodeObject encodeObject = new EncodeObject(body, msgtype, isCompressEnd, isReport,
isGroupIdTransfer, dt / 1000, idGenerator.getNextInt(), groupId, streamId, proxySend);
EncodeObject encodeObject =
new EncodeObject(Collections.singletonList(body), msgtype, isCompressEnd, isReport,
isGroupIdTransfer, dt / 1000, idGenerator.getNextInt(), groupId, streamId, proxySend);
encodeObject.setSupportLF(isSupportLF);
Function<Sender, SendResult> sendOperation = (sender) -> sender.syncSendMessage(encodeObject, msgUUID,
timeout, timeUnit);
Expand All @@ -289,13 +282,13 @@ public SendResult sendMessage(byte[] body, String groupId, String streamId, long
final long finalDt = dt;
Function<Sender, SendResult> sendOperation;
if (isCompressEnd) {
sendOperation = (sender) -> sender.syncSendMessage(new EncodeObject(body,
sendOperation = (sender) -> sender.syncSendMessage(new EncodeObject(Collections.singletonList(body),
"groupId=" + groupId + "&streamId=" + streamId + "&dt=" + finalDt + "&cp=snappy"
+ finalProxySend,
idGenerator.getNextId(), this.getMsgtype(),
true, groupId), msgUUID, timeout, timeUnit);
} else {
sendOperation = (sender) -> sender.syncSendMessage(new EncodeObject(body,
sendOperation = (sender) -> sender.syncSendMessage(new EncodeObject(Collections.singletonList(body),
"groupId=" + groupId + "&streamId=" + streamId + "&dt=" + finalDt
+ finalProxySend,
idGenerator.getNextId(), this.getMsgtype(),
Expand Down Expand Up @@ -347,9 +340,10 @@ public SendResult sendMessage(byte[] body, String groupId, String streamId, long
boolean isCompressEnd = (isCompress && (body.length > cpsSize));

if (msgtype == 7 || msgtype == 8) {
EncodeObject encodeObject = new EncodeObject(body, msgtype, isCompressEnd, isReport,
isGroupIdTransfer, dt / 1000,
idGenerator.getNextInt(), groupId, streamId, attrs.toString());
EncodeObject encodeObject =
new EncodeObject(Collections.singletonList(body), msgtype, isCompressEnd, isReport,
isGroupIdTransfer, dt / 1000,
idGenerator.getNextInt(), groupId, streamId, attrs.toString());
encodeObject.setSupportLF(isSupportLF);
Function<Sender, SendResult> sendOperation = (sender) -> sender.syncSendMessage(encodeObject, msgUUID,
timeout, timeUnit);
Expand All @@ -358,13 +352,15 @@ public SendResult sendMessage(byte[] body, String groupId, String streamId, long
attrs.append("&groupId=").append(groupId).append("&streamId=").append(streamId).append("&dt=").append(dt);
if (isCompressEnd) {
attrs.append("&cp=snappy");
Function<Sender, SendResult> sendOperation = (sender) -> sender.syncSendMessage(new EncodeObject(body,
attrs.toString(), idGenerator.getNextId(), this.getMsgtype(), true, groupId),
Function<Sender, SendResult> sendOperation = (sender) -> sender.syncSendMessage(
new EncodeObject(Collections.singletonList(body),
attrs.toString(), idGenerator.getNextId(), this.getMsgtype(), true, groupId),
msgUUID, timeout, timeUnit);
return attemptSendMessage(sendOperation);
} else {
Function<Sender, SendResult> sendOperation = (sender) -> sender.syncSendMessage(new EncodeObject(body,
attrs.toString(), idGenerator.getNextId(), this.getMsgtype(), false, groupId),
Function<Sender, SendResult> sendOperation = (sender) -> sender.syncSendMessage(
new EncodeObject(Collections.singletonList(body),
attrs.toString(), idGenerator.getNextId(), this.getMsgtype(), false, groupId),
msgUUID, timeout, timeUnit);
return attemptSendMessage(sendOperation);
}
Expand Down Expand Up @@ -503,13 +499,6 @@ public SendResult sendMessage(List<byte[]> bodyList, String groupId, String stre
return null;
}

@Deprecated
public void asyncSendMessage(SendMessageCallback callback, byte[] body, String attributes,
String msgUUID, long timeout, TimeUnit timeUnit) throws ProxysdkException {
sender.asyncSendMessage(new EncodeObject(body, attributes, idGenerator.getNextId()),
callback, msgUUID, timeout, timeUnit);
}

public void asyncSendMessage(SendMessageCallback callback, byte[] body, String groupId, String streamId, long dt,
String msgUUID, long timeout, TimeUnit timeUnit) throws ProxysdkException {
asyncSendMessage(callback, body, groupId, streamId, dt, msgUUID, timeout, timeUnit, false);
Expand Down Expand Up @@ -546,23 +535,24 @@ public void asyncSendMessage(SendMessageCallback callback, byte[] body, String g
}
boolean isCompressEnd = (isCompress && (body.length > cpsSize));
if (msgtype == 7 || msgtype == 8) {
EncodeObject encodeObject = new EncodeObject(body, this.getMsgtype(), isCompressEnd, isReport,
isGroupIdTransfer, dt / 1000, idGenerator.getNextInt(),
groupId, streamId, proxySend);
EncodeObject encodeObject =
new EncodeObject(Collections.singletonList(body), this.getMsgtype(), isCompressEnd, isReport,
isGroupIdTransfer, dt / 1000, idGenerator.getNextInt(),
groupId, streamId, proxySend);
encodeObject.setSupportLF(isSupportLF);
sender.asyncSendMessage(encodeObject, callback, msgUUID, timeout, timeUnit);
} else if (msgtype == 3 || msgtype == 5) {
if (isCompressEnd) {
if (isProxySend) {
proxySend = "&" + proxySend;
}
sender.asyncSendMessage(new EncodeObject(body, "groupId="
sender.asyncSendMessage(new EncodeObject(Collections.singletonList(body), "groupId="
+ groupId + "&streamId=" + streamId + "&dt=" + dt + "&cp=snappy" + proxySend,
idGenerator.getNextId(), this.getMsgtype(), true, groupId),
callback, msgUUID, timeout, timeUnit);
} else {
sender.asyncSendMessage(
new EncodeObject(body, "groupId=" + groupId + "&streamId="
new EncodeObject(Collections.singletonList(body), "groupId=" + groupId + "&streamId="
+ streamId + "&dt=" + dt + proxySend, idGenerator.getNextId(),
this.getMsgtype(), false, groupId),
callback,
Expand Down Expand Up @@ -611,21 +601,23 @@ public void asyncSendMessage(SendMessageCallback callback, byte[] body, String g

boolean isCompressEnd = (isCompress && (body.length > cpsSize));
if (msgtype == 7 || msgtype == 8) {
EncodeObject encodeObject = new EncodeObject(body, this.getMsgtype(), isCompressEnd,
isReport, isGroupIdTransfer, dt / 1000, idGenerator.getNextInt(),
groupId, streamId, attrs.toString());
EncodeObject encodeObject =
new EncodeObject(Collections.singletonList(body), this.getMsgtype(), isCompressEnd,
isReport, isGroupIdTransfer, dt / 1000, idGenerator.getNextInt(),
groupId, streamId, attrs.toString());
encodeObject.setSupportLF(isSupportLF);
sender.asyncSendMessage(encodeObject, callback, msgUUID, timeout, timeUnit);
} else if (msgtype == 3 || msgtype == 5) {
attrs.append("&groupId=").append(groupId).append("&streamId=").append(streamId).append("&dt=").append(dt);
if (isCompressEnd) {
attrs.append("&cp=snappy");
sender.asyncSendMessage(new EncodeObject(body, attrs.toString(),
sender.asyncSendMessage(new EncodeObject(Collections.singletonList(body), attrs.toString(),
idGenerator.getNextId(), this.getMsgtype(), true, groupId),
callback, msgUUID, timeout, timeUnit);
} else {
sender.asyncSendMessage(new EncodeObject(body, attrs.toString(), idGenerator.getNextId(),
this.getMsgtype(), false, groupId),
sender.asyncSendMessage(
new EncodeObject(Collections.singletonList(body), attrs.toString(), idGenerator.getNextId(),
this.getMsgtype(), false, groupId),
callback, msgUUID, timeout, timeUnit);
}
}
Expand Down
6 changes: 3 additions & 3 deletions inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/LoadBalance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,10 @@ public enum LoadBalance {
WEIGHT_RANDOM("weight random", 3),
WEIGHT_ROBIN("weight robin", 4);

private String name;
private int index;
private final String name;
private final int index;

private LoadBalance(String name, int index) {
LoadBalance(String name, int index) {
this.name = name;
this.index = index;
}
Expand Down
44 changes: 9 additions & 35 deletions inlong-sdk/dataproxy-sdk/src/main/java/org/apache/inlong/sdk/dataproxy/MessageSender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,25 +27,13 @@

public interface MessageSender {

/**
* This method provides a synchronized function which you want to send data
* with extra attributes except groupId,streamId,dt,etc
* This method is deprecated,we suggest you don't use it.
*
* @param body The data will be sent
* @param attributes The attributes you want to add
*/
@Deprecated
public SendResult sendMessage(byte[] body, String attributes, String msgUUID,
long timeout, TimeUnit timeUnit);

/**
* This method provides a synchronized function which you want to send data without packing
*
* @param body The data will be sent
*
*/
public SendResult sendMessage(byte[] body, String groupId, String streamId, long dt, String msgUUID,
SendResult sendMessage(byte[] body, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit);

/**
Expand All @@ -57,7 +45,7 @@ public SendResult sendMessage(byte[] body, String groupId, String streamId, long
* @param extraAttrMap The attributes you want to add,
* and each element of extraAttrMap contains a pair like attrKey,attrValue
*/
public SendResult sendMessage(byte[] body, String groupId, String streamId, long dt, String msgUUID,
SendResult sendMessage(byte[] body, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit, Map<String, String> extraAttrMap);

/**
Expand All @@ -66,7 +54,7 @@ public SendResult sendMessage(byte[] body, String groupId, String streamId, long
*
* @param bodyList The data will be sent,which is a collection consisting of byte arrays
*/
public SendResult sendMessage(List<byte[]> bodyList, String groupId, String streamId, long dt, String msgUUID,
SendResult sendMessage(List<byte[]> bodyList, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit);

/**
Expand All @@ -78,23 +66,9 @@ public SendResult sendMessage(List<byte[]> bodyList, String groupId, String stre
* @param extraAttrMap The attributes you want to add,
* and each element of extraAttrMap contains a pair like attrKey,attrValue
*/
public SendResult sendMessage(List<byte[]> bodyList, String groupId, String streamId, long dt, String msgUUID,
SendResult sendMessage(List<byte[]> bodyList, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit, Map<String, String> extraAttrMap);

/**
* This method provides an asynchronized function which you want to send data
* with extra attributes except groupId,streamId,dt,etc
* This method is deprecated,we suggest you don't use it.
*
*
* @param body The data will be sent
* @param attributes The attributes you want to add
*/
@Deprecated
public void asyncSendMessage(SendMessageCallback callback,
byte[] body, String attributes, String msgUUID,
long timeout, TimeUnit timeUnit) throws ProxysdkException;

/**
* This method provides a synchronized function which you want to send data without packing
* with extra attributes except groupId,streamId,dt,etc
Expand All @@ -104,7 +78,7 @@ public void asyncSendMessage(SendMessageCallback callback,
* @param extraAttrMap The attributes you want to add,
* and each element of extraAttrMap contains a pair like attrKey,attrValue
*/
public void asyncSendMessage(SendMessageCallback callback,
void asyncSendMessage(SendMessageCallback callback,
byte[] body, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit,
Map<String, String> extraAttrMap) throws ProxysdkException;
Expand All @@ -116,7 +90,7 @@ public void asyncSendMessage(SendMessageCallback callback,
* @param callback The implementation of callback function
* @param body The data will be sent
*/
public void asyncSendMessage(SendMessageCallback callback,
void asyncSendMessage(SendMessageCallback callback,
byte[] body, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit) throws ProxysdkException;

Expand All @@ -126,7 +100,7 @@ public void asyncSendMessage(SendMessageCallback callback,
*
* @param bodyList The data will be sent,which is a collection consisting of byte arrays
*/
public void asyncSendMessage(SendMessageCallback callback,
void asyncSendMessage(SendMessageCallback callback,
List<byte[]> bodyList, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit) throws ProxysdkException;

Expand All @@ -139,7 +113,7 @@ public void asyncSendMessage(SendMessageCallback callback,
* @param extraAttrMap The attributes you want to add, and each
* element of extraAttrMap contains a pair like attrKey,attrValue
*/
public void asyncSendMessage(SendMessageCallback callback,
void asyncSendMessage(SendMessageCallback callback,
List<byte[]> bodyList, String groupId, String streamId, long dt, String msgUUID,
long timeout, TimeUnit timeUnit,
Map<String, String> extraAttrMap) throws ProxysdkException;
Expand Down Expand Up @@ -170,5 +144,5 @@ void asyncSendMessage(String inlongGroupId, String inlongStreamId, byte[] body,
void asyncSendMessage(String inlongGroupId, String inlongStreamId, List<byte[]> bodyList,
SendMessageCallback callback) throws ProxysdkException;

public void close();
void close();
}
Loading
Loading