[runtime && dist]Configuration and packaging modifications

adapter/runtime/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/boot/common/CirculatorContext.java

@@ -241,7 +241,9 @@ private void refreshRunnerContext(TargetRunnerConfig targetRunnerConfig, Refresh
                 int endIndex = targetRunnerConfig.getComponents().size() - 1;
                 TargetKeyValue targetKeyValue = new TargetKeyValue(targetRunnerConfig.getComponents().get(endIndex));
                 SinkTask sinkTask = initTargetSinkTask(targetKeyValue);
-                pusherTaskMap.put(runnerName, sinkTask);
+                if(sinkTask != null) {
+                    pusherTaskMap.put(runnerName, sinkTask);
+                }
 
                 if (!pusherExecutorMap.containsKey(runnerName)) {
                     pusherExecutorMap.put(runnerName, initDefaultThreadPoolExecutor(runnerName));

adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/ConfigLoader.java

@@ -0,0 +1,81 @@
+/*
+ * 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.rocketmq.eventbridge.adapter.storage.rocketmq.runtimer;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.io.support.PropertiesLoaderUtils;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.util.Properties;
+
+/**
+ * support loading from runtime.properties and spring properties
+ */
+@Component
+@Slf4j
+public class ConfigLoader {
+
+    @Autowired
+    private PropertiesResolveService propertiesResolveService;
+
+    private Properties properties;
+
+    public ConfigLoader() {
+        try {
+            this.properties = PropertiesLoaderUtils.loadAllProperties("runtime.properties");
+        } catch (IOException e) {
+            log.warn("init runtime properties failed ", e);
+        }
+    }
+
+    public String getString(String name, String defaultValue){
+        String result = getValue(name);
+        if (StringUtils.isBlank(result)) {
+            result = defaultValue;
+        }
+        log.debug("Load property '" + name + "' = " + result);
+        return result;
+    }
+
+    public String getString(String name){
+        String result = getValue(name);
+        log.debug("Load property '" + name + "' = " + result);
+        return result;
+    }
+
+    private String getValue(String name){
+        if (StringUtils.isBlank(name)) {
+            return null;
+        }
+        String value = null;
+        try {
+            if (propertiesResolveService != null) {
+                value = propertiesResolveService.getPropertiesValue(name);
+            }
+        } catch (IllegalArgumentException e) {
+            log.warn("Failed load property '" + name + "' from spring.");
+        }
+        if (properties != null && StringUtils.isBlank((value))) {
+            value = properties.getProperty(name);
+        }
+        return value;
+    }
+}

adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/PropertiesResolveService.java

@@ -0,0 +1,50 @@
+/*
+ * 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.rocketmq.eventbridge.adapter.storage.rocketmq.runtimer;
+
+import org.apache.rocketmq.eventbridge.config.CommonConstants;
+import org.springframework.context.EmbeddedValueResolverAware;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.PriorityOrdered;
+import org.springframework.util.StringValueResolver;
+
+/**
+ * spring props load service
+ */
+@Configuration
+public class PropertiesResolveService implements EmbeddedValueResolverAware, PriorityOrdered {
+
+    private StringValueResolver stringValueResolver;
+
+    @Override
+    public void setEmbeddedValueResolver(StringValueResolver resolver) {
+        stringValueResolver = resolver;
+    }
+
+    public String getPropertiesValue(String name) {
+        if (!name.contains(CommonConstants.V_PREFIXE)) {
+            name = CommonConstants.V_PREFIXE + name + CommonConstants.V_POSTFIX;
+        }
+        return stringValueResolver.resolveStringValue(name);
+    }
+
+    @Override
+    public int getOrder() {
+        return PriorityOrdered.HIGHEST_PRECEDENCE;
+    }
+}

adapter/storage/src/main/java/org/apache/rocketmq/eventbridge/adapter/storage/rocketmq/runtimer/RocketMQEventSubscriber.java

@@ -32,7 +32,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Properties;
 import java.util.Set;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
@@ -58,7 +57,6 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.DependsOn;
-import org.springframework.core.io.support.PropertiesLoaderUtils;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
@@ -86,6 +84,9 @@ public class RocketMQEventSubscriber extends EventSubscriber {
     @Autowired
     private TargetRunnerConfigObserver runnerConfigObserver;
 
+    @Autowired
+    private ConfigLoader configLoader;
+
     private final BlockingQueue<MessageExt> messageBuffer = new LinkedBlockingQueue<>(50000);
 
     private Integer pullTimeOut;
@@ -180,17 +181,16 @@ public void close() {
     private void initMqProperties() {
         try {
             ClientConfig clientConfig = new ClientConfig();
-            Properties properties = PropertiesLoaderUtils.loadAllProperties("runtime.properties");
-            String namesrvAddr = properties.getProperty("rocketmq.namesrvAddr");
-            pullTimeOut = Integer.valueOf(properties.getProperty("rocketmq.consumer.pullTimeOut"));
-            pullBatchSize = Integer.valueOf(properties.getProperty("rocketmq.consumer.pullBatchSize"));
-            String accessChannel = properties.getProperty("rocketmq.accessChannel");
-            String namespace = properties.getProperty("rocketmq.namespace");
-            String accessKey = properties.getProperty("rocketmq.consumer.accessKey");
-            String secretKey = properties.getProperty("rocketmq.consumer.secretKey");
-            String socks5UserName = properties.getProperty("rocketmq.consumer.socks5UserName");
-            String socks5Password = properties.getProperty("rocketmq.consumer.socks5Password");
-            String socks5Endpoint = properties.getProperty("rocketmq.consumer.socks5Endpoint");
+            String namesrvAddr = configLoader.getString("rocketmq.namesrvAddr");
+            pullTimeOut = Integer.valueOf(configLoader.getString("rocketmq.consumer.pullTimeOut"));
+            pullBatchSize = Integer.valueOf(configLoader.getString("rocketmq.consumer.pullBatchSize"));
+            String accessChannel = configLoader.getString("rocketmq.accessChannel");
+            String namespace = configLoader.getString("rocketmq.namespace");
+            String accessKey = configLoader.getString("rocketmq.consumer.accessKey");
+            String secretKey = configLoader.getString("rocketmq.consumer.secretKey");
+            String socks5UserName = configLoader.getString("rocketmq.consumer.socks5UserName");
+            String socks5Password = configLoader.getString("rocketmq.consumer.socks5Password");
+            String socks5Endpoint = configLoader.getString("rocketmq.consumer.socks5Endpoint");
 
             clientConfig.setNameSrvAddr(namesrvAddr);
             clientConfig.setAccessChannel(AccessChannel.CLOUD.name().equals(accessChannel) ?

common/src/main/java/org/apache/rocketmq/eventbridge/config/CommonConstants.java

@@ -0,0 +1,23 @@
+/*
+ * 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.rocketmq.eventbridge.config;
+
+public interface CommonConstants {
+    String V_PREFIXE = "${";
+
+    String V_POSTFIX = "}";
+}

dist/src/main/assembly/assembly.xml

@@ -50,5 +50,10 @@
             <outputDirectory>config/</outputDirectory>
             <fileMode>0755</fileMode>
         </fileSet>
+        <fileSet>
+            <directory>src/main/plugin</directory>
+            <outputDirectory>plugin/</outputDirectory>
+            <fileMode>0755</fileMode>
+        </fileSet>
     </fileSets>
 </assembly>

docker/Dockerfile

@@ -0,0 +1,51 @@
+#
+# 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.
+#
+
+FROM eclipse-temurin:8-jdk-centos7 AS builder
+
+RUN set -eux \
+    && yum -y update \
+    && yum -y install curl gnupg unzip \
+    && yum clean all -y
+
+ENV EVENTBRIDGE_USER=apps \
+    EVENTBRIDGE_HOME=/home/apps/rocketmq-eventbridge \
+    EVENTBRIDGE_LOG_DIR=/home/apps/logs \
+    EVENTBRIDGE_LOG_FILE=/home/apps/logs/rocketmq-eventbridge.log
+
+RUN set -x \
+    && adduser "$EVENTBRIDGE_USER" \
+    && mkdir -p "$EVENTBRIDGE_LOG_DIR" \
+    && touch "$EVENTBRIDGE_LOG_FILE"
+
+WORKDIR /home/apps
+ARG version
+# Rocketmq eventbridge version
+ENV ROCKETMQ_EVENTBRIDGE_VERSION ${version}
+
+ADD rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip /home/apps
+RUN cd /home/apps && unzip rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip && rm -rf rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip
+
+RUN chown -R "$EVENTBRIDGE_USER:$EVENTBRIDGE_USER" "$EVENTBRIDGE_HOME"  "$EVENTBRIDGE_LOG_DIR"
+COPY docker-entrypoint.sh /home/apps/rocketmq-eventbridge/bin
+RUN chmod 755 /home/apps/rocketmq-eventbridge/bin/docker-entrypoint.sh
+
+EXPOSE 8083 7001 7002
+USER apps
+WORKDIR /home/apps/rocketmq-eventbridge
+
+ENTRYPOINT ["./bin/docker-entrypoint.sh"]

docker/build-image.sh

@@ -0,0 +1,55 @@
+#!/usr/bin/env bash
+
+# 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.
+
+checkVersion() {
+    echo "Version = $1"
+	  echo $1 |grep -E "^[0-9]+\.[0-9]+\.[0-9]+" > /dev/null
+    if [ $? = 0 ]; then
+        return 1
+    fi
+
+	echo "Version $1 illegal, it should be X.X.X format(e.g. 4.5.0), please check released versions in 'https://archive.apache.org/dist/rocketmq/'"
+    exit -1
+}
+
+checkZipFile() {
+
+	  file="../dist/target/rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip"
+
+    if [ -e "$file" ]; then
+        return 1
+    fi
+
+	  echo "the installation package rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip does not exist in dist/target, please check released package"
+    exit -1
+}
+
+if [ $# -lt 1 ]; then
+    echo -e "Usage: sh $0 Version"
+    exit -1
+fi
+ROCKETMQ_EVENTBRIDGE_VERSION=$1
+checkVersion $ROCKETMQ_EVENTBRIDGE_VERSION
+checkZipFile
+# copy package to docker directory
+cp ../dist/target/rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip ./
+
+# Build rocketmq-eventbridge
+
+docker build --no-cache -f Dockerfile -t apache/rocketmq-eventbridge:${ROCKETMQ_EVENTBRIDGE_VERSION} --build-arg version=${ROCKETMQ_EVENTBRIDGE_VERSION} .
+rm -rf rocketmq-eventbridge-dist-${ROCKETMQ_EVENTBRIDGE_VERSION}.zip
+

docker/docker-entrypoint.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+
+# 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.
+
+set -e
+
+START_EVENTBRIDGE="./bin/eventbridge.sh start"
+echo "${START_EVENTBRIDGE}" &&  source ${START_EVENTBRIDGE}
+
+START_CMD="tail -f /home/apps/logs/rocketmq-eventbridge.log"
+#START_TEST="tail -f /etc/hosts"
+#echo "${START_TEST}" &&  exec ${START_TEST}
+echo "${START_CMD}" &&  exec ${START_CMD}

start/src/main/resources/application.properties

@@ -32,7 +32,7 @@ rocketmq.cluster.name=DefaultCluster
 runtime.config.mode=DB
 runtime.storage.mode=ROCKETMQ
 rumtime.name=eventbridge-runtimer
-runtime.pluginpath=~/eventbridge/plugin
+runtime.pluginpath=./plugin
 
 
 ## log