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

[Issue #655] Adding send message constraints for message size and batch size #760

Merged
merged 4 commits into from
Feb 13, 2022
Merged

[Issue #655] Adding send message constraints for message size and batch size #760

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8ead1ee
[Issue #655] Adding send message constraints for message size and bat…
jinrongluo Feb 9, 2022
0761ae5
[Issue #655] fix style check issue.
jinrongluo Feb 9, 2022
3368ed8
Merge remote-tracking branch 'origin/master' into dev-batchlimit
jinrongluo Feb 9, 2022
868ba83
[Issue #655] Fix eventsize and batch eventsize property names
jinrongluo Feb 11, 2022
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
2 changes: 1 addition & 1 deletion ...sh-examples/src/main/java/org/apache/eventmesh/tcp/demo/pub/cloudevents/AsyncPublish.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ public static void main(String[] args) throws Exception {
EventMeshTCPClientFactory.createEventMeshTCPClient(eventMeshTcpClientConfig, CloudEvent.class);
client.init();

for (int i = 0; i < 5; i++) {
for (int i = 0; i < 2; i++) {
CloudEvent event = EventMeshTestUtils.generateCloudEventV1Async();
logger.info("begin send async msg[{}]==================={}", i, event);
client.publish(event, EventMeshCommon.DEFAULT_TIME_OUT_MILLS);
Expand Down
3 changes: 3 additions & 0 deletions eventmesh-runtime/conf/eventmesh.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,9 @@ eventMesh.server.tcp.msgReqnumPerSecond=15000
eventMesh.server.http.msgReqnumPerSecond=15000
eventMesh.server.session.upstreamBufferSize=20

eventmesh.server.eventSize=1000
eventmesh.server.eventBatchSize=10
jinrongluo marked this conversation as resolved.
Show resolved Hide resolved

# thread number about global scheduler
eventMesh.server.global.scheduler=5
eventMesh.server.tcp.taskHandleExecutorPoolSize=8
Expand Down
18 changes: 18 additions & 0 deletions .../src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshHTTPConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,10 @@ public class EventMeshHTTPConfiguration extends CommonConfiguration {

public int eventMeshBatchMsgRequestNumPerSecond = 20000;

public int eventMeshEventSize = 1000;

public int eventMeshEventBatchSize = 10;

public List<IPAddress> eventMeshIpv4BlackList = Collections.emptyList();

public List<IPAddress> eventMeshIpv6BlackList = Collections.emptyList();
Expand Down Expand Up @@ -266,6 +270,16 @@ public void init() {

}

String eventSize = configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SERVER_EVENTSIZE);
if (StringUtils.isNotEmpty(eventSize) && StringUtils.isNumeric(eventSize)) {
eventMeshEventSize = Integer.parseInt(eventSize);
}

String eventBatchSize = configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SERVER_EVENT_BATCHSIZE);
if (StringUtils.isNotEmpty(eventBatchSize) && StringUtils.isNumeric(eventBatchSize)) {
eventMeshEventBatchSize = Integer.parseInt(eventBatchSize);
}

String ipv4BlackList = configurationWrapper.getProp(ConfKeys.KEY_EVENTMESH_SERVER_IPV4_BLACK_LIST);
if (StringUtils.isNotEmpty(ipv4BlackList)) {
eventMeshIpv4BlackList = getBlacklist(ipv4BlackList);
Expand Down Expand Up @@ -339,6 +353,10 @@ static class ConfKeys {

public static String KEY_EVENTMESH_SERVER_MSG_REQ_NUM_PER_SECOND = "eventMesh.server.http.msgReqnumPerSecond";

public static String KEY_EVENTMESH_SERVER_EVENTSIZE = "eventmesh.server.eventSize";

public static String KEY_EVENTMESH_SERVER_EVENT_BATCHSIZE = "eventmesh.server.eventBatchSize";

public static String KEY_EVENTMESH_SERVER_IPV4_BLACK_LIST = "eventmesh.server.blacklist.ipv4";

public static String KEY_EVENTMESH_SERVER_IPV6_BLACK_LIST = "eventmesh.server.blacklist.ipv6";
Expand Down
10 changes: 10 additions & 0 deletions ...e/src/main/java/org/apache/eventmesh/runtime/configuration/EventMeshTCPConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,10 @@ public class EventMeshTCPConfiguration extends CommonConfiguration {

public int sleepIntervalInRebalanceRedirectMills = 200;

public int eventMeshEventSize = 1000;

public int eventMeshEventBatchSize = 10;

private TrafficShapingConfig gtc = new TrafficShapingConfig(0, 10_000, 1_000, 2000);
private TrafficShapingConfig ctc = new TrafficShapingConfig(0, 2_000, 1_000, 10_000);

Expand Down Expand Up @@ -155,6 +159,10 @@ public void init() {
sleepIntervalInRebalanceRedirectMills = configurationWrapper.getIntProp(
ConfKeys.KEYS_EVENTMESH_SERVER_REBALANCE_REDIRECT_SLEEP_TIME, sleepIntervalInRebalanceRedirectMills);

eventMeshEventSize = configurationWrapper.getIntProp(ConfKeys.KEYS_EVENTMESH_SERVER_EVENTSIZE, eventMeshEventSize);

eventMeshEventBatchSize = configurationWrapper.getIntProp(
ConfKeys.KEYS_EVENTMESH_SERVER_EVENT_BATCHSIZE, eventMeshEventBatchSize);
}

public TrafficShapingConfig getGtc() {
Expand Down Expand Up @@ -191,6 +199,8 @@ static class ConfKeys {
public static String KEYS_EVENTMESH_SERVER_PUSH_FAIL_ISOLATE_TIME = "eventMesh.server.tcp.pushFailIsolateTimeInMills";
public static String KEYS_EVENTMESH_SERVER_GRACEFUL_SHUTDOWN_SLEEP_TIME = "eventMesh.server.gracefulShutdown.sleepIntervalInMills";
public static String KEYS_EVENTMESH_SERVER_REBALANCE_REDIRECT_SLEEP_TIME = "eventMesh.server.rebalanceRedirect.sleepIntervalInM";
public static String KEYS_EVENTMESH_SERVER_EVENTSIZE = "eventmesh.server.eventSize";
public static String KEYS_EVENTMESH_SERVER_EVENT_BATCHSIZE = "eventmesh.server.eventBatchSize";
}

public static class TrafficShapingConfig {
Expand Down
26 changes: 26 additions & 0 deletions .../org/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -111,6 +112,18 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
String producerGroup = "";
int eventSize = eventList.size();

if (eventSize > eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventBatchSize) {
jinrongluo marked this conversation as resolved.
Show resolved Hide resolved
batchMessageLogger.error("Event batch size exceeds the limit: {}",
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventBatchSize);

responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
sendMessageBatchResponseHeader,
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
"Event batch size exceeds the limit: " + eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventBatchSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}

for (CloudEvent event : eventList) {
//validate event
if (StringUtils.isBlank(event.getId())
Expand All @@ -126,6 +139,19 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
return;
}

String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize) {
batchMessageLogger.error("Event size exceeds the limit: {}",
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize);

responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
sendMessageBatchResponseHeader,
SendMessageBatchResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_HEADER_ERR.getRetCode(),
"Event size exceeds the limit: " + eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}

String idc = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.IDC)).toString();
String pid = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.PID)).toString();
String sys = Objects.requireNonNull(event.getExtension(ProtocolKey.ClientInstanceKey.SYS)).toString();
Expand Down
14 changes: 14 additions & 0 deletions ...rg/apache/eventmesh/runtime/core/protocol/http/processor/BatchSendMessageV2Processor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@

import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

Expand Down Expand Up @@ -158,6 +159,19 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
return;
}

String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize) {
batchMessageLogger.error("Event size exceeds the limit: {}",
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize);

responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
sendMessageBatchV2ResponseHeader,
SendMessageBatchV2ResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
"Event size exceeds the limit: " + eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}

//do acl check
if (eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshServerSecurityEnable) {
String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
Expand Down
14 changes: 14 additions & 0 deletions ...java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@

import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

Expand Down Expand Up @@ -152,6 +153,19 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
return;
}

String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize) {
httpLogger.error("Event size exceeds the limit: {}",
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize);

responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
replyMessageResponseHeader,
ReplyMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
"Event size exceeds the limit: " + eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}

EventMeshProducer eventMeshProducer = eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);

if (!eventMeshProducer.getStarted().get()) {
Expand Down
14 changes: 14 additions & 0 deletions .../org/apache/eventmesh/runtime/core/protocol/http/processor/SendAsyncMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@

import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

Expand Down Expand Up @@ -195,6 +196,19 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
event = CloudEventBuilder.from(event).withExtension(SendMessageRequestBody.TTL, ttl).build();
}

String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize) {
httpLogger.error("Event size exceeds the limit: {}",
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize);

responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
"Event size exceeds the limit: " + eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}

try {
// body
//omsMsg.setBody(sendMessageRequestBody.getContent().getBytes(EventMeshConstants.DEFAULT_CHARSET));
Expand Down
14 changes: 14 additions & 0 deletions ...a/org/apache/eventmesh/runtime/core/protocol/http/processor/SendSyncMessageProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@

import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

Expand Down Expand Up @@ -191,6 +192,19 @@ public void processRequest(ChannelHandlerContext ctx, AsyncContext<HttpCommand>
return;
}

String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize) {
httpLogger.error("Event size exceeds the limit: {}",
eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize);

responseEventMeshCommand = asyncContext.getRequest().createHttpCommandResponse(
sendMessageResponseHeader,
SendMessageResponseBody.buildBody(EventMeshRetCode.EVENTMESH_PROTOCOL_BODY_ERR.getRetCode(),
"Event size exceeds the limit: " + eventMeshHTTPServer.getEventMeshHttpConfiguration().eventMeshEventSize));
asyncContext.onComplete(responseEventMeshCommand);
return;
}

EventMeshProducer eventMeshProducer =
eventMeshHTTPServer.getProducerManager().getEventMeshProducer(producerGroup);

Expand Down
7 changes: 7 additions & 0 deletions .../java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageTransferTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@

import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
Expand Down Expand Up @@ -86,6 +87,12 @@ public void run() {
throw new Exception("event is null");
}

String content = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
if (content.length() > eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshEventSize) {
throw new Exception("event size exceeds the limit: "
+ eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshEventSize);
}

//do acl check in sending msg
if (eventMeshTCPServer.getEventMeshTCPConfiguration().eventMeshServerSecurityEnable) {
String remoteAddr = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
Expand Down