RockteMQ-AI commented on code in PR #31: URL: https://github.com/apache/rocketmq-connect/pull/31#discussion_r3909720458
########## 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 { + Review Comment: `updater` is never initialized — `updater.push(...)` on line 93 will always throw a NullPointerException. The `RedisUpdater` is a stub that returns `null` anyway, making the entire sink path non-functional. ########## 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) { + Review Comment: `push()` always returns `null` (a stub). This makes the entire sink data path a no-op. The PR is marked WIP but this class should at minimum throw `UnsupportedOperationException` or be clearly documented as unimplemented to avoid silent data loss. ########## 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; + } Review Comment: When `updater.push()` fails or `parseError` is true, the record is silently dropped after logging. There is no dead-letter routing, retry, or error-reporting mechanism, which risks silent data loss in production. ########## 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()]; Review Comment: `JSONObject.parseArray((String)fieldValue)` will throw an unchecked exception if `fieldValue` is `null` or not valid JSON. No null-check or try-catch protects this call, so a single malformed record will crash the entire `put()` batch. ########## 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 Review Comment: The sink task's `start()` creates a `DefaultRedisEventProcessor` and `DefaultRedisEventHandler` — these are source-side replication components (they listen to Redis replication stream). This is architecturally wrong for a sink task that should be writing data into Redis. ########## 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; Review Comment: `taskConfigs()` always returns exactly one config. For multi-partition or scaled deployments, this prevents parallelism. The connector should accept a `maxTasks` parameter and distribute configs accordingly. ########## 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); Review Comment: `e.printStackTrace()` is used instead of logging through SLF4J. This bypasses the configured logging infrastructure and is inappropriate for production connector code. ########## 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 Review Comment: `configValid` is set to `true` in `verifyAndSetConfig` but never read anywhere. `adminStarted` is declared but never used. These are dead fields that add confusion. ########## 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; Review Comment: Missing Apache License header. All other new files in this PR include the standard ASF license block. This will also cause `apache-rat:check` to fail. ########## 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(); Review Comment: `kvEntryConverter` is initialized in `start()` but never used in `put()` or anywhere else in this class. It appears to be leftover dead code. ########## 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..."); + } Review Comment: `start()` only logs a message and performs no initialization (no connection test, no admin client setup). A connector `start()` should validate connectivity or at least initialize shared resources. ########## 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); + } Review Comment: `commit()` is empty — there is no offset tracking or acknowledgment logic. If the framework relies on the sink task to confirm processed offsets, this could lead to duplicate processing on restart. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
