You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by yu...@apache.org on 2022/06/02 08:58:26 UTC
[rocketmq-connect] 09/10: fix "null" check
This is an automated email from the ASF dual-hosted git repository.
yukon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-connect.git
commit 650b169db61afa4c5629b97b30098d51be23f646
Author: zh378814 <wb...@alibaba-inc.com>
AuthorDate: Thu Jun 2 11:23:27 2022 +0800
fix "null" check
---
.../aliyun/rocketmq-connect-eventbridge/README.md | 5 +++--
.../eventbridge/sink/EventBridgeSinkConnector.java | 7 +++++-
.../eventbridge/sink/EventBridgeSinkTask.java | 11 ++++++----
.../sink/constant/EventBridgeConstant.java | 4 ++--
.../connect/eventbridge/sink/utils/CheckUtils.java | 25 ++++++++++++++++++++++
5 files changed, 43 insertions(+), 9 deletions(-)
diff --git a/connectors/aliyun/rocketmq-connect-eventbridge/README.md b/connectors/aliyun/rocketmq-connect-eventbridge/README.md
index 6aec62f..4d3b966 100644
--- a/connectors/aliyun/rocketmq-connect-eventbridge/README.md
+++ b/connectors/aliyun/rocketmq-connect-eventbridge/README.md
@@ -16,14 +16,14 @@ mvn clean install -Dmaven.test.skip=true
```
http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-eventbridge-sink-connector-name}
?config={"source-rocketmq":"${runtime-ip}:${runtime-port}","source-cluster":"${broker-cluster}","connector-class":"org.apache.rocketmq.connect.eventbridge.sink.EventBridgeSinkConnector",“connect-topicname”:"${connect-topicname}"
-,"stsEndpoint”:"${stsEndpoint}",accessKeyId”:"${accessKeyId}",accessKeySecret”:"${accessKeySecret}",roleArn”:"${roleArn}", "roleSessionName":"${roleSessionName}", "eventTime":"${eventTime}","eventSubject":"${eventSubject}","eventType":"${eventType}","aliyuneventbusname":"${aliyuneventbusname}"}
+,"stsEndpoint”:"${stsEndpoint}",accessKeyId”:"${accessKeyId}",accessKeySecret”:"${accessKeySecret}",roleArn”:"${roleArn}", "roleSessionName":"${roleSessionName}", "eventTime":"${eventTime}","eventSubject":"${eventSubject}","eventType":"${eventType}","eventSource":"${eventSource}","aliyuneventbusname":"${aliyuneventbusname}"}
```
例子
```
http://localhost:8081/connectors/eventbridgeConnectorSink?config={"source-rocketmq":"localhost:9876","source-cluster":"DefaultCluster",
"connector-class":"org.apache.rocketmq.connect.eventbridge.sink.EventBridgeSinkConnector",“connect-topicname”:"eventbridge-topic","stsEndpoint”:"xxxx",accessKeyId”:"xxxx",accessKeySecret”:"xxxx",
-roleArn”:"xxxx", "roleSessionName":"xxxx", "eventTime":"xxxx","eventSubject":"", "eventType":"","aliyuneventbusname":"xxxx"}
+roleArn”:"xxxx", "roleSessionName":"xxxx", "eventTime":"xxxx","eventSubject":"", "eventType":"","eventSource":"","aliyuneventbusname":"xxxx"}
```
>**注：** `rocketmq-eventbridge-connect` 的启动依赖于`rocketmq-connect-runtime`项目的启动，需将打好的所有`jar`包放置到`runtime`项目中`pluginPaths`配置的路径后再执行上面的启动请求,该值配置在`runtime`项目下的`connect.conf`文件中
@@ -48,6 +48,7 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-eventbridge-connector
|eventTime | String | NO | 事件产生的时间 | xxxx |
|eventSubject | String | YES | 事件主题 | xxxx |
|eventType | String | YES | 事件类型 | xxxx |
+|eventSource | String | YES | 事件源 | xxxx |
|aliyuneventbusname | String | YES | 接收事件的事件总线名称 | xxxx |
|connect-topicname | String | YES | sink需要处理数据消息topic | xxxx |
diff --git a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkConnector.java b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkConnector.java
index 254567a..0f4af2f 100644
--- a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkConnector.java
+++ b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkConnector.java
@@ -23,6 +23,8 @@ public class EventBridgeSinkConnector extends SinkConnector {
private String eventTime;
+ private String eventSource;
+
private String eventSubject;
private String eventType;
@@ -57,6 +59,7 @@ public class EventBridgeSinkConnector extends SinkConnector {
keyValue.put(EventBridgeConstant.ACCOUNT_ENDPOINT, accountEndpoint);
keyValue.put(EventBridgeConstant.STS_ENDPOINT, stsEndpoint);
keyValue.put(EventBridgeConstant.EVENT_TYPE, eventType);
+ keyValue.put(EventBridgeConstant.EVENT_SOURCE, eventSource);
keyValueList.add(keyValue);
return keyValueList;
}
@@ -73,7 +76,8 @@ public class EventBridgeSinkConnector extends SinkConnector {
|| StringUtils.isBlank(config.getString(EventBridgeConstant.ACCOUNT_ENDPOINT))
|| StringUtils.isBlank(config.getString(EventBridgeConstant.ALIYUN_EVENT_BUS_NAME))
|| StringUtils.isBlank(config.getString(EventBridgeConstant.EVENT_SUBJECT))
- || StringUtils.isBlank(config.getString(EventBridgeConstant.EVENT_TYPE))) {
+ || StringUtils.isBlank(config.getString(EventBridgeConstant.EVENT_TYPE))
+ || StringUtils.isBlank(config.getString(EventBridgeConstant.EVENT_SOURCE))) {
throw new RuntimeException("EventBridge required parameter is null !");
}
}
@@ -90,6 +94,7 @@ public class EventBridgeSinkConnector extends SinkConnector {
accountEndpoint = config.getString(EventBridgeConstant.ACCOUNT_ENDPOINT);
stsEndpoint = config.getString(EventBridgeConstant.STS_ENDPOINT);
eventType = config.getString(EventBridgeConstant.EVENT_TYPE);
+ eventSource = config.getString(EventBridgeConstant.EVENT_SOURCE);
}
@Override
diff --git a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkTask.java b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkTask.java
index 4a4fc74..95880b2 100644
--- a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkTask.java
+++ b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/EventBridgeSinkTask.java
@@ -16,8 +16,8 @@ import io.openmessaging.connector.api.component.task.sink.SinkTask;
import io.openmessaging.connector.api.component.task.sink.SinkTaskContext;
import io.openmessaging.connector.api.data.ConnectRecord;
import io.openmessaging.connector.api.errors.ConnectException;
-import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.connect.eventbridge.sink.constant.EventBridgeConstant;
+import org.apache.rocketmq.connect.eventbridge.sink.utils.CheckUtils;
import org.apache.rocketmq.connect.eventbridge.sink.utils.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,6 +45,8 @@ public class EventBridgeSinkTask extends SinkTask {
private String eventType;
+ private String eventSource;
+
private String aliyuneventbusname;
private String accountEndpoint;
@@ -57,8 +59,8 @@ public class EventBridgeSinkTask extends SinkTask {
try {
sinkRecords.forEach(connectRecord -> cloudEventList.add(EventBuilder.builder()
.withId(connectRecord.getExtension(EventBridgeConstant.EVENT_ID))
- .withSource(URI.create(connectRecord.getExtension(EventBridgeConstant.EVENT_SOURCE)))
- .withType(StringUtils.isBlank(eventType) ? connectRecord.getExtension(EventBridgeConstant.EVENT_TYPE) : eventType)
+ .withSource(CheckUtils.checkNull(eventSource) ? URI.create(connectRecord.getExtension(EventBridgeConstant.EVENT_SOURCE)) : URI.create(eventSource))
+ .withType(CheckUtils.checkNull(eventType) ? connectRecord.getExtension(EventBridgeConstant.EVENT_TYPE) : eventType)
.withSubject(eventSubject)
.withTime(DateUtils.getDate(eventTime, DateUtils.DEFAULT_DATE_FORMAT))
.withJsonStringData(connectRecord.getData().toString())
@@ -99,6 +101,7 @@ public class EventBridgeSinkTask extends SinkTask {
accountEndpoint = config.getString(EventBridgeConstant.ACCOUNT_ENDPOINT);
stsEndpoint = config.getString(EventBridgeConstant.STS_ENDPOINT);
eventType = config.getString(EventBridgeConstant.EVENT_TYPE);
+ eventSource = config.getString(EventBridgeConstant.EVENT_SOURCE);
}
@Override
@@ -106,7 +109,7 @@ public class EventBridgeSinkTask extends SinkTask {
super.start(sinkTaskContext);
try {
Config authConfig = new Config();
- if (StringUtils.isNotBlank(roleArn) && StringUtils.isNotBlank(roleSessionName)) {
+ if (CheckUtils.checkNotNull(roleArn) && CheckUtils.checkNotNull(roleSessionName)) {
DefaultProfile.addEndpoint("", "", "Sts", stsEndpoint);
DefaultProfile profile = DefaultProfile.getProfile("", accessKeyId, accessKeySecret);
IAcsClient client = new DefaultAcsClient(profile);
diff --git a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/constant/EventBridgeConstant.java b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/constant/EventBridgeConstant.java
index 8c96842..2d65eb9 100644
--- a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/constant/EventBridgeConstant.java
+++ b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/constant/EventBridgeConstant.java
@@ -8,9 +8,9 @@ public class EventBridgeConstant {
public static final String ROLE_ARN = "roleArn";
public static final String ROLE_SESSION_NAME = "roleSessionName";
public static final String EVENT_ID = "id";
- public static final String EVENT_SOURCE = "source";
+ public static final String EVENT_SOURCE = "eventSource";
public static final String EVENT_TIME = "eventTime";
- public static final String EVENT_TYPE = "type";
+ public static final String EVENT_TYPE = "eventType";
public static final String EVENT_SUBJECT = "eventSubject";
public static final String ALIYUN_EVENT_BUS_NAME = "aliyuneventbusname";
public static final String STS_ENDPOINT = "stsEndpoint";
diff --git a/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/utils/CheckUtils.java b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/utils/CheckUtils.java
new file mode 100644
index 0000000..067cca4
--- /dev/null
+++ b/connectors/aliyun/rocketmq-connect-eventbridge/src/main/java/org/apache/rocketmq/connect/eventbridge/sink/utils/CheckUtils.java
@@ -0,0 +1,25 @@
+package org.apache.rocketmq.connect.eventbridge.sink.utils;
+
+import org.apache.commons.lang3.StringUtils;
+
+public class CheckUtils {
+ private static final String NULL_CONSTANT = "null";
+
+ public static Boolean checkNull(String check) {
+ if (StringUtils.isBlank(check)) {
+ return Boolean.TRUE;
+ }
+ if (StringUtils.isNotBlank(check) && NULL_CONSTANT.equalsIgnoreCase(check)) {
+ return Boolean.TRUE;
+ }
+ return Boolean.FALSE;
+ }
+
+ public static Boolean checkNotNull(String check) {
+ if (StringUtils.isNotBlank(check) && !NULL_CONSTANT.equalsIgnoreCase(check)) {
+ return Boolean.TRUE;
+ }
+ return Boolean.FALSE;
+ }
+
+}