This is an automated email from the ASF dual-hosted git repository.
wqliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 140bebd40 [ISSUE #4133]Optimize OpenFunctionSourceConnector poll
new 006a36d6f Merge pull request #4134 from mxsm/eventmesh-4133
140bebd40 is described below
commit 140bebd40363c4318547600891b940bc1e2e02f7
Author: mxsm <[email protected]>
AuthorDate: Sun Jun 18 23:23:34 2023 +0800
[ISSUE #4133]Optimize OpenFunctionSourceConnector poll
---
.../connector/OpenFunctionSourceConnector.java | 21 +++++++++++++++++----
.../apache/eventmesh/openconnect/SourceWorker.java | 12 ++++++++----
2 files changed, 25 insertions(+), 8 deletions(-)
diff --git
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
index 4e79a56d9..c25948575 100644
---
a/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-openfunction/src/main/java/org/apache/eventmesh/connector/openfunction/source/connector/OpenFunctionSourceConnector.java
@@ -26,12 +26,15 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class OpenFunctionSourceConnector implements Source {
+ private static final int DEFAULT_BATCH_SIZE = 10;
+
private OpenFunctionSourceConfig sourceConfig;
private BlockingQueue<ConnectRecord> queue;
@@ -74,10 +77,20 @@ public class OpenFunctionSourceConnector implements Source {
@Override
public List<ConnectRecord> poll() {
- List<ConnectRecord> connectRecords = new ArrayList<>();
- ConnectRecord connectRecord = queue.poll();
- if (connectRecord != null) {
- connectRecords.add(connectRecord);
+
+ List<ConnectRecord> connectRecords = new
ArrayList<>(DEFAULT_BATCH_SIZE);
+
+ for (int count = 0; count < DEFAULT_BATCH_SIZE; ++count) {
+ try {
+ ConnectRecord connectRecord = queue.poll(3, TimeUnit.SECONDS);
+ if (connectRecord == null) {
+ break;
+ }
+ connectRecords.add(connectRecord);
+ } catch (InterruptedException e) {
+ // nothing to do
+ break;
+ }
}
return connectRecords;
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
index 91e494a33..0f0c2f478 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
@@ -29,6 +29,8 @@ import
org.apache.eventmesh.openconnect.api.config.SourceConfig;
import org.apache.eventmesh.openconnect.api.data.ConnectRecord;
import org.apache.eventmesh.openconnect.api.source.Source;
+import org.apache.commons.collections4.CollectionUtils;
+
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.List;
@@ -120,7 +122,7 @@ public class SourceWorker implements ConnectorWorker {
try {
connectRecord = queue.poll(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
- e.printStackTrace();
+ Thread.currentThread().interrupt();
log.error("poll connect record error", e);
}
if (connectRecord == null) {
@@ -136,8 +138,11 @@ public class SourceWorker implements ConnectorWorker {
source.start();
while (isRunning) {
List<ConnectRecord> connectorRecordList = source.poll();
- for (ConnectRecord connectRecord : connectorRecordList) {
- queue.put(connectRecord);
+ if (CollectionUtils.isEmpty(connectorRecordList)) {
+ continue;
+ }
+ for (ConnectRecord record : connectorRecordList) {
+ queue.put(record);
}
}
}
@@ -175,7 +180,6 @@ public class SourceWorker implements ConnectorWorker {
try {
eventMeshTCPClient.close();
} catch (Exception e) {
- e.printStackTrace();
log.error("event mesh client close error", e);
}
log.info("source worker stopped");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
