RockteMQ-AI commented on code in PR #31:
URL: https://github.com/apache/rocketmq-connect/pull/31#discussion_r3839571102
##########
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:
RedisUpdater.push() is a stub that always returns null. RedisSinkTask.put()
calls this method and checks the Boolean return value without null-safety,
which will cause a NullPointerException at `if (!isSuccess)` on every
invocation. The sink task is completely non-functional.
##########
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();
Review Comment:
`Boolean parseError` is declared with boxed `Boolean` (object type) instead
of primitive `boolean`. While initialized to `false`, using the boxed type is
unnecessary here and could theoretically cause a NullPointerException if the
variable were ever left uninitialized in a refactored code path.
##########
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();
+
Review Comment:
`e.printStackTrace()` is called in `start()` before the structured log
statement. This sends the stack trace to stderr outside the logging framework,
making it invisible in log aggregators. Use only `LOGGER.error(...)` with the
exception as the last argument.
##########
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 {
Review Comment:
Unsafe cast: `(String) fieldValue` will throw ClassCastException if the
payload value is not a String. There is no type check or null guard before the
cast and the subsequent JSONObject.parseArray() call.
##########
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) {
Review Comment:
`updater` field is never initialized — it is declared but no assignment
appears in `start()` or anywhere else. Every call to `put()` will throw a
NullPointerException when `updater.push(...)` is invoked.
##########
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
Review Comment:
The `eventProcessor` field and its associated `RedisEventHandler` are
started in `start()`, but the sink task's `put()` method never reads from the
processor — it only writes to Redis via `updater`. Starting a full replication
event processor (which connects to Redis as a replica) in a sink task is
architecturally wrong and will create a redundant Redis replication stream.
##########
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:
The `kvEntryConverter` field is initialized in `start()` but never used
anywhere in the class. The sink path bypasses the converter and directly parses
JSON in `put()`, making the converter dead code and leaving the abstraction
incomplete.
##########
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;
Review Comment:
`taskConfigs()` always returns a single-element list containing the full
config regardless of the requested task count. The framework may call this with
a parallelism hint; the connector ignores it and cannot scale to multiple tasks.
##########
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() {
Review Comment:
`commit()` is a no-op. For at-least-once delivery guarantees the framework
relies on this callback to advance committed offsets. Leaving it empty means
offsets are never acknowledged, which may cause the framework to redeliver all
messages on restart.
##########
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/modified files in this PR have
the ASF license header added, but RedisUpdater.java does not, which will fail
the apache-rat license check added in pom.xml.
--
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]
