[WIP][ISSUE #22]rocketmq-connect-redis adapt to the new connect api

connectors/rocketmq-connect-redis/pom.xml

@@ -1,4 +1,20 @@
 <?xml version="1.0" encoding="UTF-8"?>
+<!--
+    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.
+-->
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
@@ -107,6 +123,17 @@
                     <encoding>${project.build.sourceEncoding}</encoding>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.rat</groupId>
+                <artifactId>apache-rat-plugin</artifactId>
+                <version>0.12</version>
+                <configuration>
+                    <excludes>
+                        <exclude>README.md</exclude>
+                        <exclude>README-CN.md</exclude>
+                    </excludes>
+                </configuration>
+            </plugin>
             <plugin>
                 <groupId>org.codehaus.mojo</groupId>
                 <artifactId>findbugs-maven-plugin</artifactId>

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/common/Config.java → connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/config/Config.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.rocketmq.connect.redis.common;
+package org.apache.rocketmq.connect.redis.config;
 
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
@@ -26,6 +26,8 @@
 
 import com.moilioncircle.redis.replicator.RedisURI;
 import org.apache.commons.lang.StringUtils;
+import org.apache.rocketmq.connect.redis.common.RedisConstants;
+import org.apache.rocketmq.connect.redis.common.SyncMod;
 import org.apache.rocketmq.connect.redis.util.PropertyToObjectUtils;
 import io.openmessaging.KeyValue;
 import org.slf4j.Logger;

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkConnector.java

@@ -0,0 +1,82 @@
+/*
+ * 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.connect.redis.connector;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.Task;
+import io.openmessaging.connector.api.sink.SinkConnector;
+import org.apache.rocketmq.connect.redis.config.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * author: doubleDimple
+ */
+public class RedisSinkConnector extends SinkConnector {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RedisSinkConnector.class);
+
+    private volatile boolean configValid = false;
+    private volatile boolean adminStarted;
+    private KeyValue keyValue;
+
+    @Override
+    public String verifyAndSetConfig(KeyValue config) {
+        this.keyValue = config;
+        String msg = Config.checkConfig(keyValue);
+        if (msg != null) {
+            return msg;
+        }
+        this.configValid = true;
+        return null;
+    }
+
+    @Override
+    public void start() {
+        LOGGER.info("the redisSinkConnector is start...");
+    }
+
+    @Override
+    public void stop() {
+
+    }
+
+    @Override
+    public void pause() {
+
+    }
+
+    @Override
+    public void resume() {
+
+    }
+
+    @Override
+    public Class<? extends Task> taskClass() {
+        return RedisSinkTask.class;
+    }
+
+    @Override
+    public List<KeyValue> taskConfigs() {
+        List<KeyValue> keyValues = new ArrayList<>();
+        keyValues.add(this.keyValue);
+        return keyValues;
+    }
+}

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java

@@ -0,0 +1,155 @@
+/*
+ * 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.connect.redis.connector;
+
+import com.alibaba.fastjson.JSONObject;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.common.QueueMetaData;
+import io.openmessaging.connector.api.data.EntryType;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SinkDataEntry;
+import io.openmessaging.connector.api.sink.SinkTask;
+import org.apache.rocketmq.connect.redis.config.Config;
+import org.apache.rocketmq.connect.redis.converter.KVEntryConverter;
+import org.apache.rocketmq.connect.redis.converter.RedisEntryConverter;
+import org.apache.rocketmq.connect.redis.handler.DefaultRedisEventHandler;
+import org.apache.rocketmq.connect.redis.handler.RedisEventHandler;
+import org.apache.rocketmq.connect.redis.processor.DefaultRedisEventProcessor;
+import org.apache.rocketmq.connect.redis.processor.RedisEventProcessor;
+import org.apache.rocketmq.connect.redis.sink.RedisUpdater;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * author doubleDimple
+ */
+public class RedisSinkTask extends SinkTask {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(RedisSinkTask.class);
+
+    private RedisUpdater updater;
+
+    /**
+     * listening and handle Redis event.
+     */
+    private RedisEventProcessor eventProcessor;
+    private Config config;
+    /**
+     * convert kVEntry to list of sourceDataEntry
+     */
+    private KVEntryConverter kvEntryConverter;
+
+    public RedisEventProcessor getEventProcessor() {
+        return eventProcessor;
+    }
+
+    public void setEventProcessor(RedisEventProcessor eventProcessor) {
+        this.eventProcessor = eventProcessor;
+    }
+
+    public Config getConfig() {
+        return config;
+    }
+
+    @Override
+    public void put(Collection<SinkDataEntry> sinkDataEntries) {
+             //save data from MQ to redis
+            for (SinkDataEntry sinkDataEntry : sinkDataEntries) {
+                Map<Field, Object[]> fieldMap = new HashMap<>();
+                Object[] payloads = sinkDataEntry.getPayload();
+
+                Schema schema = sinkDataEntry.getSchema();
+                EntryType entryType = sinkDataEntry.getEntryType();
+
+                List<Field> fields = schema.getFields();
+                Boolean parseError = false;
+                if (!fields.isEmpty()) {
+                    for (Field field : fields) {
+                        Object fieldValue = payloads[field.getIndex()];
+                        Object[] value = JSONObject.parseArray((String)fieldValue).toArray();
+                        if (value.length == 2) {
+                            fieldMap.put(field, value);
+                        } else {
+                            LOGGER.error("parseArray error, fieldValue:{}", fieldValue);
+                            parseError = true;
+                        }
+                    }
+                }
+                if (!parseError) {
+                    Boolean isSuccess = updater.push(fieldMap, entryType);
+                    if (!isSuccess) {
+                        LOGGER.error("push data error, entryType:{}, fieldMap:{}", fieldMap, entryType);
+                    }
+                }
+            }
+    }
+
+    @Override
+    public void commit(Map<QueueMetaData, Long> offsets) {
+
+    }
+
+    @Override
+    public void start(KeyValue keyValue) {
+        this.kvEntryConverter = new RedisEntryConverter();
+
+        this.config = new Config();
+        this.config.load(keyValue);
+        LOGGER.info("task config msg: {}", this.config.toString());
+
+        this.eventProcessor = new DefaultRedisEventProcessor(config);
+        RedisEventHandler eventHandler = new DefaultRedisEventHandler(this.config);
+        this.eventProcessor.registEventHandler(eventHandler);
+        try {
+            this.eventProcessor.start();
+            LOGGER.info("Redis task start.");
+        } catch (IOException e) {
+            e.printStackTrace();
+            LOGGER.error("processor start error: [{}]", e.getMessage());
+            this.stop();
+        }
+    }
+
+    @Override
+    public void stop() {
+        if (this.eventProcessor != null) {
+            try {
+                this.eventProcessor.stop();
+                LOGGER.info("Redis task is stopped.");
+            } catch (IOException e) {
+                LOGGER.error("processor stop error: {}", e);
+            }
+        }
+    }
+
+    @Override
+    public void pause() {
+
+    }
+
+    @Override
+    public void resume() {
+
+    }
+}

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSourceConnector.java

@@ -22,7 +22,7 @@
 import io.openmessaging.KeyValue;
 import io.openmessaging.connector.api.Task;
 import io.openmessaging.connector.api.source.SourceConnector;
-import org.apache.rocketmq.connect.redis.common.Config;
+import org.apache.rocketmq.connect.redis.config.Config;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSourceTask.java

@@ -25,7 +25,7 @@
 import io.openmessaging.KeyValue;
 import io.openmessaging.connector.api.data.SourceDataEntry;
 import io.openmessaging.connector.api.source.SourceTask;
-import org.apache.rocketmq.connect.redis.common.Config;
+import org.apache.rocketmq.connect.redis.config.Config;
 import org.apache.rocketmq.connect.redis.common.Options;
 import org.apache.rocketmq.connect.redis.converter.KVEntryConverter;
 import org.apache.rocketmq.connect.redis.converter.RedisEntryConverter;

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/handler/DefaultRedisEventHandler.java

@@ -23,7 +23,7 @@
 import com.moilioncircle.redis.replicator.event.Event;
 import com.moilioncircle.redis.replicator.rdb.datatype.KeyValuePair;
 import com.moilioncircle.redis.replicator.rdb.iterable.datatype.BatchedKeyValuePair;
-import org.apache.rocketmq.connect.redis.common.Config;
+import org.apache.rocketmq.connect.redis.config.Config;
 import org.apache.rocketmq.connect.redis.common.SyncMod;
 import org.apache.rocketmq.connect.redis.parser.DefaultRedisRdbParser;
 import org.apache.rocketmq.connect.redis.parser.RedisRdbParser;

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/processor/DefaultRedisEventProcessor.java

@@ -40,7 +40,7 @@
 import com.moilioncircle.redis.replicator.Replicator;
 import com.moilioncircle.redis.replicator.event.EventListener;
 import org.apache.commons.lang.StringUtils;
-import org.apache.rocketmq.connect.redis.common.Config;
+import org.apache.rocketmq.connect.redis.config.Config;
 import org.apache.rocketmq.connect.redis.common.RedisConstants;
 import org.apache.rocketmq.connect.redis.common.SyncMod;
 import org.apache.rocketmq.connect.redis.handler.RedisEventHandler;

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/processor/RedisEventListener.java

@@ -25,7 +25,7 @@
 import com.moilioncircle.redis.replicator.event.PreRdbSyncEvent;
 import com.moilioncircle.redis.replicator.rdb.datatype.AuxField;
 import java.io.IOException;
-import org.apache.rocketmq.connect.redis.common.Config;
+import org.apache.rocketmq.connect.redis.config.Config;
 import org.apache.rocketmq.connect.redis.pojo.RedisEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/sink/RedisUpdater.java

@@ -0,0 +1,15 @@
+package org.apache.rocketmq.connect.redis.sink;
+
+import io.openmessaging.connector.api.data.EntryType;
+import io.openmessaging.connector.api.data.Field;
+
+import java.util.Map;
+
+public class RedisUpdater {
+
+
+    public Boolean push(Map<Field, Object[]> fieldMap, EntryType entryType) {
+
+        return null;
+    }
+}

connectors/rocketmq-connect-redis/src/test/java/org/apache/rocketmq/redis/test/common/ConfigTest.java

@@ -1,3 +1,20 @@
+/*
+ * 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.redis.test.common;
 
 import java.nio.ByteBuffer;
@@ -6,7 +23,7 @@
 import com.moilioncircle.redis.replicator.RedisURI;
 import io.openmessaging.KeyValue;
 import io.openmessaging.internal.DefaultKeyValue;
-import org.apache.rocketmq.connect.redis.common.Config;
+import org.apache.rocketmq.connect.redis.config.Config;
 import org.apache.rocketmq.connect.redis.common.SyncMod;
 import org.junit.Assert;
 import org.junit.Test;