Adding Redis to Cap by MahmoudSamir101 · Pull Request #817 · dotnetcore/CAP

MahmoudSamir101 · 2021-03-31T21:57:58Z

Hello their,

this is my PR for adding redis to Cap project, the PR provides pub/sub using redis channels, and later on, we can add redis streams support.

the PR has a sample for testing, you may need docker to run redis image, as there is a post build command that will run redis image container

let me know if it helps, your observations most welcome.

Thanks.

yang-xiaodong · 2021-04-06T10:30:49Z

Hello , Thanks for your PR.

Do as you said, I noticed that Redis 5.0 streams support acknowledgment of tasks as they are completed by a group. It seems to be the feature we need, so we need to discuss whether to complete this work using streams now.

Useful link: https://redis.io/topics/streams-intro

What do you think? @xiangxiren

xiangxiren · 2021-04-07T01:15:02Z

Hi,
As far as I know, Stream looks great and we should use it.

yang-xiaodong · 2021-04-07T09:28:54Z

Hello, @MahmoudSamir101

What do you think about using Redis Stream to implement publish and subscribe？

MahmoudSamir101 · 2021-04-07T19:37:00Z

@yang-xiaodong, yes redis streams are a good feature to be added to CAP, we can plan to implement this feature with this PR or another one.

As I see, streams and channels can be used interchangeably based on the use case, many cases can benefit from the pub/sub nature of channels without the need to append these messages indefinitely to the storage or even go forward /backward with messages like streams.

Publishing messages to redis as a distributed cache through CAP is a use case that can benefit from channels, these messages are used to sync the state of cached data with clients who are concern about updating their cache in a pub/sub manner without the need to persist the messages on the storage.

Another use case can be, using redis as a backplane for signalr notifications in a web farm deployment, typically a web application that propagates real-time notifications through signalr connections, these notifications should be sent after business transactions committed.

the notifications will be published through CAP to redis channels, CAP consumers (multiple deployed instances of the web application) consume these messages and broadcasting them through signalr HubLifetimeManager to connected clients.

Anyway, if you have concerns about the nature of channels' work and they will not fit with the CAP project, let's go for streams implementation with CAP.

yang-xiaodong · 2021-04-08T01:55:27Z

Hi @MahmoudSamir101 ,

Let me explain the goal of CAP

CAP is a solution to deal with transaction consistency in distributed scenarios. We need to try our best to ensure the reliability of messages in most unreliable scenarios, so we adopt the outbox pattern and use local transactions on the producer side to achieve strong consistency. This allows us to first ensure that the message on the producer side will not be lost due to failure to reach the MQ or exceptions in the delivery process.

On the consumer side, we need to use the Ack feature provided by the message queue to ensure that the message is stored in the inbox, so as to achieve "at least once". At the same time, we provide features similar to Kafka consumer groups to implement load balancing or broadcasting of messages on the consumer side.

Prior to this, we did not implement Redis Transport because Redis did not have the features we needed and could not achieve our goals. Now when I learned about the Streams feature, I found that it provides the functionality we need and is able to do this and I investigated other EDA architectures, for example, Dapr also use Redis Streams to implement publish and subscribe.

Therefore, I think Streams is more suitable for CAP.

At the same time I want to invite you to join us, do you have interest to join us as a maintainer?

MahmoudSamir101 · 2021-04-08T06:30:41Z

@yang-xiaodong, I got your point, and I'm interested to have a look at implementing Redis streams with CAP.

I would like to join the CAP community for contribution and maintenance for sure.

yang-xiaodong · 2021-04-19T01:56:29Z

Hello @MahmoudSamir101, Any updates about this PR?

MahmoudSamir101 · 2021-04-19T18:26:09Z

@yang-xiaodong , yes I'm planning to push my updates soon, may be next Thursday or before that.

MahmoudSamir101 · 2021-04-24T12:01:48Z

@yang-xiaodong could you please check the updates.

Some points to be considered:

StackExchange.Redis doesn't support blocking commands which are used to enable redis to push new deliverable messages to consumer groups, we do polling as a fallback strategy.
Pending messages are messages delivered to a consumer group without ack back from the consumer, we pull pending messages only when the CAP server starts, when new messages delivered to a consumer and it fails to ack them, redis marked them as a pending message and we pull them again with the next start of CAP server, it was designed like that to limit the polling of messages while the CAP server is running for new messages only.

Let me know your thoughts about that.

xiangxiren

Why is there kafka here?

yang-xiaodong · 2021-04-26T01:30:29Z

src/DotNetCore.CAP.Redis/CapOptions.Redis.Extensions.cs

+using DotNetCore.CAP;
+using DotNetCore.CAP.Internal;
+using StackExchange.Redis;
+using System;


According to our specifications, the System namespace needs to be placed at the first

yang-xiaodong · 2021-04-26T01:32:42Z

src/DotNetCore.CAP.Redis/CapOptions.Redis.Extensions.cs

+        public static CapOptions UseRedis(this CapOptions options, Action<CapRedisOptions> configure)
+        {
+            if (configure is null)
+                throw new ArgumentNullException(nameof(configure));


Add directly after the if line, or use {} to include on a new line

yang-xiaodong · 2021-04-26T02:04:36Z

src/DotNetCore.CAP.Redis/IProcessingServer.PollPendingTopic.cs

@@ -0,0 +1,71 @@
+// Copyright (c) .NET Core Community. All rights reserved.


Is this the file to be deleted?

<ItemGroup> <Compile Remove="IProcessingServer.PollPendingTopic.cs" /> </ItemGroup>

yang-xiaodong · 2021-04-26T02:32:10Z

src/DotNetCore.CAP.Redis/CapOptions.Redis.Extensions.cs

+using System.Threading.Tasks;
+using System.Reflection;
+
+namespace DotNetCore.CAP.Redis


Should be namespace Microsoft.Extensions.DependencyInjection

src/DotNetCore.CAP.Redis/CapOptions.Redis.cs

yang-xiaodong · 2021-04-26T02:39:56Z

src/DotNetCore.CAP.Redis/CapOptions.Redis.PostConfigure.cs

+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Text;
+using System.Threading.Tasks;


Place System namespace at first

yang-xiaodong · 2021-04-26T02:40:52Z

src/DotNetCore.CAP.Redis/DotNetCore.CAP.Redis.csproj

+	<ItemGroup>
+	  <Compile Remove="IProcessingServer.PollPendingTopic.cs" />
+	</ItemGroup>


Unexpected file？

yang-xiaodong · 2021-04-26T02:41:17Z

src/DotNetCore.CAP.Redis/ICapOptionsExtension.Redis.cs

+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;


Place System namespace at first

yang-xiaodong · 2021-04-26T02:41:57Z

src/DotNetCore.CAP.Redis/ICapOptionsExtension.Redis.cs

+using System.Text;
+using System.Threading.Tasks;
+
+namespace DotNetCore.CAP.Redis


namespace DotNetCore.CAP

yang-xiaodong · 2021-04-26T06:30:57Z

src/DotNetCore.CAP.Redis/IRedisStream.Manager.Default.cs

+            catch (Exception ex)
+            {
+                logger.LogError($"Redis error when trying read consumer group {consumerGroup}", ex);
+                return (false, Array.Empty<RedisStream>());
+            }


According to my test, we will encounter two exception here.

When the Redis server restart, the first exception will be encountered

SocketFailure (ReadSocketError/ConnectionReset, last-recv: 5) on 192.168.3.57:6379/Interactive, Idle/Faulted, last:
XREADGROUP

After the redis server restarts, the consumer group will be lost, The exception information is as follows

NOGROUP No such key 'test-message' or consumer group 'Samples.Redis.SqlServer.v1' in XREADGROUP with GROUP option

PS: If transport throw BrokerConnectionException exception, the TransportCheckProcessor will restart consumer every 30 seconds

yang-xiaodong · 2021-04-26T07:08:31Z

Hello @MahmoudSamir101 , Thanks for your PR.

During my review process, I think we need to consider several points

DotNetCore.CAP.Redis may not be able to accurately distinguish between Transport and Storage, do we consider changing the name to DotNetCore.CAP.RedisStreams or other names that are easy to distinguish.
When the Redis server restarts, we need to consider how to re-register the consumer group, otherwise we will encounter NOGROUP exceptions. This may be because I did not enable AOF, but we need to consider it
When the storage fails, the consumer uses Reject to make the Transport resend the message. How do we do this now?

ensure consumer group creation while reading from stream

MahmoudSamir101 · 2021-04-27T10:52:18Z

@yang-xiaodong @xiangxiren thanks for your review, kindly find my response inline

DotNetCore.CAP.Redis may not be able to accurately distinguish between Transport and Storage, do we consider changing the name to DotNetCore.CAP.RedisStreams or other names that are easy to distinguish.

renamed to RedisStreams

When the Redis server restarts, we need to consider how to re-register the consumer group, otherwise we will encounter NOGROUP exceptions. This may be because I did not enable AOF, but we need to consider it

OK, good point, now while reading a consumer group, if it does not exist or the stream was not created, we are creating both again.

When the storage fails, the consumer uses Reject to make the Transport resend the message. How do we do this now?

Let me explain why we might not need that. Messages are not marked as ack until the CAP server invokes the client commit method while reading from a consumer group; therefore, if the storage fails, there will be no invoke for client commit to ack the message; instead, the message is marked as pending on the consumer group, and pending messages are consumed again with the next START of Cap server.

I believe we should double-check this assumption. The CAP server consumed pending messages again with the next start.

MahmoudSamir101 · 2021-04-29T08:19:12Z

@yang-xiaodong Will you want me to change the docs ?

yang-xiaodong · 2021-04-29T08:34:32Z

@MahmoudSamir101 Yes, that will be nice

MahmoudSamir101 · 2021-05-09T14:08:16Z

@yang-xiaodong, I created a PR #862 for documentation updates

mahmoud.samir added 2 commits April 1, 2021 00:42

add redis

05e0f53

add redis sample

1522d1c

add redis streams support

433460c

yang-xiaodong requested review from xiangxiren and yang-xiaodong April 25, 2021 04:28

xiangxiren reviewed Apr 26, 2021

View reviewed changes

yang-xiaodong reviewed Apr 26, 2021

View reviewed changes

Mahmoud S. Zeid added 3 commits April 27, 2021 12:52

change project name from Redis to RedisStreams

93ee3f2

ensure consumer group creation while reading from stream

rename project folder to RedisStreams

671b774

return stream position for exsiting stream/group

3cddb17

yang-xiaodong approved these changes Apr 28, 2021

View reviewed changes

xiangxiren merged commit ef1d1fa into dotnetcore:master Apr 29, 2021

		@@ -0,0 +1,71 @@
		// Copyright (c) .NET Core Community. All rights reserved.

Conversation

MahmoudSamir101 commented Mar 31, 2021

Uh oh!

yang-xiaodong commented Apr 6, 2021

Uh oh!

xiangxiren commented Apr 7, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

yang-xiaodong commented Apr 7, 2021

Uh oh!

MahmoudSamir101 commented Apr 7, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

yang-xiaodong commented Apr 8, 2021

Uh oh!

MahmoudSamir101 commented Apr 8, 2021

Uh oh!

yang-xiaodong commented Apr 19, 2021

Uh oh!

MahmoudSamir101 commented Apr 19, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

MahmoudSamir101 commented Apr 24, 2021

Uh oh!

xiangxiren left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yang-xiaodong commented Apr 26, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

MahmoudSamir101 commented Apr 27, 2021

Uh oh!

MahmoudSamir101 commented Apr 29, 2021

Uh oh!

yang-xiaodong commented Apr 29, 2021

Uh oh!

MahmoudSamir101 commented May 9, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

xiangxiren commented Apr 7, 2021 •

edited

Loading

MahmoudSamir101 commented Apr 7, 2021 •

edited

Loading

MahmoudSamir101 commented Apr 19, 2021 •

edited

Loading

xiangxiren left a comment •

edited

Loading

yang-xiaodong commented Apr 26, 2021 •

edited

Loading