AWS Kinesis NextShardIterator永远不会为空

问题描述

上下文：我正在尝试从Kinesis流中获取带有API参考的记录。我正在使用.Net Core（3.1版）。

我正在使用API​​将数据写入Kinesis Stream。这个问题没有任何问题。但是我在读取数据时遇到了一些问题。我将getRecord方法放在do-while循环中。 nextShardIterator值是否为null的条件是什么？但是这个值永远不会为空。我无法打破循环。

一些答案​​包括以下短语： “ NextShardIterator从分片开始顺序读取数据记录的分片中的下一个位置。如果设置为null，则分片已关闭并且请求的迭代器不再返回任何数据。”

我只有1个流和1个分片。我要放2张唱片。之后，我执行读取方法。它正在获取这些记录。但是在那之后，即使消耗了记录，nextShardIterator也永远不会为空。

• 列出流（我只有1个流）
• 描述流并获取分片（我只有1个分片）
• 获取TRIM_HORIZON类型的ShardIterator
• GetRecord循环执行（而{if nextShardIterator！= null}）
• 找到记录后，我正在尝试使用AFTER_SEQUENCE分片迭代器
• 如果无法获取任何记录，我正在尝试使用最新的分片迭代器

putRecord代码如下：

public async Task<ResponseModel> PutRecord(string orderFlow,string documentId)
        {
            byte[] bytedata = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(orderFlow));
            using MemoryStream memoryStream = new MemoryStream(bytedata);
            var putRecordRequest = new PutRecordRequest();
            putRecordRequest.StreamName = myStreamName;
            putRecordRequest.PartitionKey = "partition" + documentId;
            putRecordRequest.Data = memoryStream;
            try
            {
                var putRecordResponse = await kinesisClient.PutRecordAsync(putRecordRequest);
                return new ResponseModel
                {
                    Data = putRecordResponse,Status = ResponseStatus.Success,Message = "Successfully put record!"
                };
            }
            catch (Exception e)
            {
                return new ResponseModel
                {
                    Data = null,Status = ResponseStatus.Error,Message = "PutRecord Error: " + e.Message
                };
            }
        }

但是我在读取数据时遇到了一些问题。 GetRecord代码是这样的：

public async Task<ResponseModel> GetRecords()
        {
            var listStreams = await ListStreams();
            if (listStreams.StreamNames.Count == 0)
            {
                return new ResponseModel
                {
                    Data = null,Status = ResponseStatus.Warning,Message = "Do not have any Stream!"
                };
            }
            myStreamName = listStreams.StreamNames[0];

            var describeStreams = await DescribeStream(listStreams.StreamNames[0]);
            if (describeStreams.StreamDescription.StreamStatus != "ACTIVE")
            {
                return new ResponseModel
                {
                    Data = null,Message = "Stream status: " + describeStreams.StreamDescription.StreamStatus
                };
            }

            var shards = describeStreams.StreamDescription.Shards;
            if (shards.Count == 0)
            {
                return new ResponseModel
                {
                    Data = null,Message = "Do not have any Shard (or data)!"
                };
            }
            var shardId = shards[0].ShardId;

            var shardIterator = await GetShardIterator(shardId);
            if (string.IsNullOrWhiteSpace(shardIterator.ShardIterator))
            {
                return new ResponseModel
                {
                    Data = null,Message = "ShardIterator is null or empty!"
                };
            }

            var getRecords = await GetRecords(shardIterator.ShardIterator);
            Console.WriteLine("First Iterator: " + shardIterator);
            var dataList = new List<Record>();

            do
            {
                if (getRecords.Records.Count == 0)
                {
                    Console.WriteLine("Records are empty!");
                    var nextShardIterator = GetShardLatest(shardId).Result.ShardIterator;
                    getRecords = await GetRecords(nextShardIterator);
                    Console.WriteLine("Latest Iterator: " + nextShardIterator);
                }
                else
                {
                    Console.WriteLine("We have records!");
                    foreach (var record in getRecords.Records)
                    {
                        dataList.Add(record);
                    }
                    var nextShardIterator = GetShardIteratorWithSequence(shardId,getRecords.Records[getRecords.Records.Count-1].SequenceNumber).Result.ShardIterator;
                    Console.WriteLine("AfterSequence Iterator: " + nextShardIterator);
                    getRecords = await GetRecords(nextShardIterator);
                }
            } while (getRecords.NextShardIterator != null);

            return new ResponseModel
            {
                Data = dataList,Message = "Successfull"
            };
        }

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）