AWS Kinesis Client Libraryでコンシューマ開発 のバックアップ(No.1) - 闘うITエンジニアの覚え書き

aws kinesis create-stream --stream-name myFirstStream --shard-count 1

サンプルは以下にあるが、なぜか Antベースなので、ここではMavenで作成し直す。
https://github.com/aws/aws-sdk-java/tree/master/src/samples/AmazonKinesis

やる事は、EclipseからMavenプロジェクトを作成し、amazon-kinesis-clientとgsonを依存ライブラリとして追記するだけ。

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>KclSample</groupId>
  <artifactId>KclSample</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <dependencies>
    <dependency>
      <groupId>com.amazonaws</groupId>
      <artifactId>amazon-kinesis-client</artifactId>
      <version>1.9.0</version>
    </dependency>
    <dependency>
      <groupId>com.google.code.gson</groupId>
      <artifactId>gson</artifactId>
      <version>2.8.2</version>
    </dependency>
  </dependencies>
</project>

https://github.com/aws/aws-sdk-java/tree/master/src/samples/AmazonKinesis

サンプルは少し古いKCLのバージョン用のものになっているのと、ローカルでワーカーを起動したいので一部変更する。
また、IRecordProcessor バージョン1用のサンプルなので、バージョン2用を試したい場合は、必要に応じて書き換える。
https://docs.aws.amazon.com/ja_jp/streams/latest/dev/kinesis-record-processor-implementation-app-java.html#kcl-java-interface-v2

AmazonKinesisApplicationSample.java

    public static void main(String[] args) throws Exception {
          ・
          ・
        KinesisClientLibConfiguration kinesisClientLibConfiguration =
                new KinesisClientLibConfiguration(SAMPLE_APPLICATION_NAME,
                        SAMPLE_APPLICATION_STREAM_NAME,
                        credentialsProvider,
                        workerId)
        .withInitialPositionInStream(SAMPLE_APPLICATION_INITIAL_POSITION_IN_STREAM)
        .withKinesisEndpoint("https://kinesis.ap-northeast-1.amazonaws.com");  // 使用するリージョンのエンドポイント
/*     LocalStackで試したかったが動かなかった・・（未調査）
        .withKinesisEndpoint("http://localhost:4568/")
        .withDynamoDBEndpoint("http://localhost:4569/");
*/
          ・
          ・

AmazonKinesisApplicationSampleRecordProcessor.java

import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownReason;
↓
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.ShutdownReason;

あとは AmazonKinesisApplicationSampleRecordProcessor を弄っていく。

AmazonKinesisApplicationSample.java を Eclipseから実行する。
注意点としては、デフォルトの aws プロファイルが使用されるので、事前にaws configureしておく事。
（別のプロファイルを使用する場合は、プロパティファイルで指定する。）

testdata.json

{
  "var1":  "ABCDEFG",
  "timeStamp": "2018-02-12 18:50:05",
  "data": { "data1": "11111", "data2": "22222" }
}

AWS CLI からKinesisにレコード出力

aws kinesis put-record --stream-name myFirstStream --partition-key 123 --data file://testdata.json

aws kinesis delete-stream --stream-name myFirstStream