You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@eventmesh.apache.org by "yanrongzhen (via GitHub)" <gi...@apache.org> on 2023/11/02 10:02:47 UTC
[PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
yanrongzhen opened a new pull request, #4533:
URL: https://github.com/apache/eventmesh/pull/4533
Fixes #4531 .
### Modifications
Add unit test for RedisSinkConnector.
--
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#issuecomment-1790464841
## [Codecov](https://app.codecov.io/gh/apache/eventmesh/pull/4533?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#4533](https://app.codecov.io/gh/apache/eventmesh/pull/4533?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (3e87347) into [master](https://app.codecov.io/gh/apache/eventmesh/commit/40a2aea812e75895e70e42be99c3736bcfcb10b1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (40a2aea) will **increase** coverage by `0.04%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 3e87347 differs from pull request most recent head 045280d. Consider uploading reports for the commit 045280d to get more accurate results
```diff
@@ Coverage Diff @@
## master #4533 +/- ##
============================================
+ Coverage 16.09% 16.14% +0.04%
- Complexity 1550 1565 +15
============================================
Files 730 739 +9
Lines 28640 28739 +99
Branches 2523 2526 +3
============================================
+ Hits 4611 4639 +28
- Misses 23581 23649 +68
- Partials 448 451 +3
```
[see 17 files with indirect coverage changes](https://app.codecov.io/gh/apache/eventmesh/pull/4533/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1380164882
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ Config config = ConfigUtil.parse(connector.configClass());
+ connector.init(config);
+ connector.start();
+ }
+
+ @Test
+ public void testPutConnectRecords() {
+ final int count = 5;
+ final String message = "testRedisMessage";
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(), (message + i).getBytes());
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
+ records.add(connectRecord);
+ }
+ connector.put(records);
Review Comment:
Actually, since you used redis-mock to simulate Redis, you can try using `Jedis` to connect to `RedisServer` (`new Jedis (redisServer. getHost(), redisServer. getBindPort())`), and then verify the data passed in by the connector here.
---
其实既然你使用了redis-mock来模拟Redis，这里你就可以尝试用`Jedis`来连接`RedisServer`（`new Jedis(redisServer.getHost(), redisServer.getBindPort())`），然后在这校验connector传进去的数据。
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ Config config = ConfigUtil.parse(connector.configClass());
+ connector.init(config);
+ connector.start();
+ }
+
+ @Test
+ public void testPutConnectRecords() {
+ final int count = 5;
+ final String message = "testRedisMessage";
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(), (message + i).getBytes());
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
+ records.add(connectRecord);
+ }
+ connector.put(records);
Review Comment:
Actually, since you used redis-mock to simulate Redis, you can try using `Jedis` to connect to `RedisServer` (`new Jedis (redisServer. getHost(), redisServer. getBindPort())`), and then verify the data passed in by the connector here.
---
其实既然你使用了redis-mock来模拟Redis，这里你就可以尝试用`Jedis`来连接`RedisServer`（`new Jedis(redisServer.getHost(), redisServer.getBindPort())`），然后在这校验connector传进去的数据。
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1381321682
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+import org.redisson.config.Config;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ setupRedisServer(getPortFromUrl(sinkConfig.getConnectorConfig().getServer()));
+ connector.init(sinkConfig);
+ connector.start();
+ Config config = new Config();
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(sinkConfig.getConnectorConfig().getServer());
+ redisson = (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
Review Comment:
Can `ConnectRecord` be directly passed into a `String` object? So that there is no converting `Map` to `String` and converting `String` to `Map`. Simplify the code and achieve the same verification effect.
---
能否直接给`ConnectRecord`传入`String`对象？这样就不需要把`Map`转成`String`和再把`String`转成`Map`的过程，简化代码，且一样达到验证效果。
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+import org.redisson.config.Config;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ setupRedisServer(getPortFromUrl(sinkConfig.getConnectorConfig().getServer()));
+ connector.init(sinkConfig);
+ connector.start();
+ Config config = new Config();
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(sinkConfig.getConnectorConfig().getServer());
+ redisson = (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
Review Comment:
Could you remove these three items that have not been verified anywhere?
---
这三项并没有被验证的地方，能否去掉？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+import org.redisson.config.Config;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ setupRedisServer(getPortFromUrl(sinkConfig.getConnectorConfig().getServer()));
+ connector.init(sinkConfig);
+ connector.start();
+ Config config = new Config();
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(sinkConfig.getConnectorConfig().getServer());
+ redisson = (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
Review Comment:
Could you remove these three items that have not been verified anywhere?
---
这三项并没有被验证的地方，能否去掉？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+import org.redisson.config.Config;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ setupRedisServer(getPortFromUrl(sinkConfig.getConnectorConfig().getServer()));
+ connector.init(sinkConfig);
+ connector.start();
+ Config config = new Config();
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(sinkConfig.getConnectorConfig().getServer());
+ redisson = (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
Review Comment:
Can `ConnectRecord` be directly passed into a `String` object? So that there is no converting `Map` to `String` and converting `String` to `Map`. Simplify the code and achieve the same verification effect.
---
能否直接给`ConnectRecord`传入`String`对象？这样就不需要把`Map`转成`String`和再把`String`转成`Map`的过程，简化代码，且一样达到验证效果。
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1380164882
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ Config config = ConfigUtil.parse(connector.configClass());
+ connector.init(config);
+ connector.start();
+ }
+
+ @Test
+ public void testPutConnectRecords() {
+ final int count = 5;
+ final String message = "testRedisMessage";
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(), (message + i).getBytes());
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
+ records.add(connectRecord);
+ }
+ connector.put(records);
Review Comment:
Actually, since you used redis-mock to simulate Redis, you can try using `Jedis` to connect to redis-mock's `RedisServer` (`new Jedis (redisServer. getHost(), redisServer. getBindPort())`), and then verify the data passed in by the connector here.
---
其实既然你使用了redis-mock来模拟Redis，这里你就可以尝试用`Jedis`来连接redis-mock的`RedisServer`（`new Jedis(redisServer.getHost(), redisServer.getBindPort())`），然后在这校验connector传进去的数据。
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1381172508
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/AbstractRedisServer.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.eventmesh.connector.redis;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import ai.grakn.redismock.RedisServer;
+
+public abstract class AbstractRedisServer {
+
+ protected static RedisServer redisServer;
+
+ @BeforeAll
+ public static void setupRedisServer() throws Exception {
+ redisServer = RedisServer.newRedisServer(6379);
Review Comment:
Is it more reliable to obtain this port from the configuration file?
---
这个端口也从配置文件中获取是不是更可靠？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.RedisProperties;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties properties = configService.buildConfigInstance(RedisProperties.class);
+ redisson = createRedisson(properties);
Review Comment:
Why not directly utilize the Redis configuration in sink-config.yml and add `RedisProperties` and redis-client.properties to increase code complexity?
---
为什么不直接利用sink-config.yml中的获取redis配置，而新增`RedisProperties`、redis-client.properties增加代码的复杂性？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/AbstractRedisServer.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.eventmesh.connector.redis;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import ai.grakn.redismock.RedisServer;
+
+public abstract class AbstractRedisServer {
+
+ protected static RedisServer redisServer;
+
+ @BeforeAll
+ public static void setupRedisServer() throws Exception {
+ redisServer = RedisServer.newRedisServer(6379);
Review Comment:
Is it more reliable to obtain this port from the configuration file?
---
这个端口也从配置文件中获取是不是更可靠？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.RedisProperties;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties properties = configService.buildConfigInstance(RedisProperties.class);
+ redisson = createRedisson(properties);
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ connector.init(sinkConfig);
+ connector.start();
+ }
+
+ private Redisson createRedisson(RedisProperties properties) {
+ String serverAddress = properties.getServerAddress();
+ String serverPassword = properties.getServerPassword();
+
+ org.redisson.config.Config config =
+ OBJECT_MAPPER.convertValue(properties.getRedissonProperties(), org.redisson.config.Config.class);
+
+ if (config == null) {
+ config = new org.redisson.config.Config();
+ }
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(serverAddress)
+ .setPassword(serverPassword);
+
+ return (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
+ records.add(connectRecord);
+ }
+ connector.put(records);
+ Assertions.assertTrue(downLatch.await(10, TimeUnit.SECONDS));
Review Comment:
What is the meaning of the assertion here? Is it OK to directly write write `downLatch.await(10, TimeUnit.SECONDS)`?
---
这里的断言是什么意义？直接写`downLatch.await(10, TimeUnit.SECONDS)`可不可以？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.RedisProperties;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties properties = configService.buildConfigInstance(RedisProperties.class);
+ redisson = createRedisson(properties);
Review Comment:
Why not directly utilize the Redis configuration in sink-config.yml and add `RedisProperties` and redis-client.properties to increase code complexity?
---
为什么不直接利用sink-config.yml中的获取redis配置，而新增`RedisProperties`、redis-client.properties增加代码的复杂性？
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.RedisProperties;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties properties = configService.buildConfigInstance(RedisProperties.class);
+ redisson = createRedisson(properties);
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ connector.init(sinkConfig);
+ connector.start();
+ }
+
+ private Redisson createRedisson(RedisProperties properties) {
+ String serverAddress = properties.getServerAddress();
+ String serverPassword = properties.getServerPassword();
+
+ org.redisson.config.Config config =
+ OBJECT_MAPPER.convertValue(properties.getRedissonProperties(), org.redisson.config.Config.class);
+
+ if (config == null) {
+ config = new org.redisson.config.Config();
+ }
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(serverAddress)
+ .setPassword(serverPassword);
+
+ return (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
+ records.add(connectRecord);
+ }
+ connector.put(records);
+ Assertions.assertTrue(downLatch.await(10, TimeUnit.SECONDS));
Review Comment:
What is the meaning of the assertion here? Is it OK to directly write write `downLatch.await(10, TimeUnit.SECONDS)`?
---
这里的断言是什么意义？直接写`downLatch.await(10, TimeUnit.SECONDS)`可不可以？
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "yanrongzhen (via GitHub)" <gi...@apache.org>.
yanrongzhen commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1381195628
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.RedisProperties;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties properties = configService.buildConfigInstance(RedisProperties.class);
+ redisson = createRedisson(properties);
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ connector.init(sinkConfig);
+ connector.start();
+ }
+
+ private Redisson createRedisson(RedisProperties properties) {
+ String serverAddress = properties.getServerAddress();
+ String serverPassword = properties.getServerPassword();
+
+ org.redisson.config.Config config =
+ OBJECT_MAPPER.convertValue(properties.getRedissonProperties(), org.redisson.config.Config.class);
+
+ if (config == null) {
+ config = new org.redisson.config.Config();
+ }
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(serverAddress)
+ .setPassword(serverPassword);
+
+ return (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
+ records.add(connectRecord);
+ }
+ connector.put(records);
+ Assertions.assertTrue(downLatch.await(10, TimeUnit.SECONDS));
Review Comment:
CountDownLatch.await () will not throw an exception, so we need to verify the returned result.
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "yanrongzhen (via GitHub)" <gi...@apache.org>.
yanrongzhen commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1381421270
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnectorTest.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.eventmesh.connector.redis.sink.connector;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.connector.redis.AbstractRedisServer;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordOffset;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.RecordPartition;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+import org.redisson.config.Config;
+
+import io.cloudevents.CloudEvent;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+public class RedisSinkConnectorTest extends AbstractRedisServer {
+
+ private RedisSinkConnector connector;
+
+ private Redisson redisson;
+
+ private RedisSinkConfig sinkConfig;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ connector = new RedisSinkConnector();
+ sinkConfig = (RedisSinkConfig) ConfigUtil.parse(connector.configClass());
+ setupRedisServer(getPortFromUrl(sinkConfig.getConnectorConfig().getServer()));
+ connector.init(sinkConfig);
+ connector.start();
+ Config config = new Config();
+ config.setCodec(CloudEventCodec.getInstance());
+ config.useSingleServer()
+ .setAddress(sinkConfig.getConnectorConfig().getServer());
+ redisson = (Redisson) Redisson.create(config);
+ }
+
+ @Test
+ public void testPutConnectRecords() throws InterruptedException {
+ RTopic topic = redisson.getTopic(sinkConfig.connectorConfig.getTopic());
+
+ final String expectedMessage = "testRedisMessage";
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+ topic.addListener(CloudEvent.class, (channel, msg) -> {
+ downLatch.countDown();
+ Assertions.assertNotNull(msg.getData());
+ HashMap<String, String> contentMap = JsonUtils.parseTypeReferenceObject(new String(msg.getData().toBytes()),
+ new TypeReference<HashMap<String, String>>() {
+ });
+ Assertions.assertNotNull(contentMap);
+ Assertions.assertEquals(expectedMessage, contentMap.get("content"));
+ });
+
+ List<ConnectRecord> records = new ArrayList<>();
+ for (int i = 0; i < expectedCount; i++) {
+ RecordPartition partition = new RecordPartition();
+ RecordOffset offset = new RecordOffset();
+ Map<String, String> content = new HashMap<>();
+ content.put("content", expectedMessage);
+ ConnectRecord connectRecord = new ConnectRecord(partition, offset, System.currentTimeMillis(),
+ JsonUtils.toJSONString(content).getBytes(StandardCharsets.UTF_8));
+ connectRecord.addExtension("id", String.valueOf(UUID.randomUUID()));
+ connectRecord.addExtension("source", "testSource");
+ connectRecord.addExtension("type", "testType");
Review Comment:
If removed, an error will occured when ConnectRecord is converted to CloudEvent.
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533#discussion_r1381217644
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/AbstractRedisServer.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.eventmesh.connector.redis;
+
+import ai.grakn.redismock.RedisServer;
+
+public abstract class AbstractRedisServer {
+
+ protected RedisServer redisServer;
+
+ public void setupRedisServer(int port) throws Exception {
+ redisServer = RedisServer.newRedisServer(port);
+ redisServer.start();
Review Comment:
After you change the startup of `RedisServer` to `@BeforeEach`, you need to close it in `@AfterEach`.
---
你将`RedisServer`的启动改到`@BeforeEach`中以后，需要在`@AfterEach`中关闭。
##########
eventmesh-connectors/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/AbstractRedisServer.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * 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.eventmesh.connector.redis;
+
+import ai.grakn.redismock.RedisServer;
+
+public abstract class AbstractRedisServer {
+
+ protected RedisServer redisServer;
+
+ public void setupRedisServer(int port) throws Exception {
+ redisServer = RedisServer.newRedisServer(port);
+ redisServer.start();
Review Comment:
After you change the startup of `RedisServer` to `@BeforeEach`, you need to close it in `@AfterEach`.
---
你将`RedisServer`的启动改到`@BeforeEach`中以后，需要在`@AfterEach`中关闭。
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE#4531] Add unit test for RedisSinkConnector. (eventmesh)
Posted by "mxsm (via GitHub)" <gi...@apache.org>.
mxsm merged PR #4533:
URL: https://github.com/apache/eventmesh/pull/4533
--
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org