适用于消费者的 Amazon Kinesis KCL 客户端无法在 .NET 中运行

Question

请帮忙。我在 .NET 控制台应用程序中为 Kinesis 数据流设置消费者时遇到问题。

我已根据文档完成所有操作，但每当我运行消费者时，我仍然会得到一个空白的控制台屏幕。到目前为止，生产者工作正常，AWS 凭证正在工作。

1. 我的系统上的 JDK 配置良好（不是 Java 开发新手）
2. 我的 IAM 用户附加了所有必要的策略
3. 我可以看到生产者可以使用相同的 AWS 凭证以编程方式创建流、描述流等

我可以在创建 KclProcess 时在 Program 中打断点，但我无法在下面的 KinesisTest 类中打任何断点

对于消费者，我创建了一个类 Program.cs，如下所示：

class Program  
{

    public static void Main(string[] args)
    {
        //added these lines after trying everything
        Environment.SetEnvironmentVariable("AWS_ACCESS_KEY_ID", "***");
        Environment.SetEnvironmentVariable("AWS_SECRET_ACCESS_KEY", "***");
        Environment.SetEnvironmentVariable("AWS_REGION", "us-east-1");

        try
        {
            KclProcess.Create(new KinesisTest()).Run();
        }
        catch (Exception e)
        {
            Console.Error.WriteLine("ERROR: " + e);
        }
    }

}

和另一个班级

public class KinesisTest: IRecordProcessor
{

    private static readonly TimeSpan Backoff = TimeSpan.FromSeconds(3);
    private static readonly TimeSpan CheckpointInterval = TimeSpan.FromMinutes(1);
    private static readonly int NumRetries = 10;

    /// <value>The shard ID on which this record processor is working.</value>
    private string _kinesisShardId;

    private DateTime _nextCheckpointTime = DateTime.UtcNow;


    public void Initialize(InitializationInput input)
    {
        Console.Error.WriteLine("Initializing record processor for shard: " + input.ShardId);
        this._kinesisShardId = input.ShardId;
    }

    public void ProcessRecords(ProcessRecordsInput input)
    {
        Console.Error.WriteLine("Processing " + input.Records.Count + " records from " + _kinesisShardId);
        ProcessRecordsWithRetries(input.Records);

        // Checkpoint once every checkpoint interval.
        if (DateTime.UtcNow >= _nextCheckpointTime)
        {
            Checkpoint(input.Checkpointer);
            _nextCheckpointTime = DateTime.UtcNow + CheckpointInterval;
        }
    }

    public void Shutdown(ShutdownInput input)
    {
        Console.Error.WriteLine("Shutting down record processor for shard: " + _kinesisShardId);
        // Checkpoint after reaching end of shard, so we can start processing data from child shards.
        if (input.Reason == ShutdownReason.TERMINATE)
        {
            Checkpoint(input.Checkpointer);
        }
    }

    private void ProcessRecordsWithRetries(List<Record> records)
    {
        foreach (Record rec in records)
        {
            bool processedSuccessfully = false;
            string data = null;
            for (int i = 0; i < NumRetries; ++i)
            {
                try
                {
                    data = System.Text.Encoding.UTF8.GetString(rec.Data);

                    Console.Error.WriteLine( String.Format("Retrieved record:\n\tpartition key = {0},\n\tsequence number = {1},\n\tdata = {2}", rec.PartitionKey, rec.SequenceNumber, data));

                    // Your own logic to process a record goes here.

                    processedSuccessfully = true;
                    break;
                }
                catch (Exception e)
                {
                    Console.Error.WriteLine("Exception processing record data: " + data, e);
                }

                //Back off before retrying upon an exception.
                Thread.Sleep(Backoff);
            }

            if (!processedSuccessfully)
            {
                Console.Error.WriteLine("Couldn't process record " + rec + ". Skipping the record.");
            }
        }
    }

    private void Checkpoint(Checkpointer checkpointer)
    {
        Console.Error.WriteLine("Checkpointing shard " + _kinesisShardId);

        checkpointer.Checkpoint(RetryingCheckpointErrorHandler.Create(NumRetries, Backoff));
    }
}

最后是 kcl.properties 文件：

executableName = dotnet KinesisTest.dll

streamName = testStream

applicationName = KinesisTest

AWSCredentialsProvider = DefaultAWSCredentialsProviderChain

processingLanguage = C#

initialPositionInStream = TRIM_HORIZON

regionName = us-east-1

maxRecords = 5000

idleTimeBetweenReadsInMillis = 1000

# failoverTimeMillis = 10000
# workerId =
# shardSyncIntervalMillis = 60000
# callProcessRecordsEvenForEmptyRecordList = false
# parentShardPollIntervalMillis = 10000
# cleanupLeasesUponShardCompletion = true
# taskBackoffTimeMillis = 500
# metricsBufferTimeMillis = 10000
# metricsMaxQueueSize = 10000
# validateSequenceNumberBeforeCheckpointing = true
# maxActiveThreads = 0

如果我做错了什么，请告诉我。

我期待看到消费者处理流中的数据，但它只是一个空控制台

Answer 1

虽然我从未找到这个问题的答案，但我找到了使用 lambda 更好、更合适的方法。

我的最终设置涉及使用 SNS 从 Kinesis 获取消息/事件，然后将消息散播给任何订阅者（在我的例子中是 SQS 队列），然后将排队的消息提供给 lambdas。

类似这样的：

                             SQS Queue ----------------> Lambda
                             ^
                            /
                           /
                          /
KINESIS STREAM ------->SNS-------->SQS Queue ------> Lambda
                          \
                           \
                            \
                             >SQS Queue ----------------> Lambda

您可能会问为什么不将 Kinesis 直接连接到 SQS？或者为什么不直接 SNS 到 Lambda？

第一个问题的答案是 Kinesis 不是为多个 读。 SNS 允许您阅读一次并将其散播给尽可能多的人 您需要的订阅者。如果您说有 20 个正在等待的听众 要对事件采取行动，您不能将它们全部插入 Kinesis。所以去SNS吧。

第二个问题的答案是，SNS 有时会丢失消息并且 不提供或触发 lambdas（非常正确）。然而（推测 根据经验）如果您将其连接到 SQS，它几乎总是在队列中。然后队列可以触发 lambdas.. 反正他们只有一份工作。

所以我希望这对某人有所帮助。