获取服务：AmazonKinesis；状态代码：502 与 apache-flink 和 localstack Kinesis

Question

我的本地设置包括 local apache-flink（通过 brew 安装）和 localstack 以及 Kinesis 服务运行ning。

我的 docker-compose 有

  localstack:
    image: localstack/localstack:0.10.7
    environment:
      - SERVICES=kinesis
    ports:
      - "4568:4568"

和我的 Kinesis 消费者：

kinesisConsumerConfig.setProperty(ConsumerConfigConstants.AWS_ACCESS_KEY_ID, "123");
kinesisConsumerConfig.setProperty(ConsumerConfigConstants.AWS_SECRET_ACCESS_KEY, "123");
kinesisConsumerConfig.setProperty(ConsumerConfigConstants.AWS_ENDPOINT, "http://localhost:4568");

但是当我运行 Flink 程序时我得到这个错误：

Caused by: org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.model.AmazonKinesisException: null (Service: AmazonKinesis; Status Code: 502; Error Code: null; Request ID: null)

只有在使用 localstack 时才会发生。如果我在我的 AWS 账户上连接到我的 Kinesis 流，它会完美运行。

Answer 1

如果您正在使用 Java，您可以使用 jar 库来模拟一些亚马逊组件：

首先，您需要在 pom.xml 中添加以下组件，以便能够在测试期间直接初始化本地堆栈：

<dependency>
    <groupId>cloud.localstack</groupId>
    <artifactId>localstack-utils</artifactId>
    <version>0.2.0</version>
    <scope>test</scope>
</dependency>

然后，如果需要使用kinesis和dynamo，则需要指定以下库，因为aws提供的最新版本与最新版本不兼容本地堆栈：

<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>aws-java-sdk-core</artifactId>
    <version>1.11.642</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>com.amazonaws</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>1.8.10</version>
    <scope>test</scope>
</dependency>

现在您可以使用以下注释来实例化使用 docker 的堆栈，如果系统中不存在图像，系统将自动拉取这些图像。所以没有必要运行任何 docker/docker-compose 图片。

@LocalstackDockerProperties(services = {"kinesis", "dynamodb"})
@ExtendWith(LocalstackDockerExtension.class)
@Slf4j
public class TestPipelineComplete {

public static final String AWS_ACCESS_KEY_ID = "foo";
public static final String AWS_SECRET_ACCESS_KEY = "bar";
    static {
        System.setProperty("AWS_ACCESS_KEY_ID", AWS_ACCESS_KEY_ID);
        System.setProperty("AWS_SECRET_ACCESS_KEY", AWS_SECRET_ACCESS_KEY);
        // These two lines are fundamental
        cloud.localstack.TestUtils.setEnv("aws.cborEnabled", "false");
        cloud.localstack.TestUtils.setEnv("AWS_CBOR_DISABLE", "true");
    }
}

现在，如果您需要初始化一个 DynamoDB 客户端，您可以使用以下行：

final AmazonDynamoDB clientDynamoDB = cloud.localstack.TestUtils.getClientDynamoDB();

现在，如果你需要初始化一个Kinesis客户端，你可以使用下面一行：

final AmazonKinesis kinesisClient = cloud.localstack.TestUtils.getClientKinesis();

如果需要从kinesis中读取数据（测试目的），可以使用以下代码片段作为模板（https://gist.github.com/alessiosavi/4ea88d73d6853de695843631207b7bc6）：

package org.example;

import com.amazonaws.services.kinesis.AmazonKinesis;
import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
import com.amazonaws.services.kinesis.model.*;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class App {

    private static final String streamName = "API_NAME" + "_kineis-notification-stream";
    private static final AmazonKinesis client = AmazonKinesisClientBuilder.defaultClient();

    public static void main(String[] args) {
        printKinesisRecords(getRecordsFromKinesis(client));
    }


    private static List<Record> getRecordsFromKinesis(AmazonKinesis kClient) {
        final ListShardsRequest listShardsRequest = new ListShardsRequest().withStreamName(streamName).withMaxResults(1);

        Shard shard = kClient.listShards(listShardsRequest).getShards().get(0);
        GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
        getShardIteratorRequest.setStreamName(streamName);
        getShardIteratorRequest.setShardId(shard.getShardId());
        getShardIteratorRequest.setShardIteratorType("TRIM_HORIZON");

        final GetShardIteratorResult getShardIteratorResult = kClient.getShardIterator(getShardIteratorRequest);
        String shardIterator = getShardIteratorResult.getShardIterator();

        // Create a new getRecordsRequest with an existing shardIterator
        // Set the maximum records to return to 1
        GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
        getRecordsRequest.setShardIterator(shardIterator);
        getRecordsRequest.setLimit(10);

        final GetRecordsResult result = kClient.getRecords(getRecordsRequest);
        // Put the result into record list. The result can be empty.
        return result.getRecords();
    }

    private static void printKinesisRecords(List<Record> records) {
        for (Record record : records) {
            System.err.println("RECORD: " + StandardCharsets.UTF_8.decode(record.getData()).toString());
        }
    }
}

Answer 2

原来我们需要通过 ENV var 禁用 cbor 和证书检查并在同一个控制台中启动 flink

export AWS_CBOR_DISABLE=1
DISABLE_CERT_CHECKING_JAVA_OPTS="-Dorg.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCertChecking"
export FLINK_ENV_JAVA_OPTS=${DISABLE_CERT_CHECKING_JAVA_OPTS}
/usr/local/Cellar/apache-flink/1.9.1/libexec/bin/start-cluster.sh

Answer 3

在将 FlinkKinesisConsumer 添加为源之前添加这些行：

System.setProperty("com.amazonaws.sdk.disableCbor", "true") System.setProperty("org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCbor", "true")

这与导出 ENV 变量具有相同的效果，但在代码中使用它可以减少设置环境所花费的时间。

获取服务：AmazonKinesis；状态代码：502 与 apache-flink 和 localstack Kinesis

Getting Service: AmazonKinesis; Status Code: 502 with apache-flink and localstack Kinesis

error-code

apache-flink

localstack