StackExchange / StackExchange/StackExchange.Redis
When xreadgroup block reads data after a redistimeout, the first message cannot be consumed
Nobody has claimed this yet.
- Dominant language
- C#
- Stars
- 6.2k
- Forks
- 1.6k
- Avg merge
- 1d 15h
- Merged PRs (30d)
- 43
Description
- windows10
- redis 5.0.10
- dotnet any version
- StackExchange.Redis 2.6.111
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
class Program
{
static void Main()
{
var config = ConfigurationOptions.Parse("127.0.0.1:6379");
config.CommandMap = CommandMap.Create(new HashSet<string> { "SUBSCRIBE" }, false);
var redisRead = ConnectionMultiplexer.Connect(config);
var redisReadDb = redisRead.GetDatabase(1);
try
{
redisReadDb.StreamCreateConsumerGroup("test", "default", StreamPosition.NewMessages, true);
}
catch { }
var redisWrite = ConnectionMultiplexer.Connect(config);
var redisWriteDb = redisWrite.GetDatabase(1);
var writeTask = Task.Run(async () =>
{
while (true)
{
var id = await redisWriteDb.StreamAddAsync("test", "time", DateTime.Now.ToString());
var id2 = await redisWriteDb.StreamAddAsync("test", "time", DateTime.Now.ToString());
Console.WriteLine($"{DateTime.Now:O} write message:{id},{id2}");
Thread.Sleep(10000);
}
});
var readTask = Task.Run(async () =>
{
var arguments = new List<object>
{
"GROUP",
"default",
$"dotnet",
"COUNT",
10,
"BLOCK",
"0",
"STREAMS",
"test",
">"
};
while (true)
{
try
{
Console.WriteLine($"{DateTime.Now:O} redis xreadgroup block read ...");
var result = await redisReadDb.ExecuteAsync("xreadgroup", arguments).ConfigureAwait(false);
if (result.IsNull)
{
Console.WriteLine($"{DateTime.Now:O} redis xreadgroup isnull");
}
else
{
foreach (RedisResult[] subresults in (RedisResult[])result)
{
var streamName = (RedisValue)subresults[0];
foreach (RedisResult[] messages in (RedisResult[])subresults[1])
{
var id = (RedisValue)messages[0];
Console.WriteLine($"{DateTime.Now:O} read message:{id}");
await redisReadDb.StreamAcknowledgeAsync("test", "default", id);
}
}
}
}
catch (RedisTimeoutException)
{
Console.WriteLine($"{DateTime.Now:O} redis timeout ...");
}
}
});
Console.ReadLine();
}
}
2023-05-31T14:25:27.4733370+08:00 redis xreadgroup block read ...
2023-05-31T14:25:27.4976884+08:00 read message:1685514327495-0
2023-05-31T14:25:27.4994336+08:00 write message:1685514327495-0,1685514327498-0
2023-05-31T14:25:27.5082905+08:00 redis xreadgroup block read ...
2023-05-31T14:25:27.5101550+08:00 read message:1685514327498-0
2023-05-31T14:25:27.5107652+08:00 redis xreadgroup block read ...
2023-05-31T14:25:33.4848775+08:00 redis timeout ...
2023-05-31T14:25:33.4851831+08:00 redis xreadgroup block read ...
2023-05-31T14:25:37.5174767+08:00 write message:1685514337514-0,1685514337515-0
2023-05-31T14:25:37.5175269+08:00 read message:1685514337515-0
2023-05-31T14:25:37.5189520+08:00 redis xreadgroup block read ...
2023-05-31T14:25:43.4835083+08:00 redis timeout ...
2023-05-31T14:25:43.4837146+08:00 redis xreadgroup block read ...
2023-05-31T14:25:47.5262835+08:00 write message:1685514347525-0,1685514347526-0
2023-05-31T14:25:47.5262841+08:00 read message:1685514347526-0
2023-05-31T14:25:47.5276668+08:00 redis xreadgroup block read ...
2023-05-31T14:25:53.4743876+08:00 redis timeout ...
2023-05-31T14:25:53.4745574+08:00 redis xreadgroup block read ...
2023-05-31T14:25:57.5292349+08:00 write message:1685514357528-0,1685514357529-0
2023-05-31T14:25:57.5292381+08:00 read message:1685514357529-0
2023-05-31T14:25:57.5314306+08:00 redis xreadgroup block read ...
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the supplied reproduction and trace the ExecuteAsync("xreadgroup", arguments) call, RedisTimeoutException handling, and subsequent StreamAcknowledgeAsync calls. Verify the behavior around a blocking read timing out, and consider the issue done when the first message after the timeout is consumed correctly and the reproduction no longer shows it being skipped.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- csharp, redis
- Domain
- databases
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100