获得服务:亚马逊运动;状态代码:502,带有apache flink和localstack kinesis

wko9yo5t  于 2021-06-24  发布在  Flink
关注(0)|答案(3)|浏览(380)

我的本地设置包括 local apache-flink (通过brew安装)和 localstack 在kinesis服务运行的情况下。
我的 Docker

  1. localstack:
  2. image: localstack/localstack:0.10.7
  3. environment:
  4. - SERVICES=kinesis
  5. ports:
  6. - "4568:4568"

还有我的运动消费者:

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

但是当我运行flink程序时,我得到了一个错误:
原因:org.apache.flink.kinesis.shaded.com.amazonaws.services.kinesis.model.amazonkinesexception:null(服务:amazonkinesis;状态代码:502;错误代码:空;请求id:空)
只有在使用 localstack . 如果我用我的aws帐户连接到我的kinesis流,它会工作得很好。

9lowa7mx

9lowa7mx1#

如果您使用的是java,那么可以使用 jar 一些amazon组件的库:
首先,您需要在 pom.xml 为了能够在测试期间直接初始化localstack:

  1. <dependency>
  2. <groupId>cloud.localstack</groupId>
  3. <artifactId>localstack-utils</artifactId>
  4. <version>0.2.0</version>
  5. <scope>test</scope>
  6. </dependency>

然后,如果需要使用 kinesis 以及 dynamo ,因为最新的一个 aws 与最新版本的localstack不兼容:

  1. <dependency>
  2. <groupId>com.amazonaws</groupId>
  3. <artifactId>aws-java-sdk-core</artifactId>
  4. <version>1.11.642</version>
  5. <scope>test</scope>
  6. </dependency>
  7. <dependency>
  8. <groupId>com.amazonaws</groupId>
  9. <artifactId>amazon-kinesis-client</artifactId>
  10. <version>1.8.10</version>
  11. <scope>test</scope>
  12. </dependency>

现在您可以使用以下注解来使用docker示例化堆栈,如果系统中不存在图像,则会自动提取图像。因此,不需要运行任何docker/docker compose映像。

  1. @LocalstackDockerProperties(services = {"kinesis", "dynamodb"})
  2. @ExtendWith(LocalstackDockerExtension.class)
  3. @Slf4j
  4. public class TestPipelineComplete {
  5. public static final String AWS_ACCESS_KEY_ID = "foo";
  6. public static final String AWS_SECRET_ACCESS_KEY = "bar";
  7. static {
  8. System.setProperty("AWS_ACCESS_KEY_ID", AWS_ACCESS_KEY_ID);
  9. System.setProperty("AWS_SECRET_ACCESS_KEY", AWS_SECRET_ACCESS_KEY);
  10. // These two lines are fundamental
  11. cloud.localstack.TestUtils.setEnv("aws.cborEnabled", "false");
  12. cloud.localstack.TestUtils.setEnv("AWS_CBOR_DISABLE", "true");
  13. }
  14. }

现在,如果你需要初始化 DynamoDB 客户端可以使用以下行:

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

现在,如果你需要初始化 Kinesis 客户端,您可以使用以下行:

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

如果需要从kinesis(testpurpouse)读取数据,可以使用以下代码段作为模板(https://gist.github.com/alessiosavi/4ea88d73d6853de695843631207b7bc6):

  1. package org.example;
  2. import com.amazonaws.services.kinesis.AmazonKinesis;
  3. import com.amazonaws.services.kinesis.AmazonKinesisClientBuilder;
  4. import com.amazonaws.services.kinesis.model.*;
  5. import java.nio.charset.StandardCharsets;
  6. import java.util.List;
  7. public class App {
  8. private static final String streamName = "API_NAME" + "_kineis-notification-stream";
  9. private static final AmazonKinesis client = AmazonKinesisClientBuilder.defaultClient();
  10. public static void main(String[] args) {
  11. printKinesisRecords(getRecordsFromKinesis(client));
  12. }
  13. private static List<Record> getRecordsFromKinesis(AmazonKinesis kClient) {
  14. final ListShardsRequest listShardsRequest = new ListShardsRequest().withStreamName(streamName).withMaxResults(1);
  15. Shard shard = kClient.listShards(listShardsRequest).getShards().get(0);
  16. GetShardIteratorRequest getShardIteratorRequest = new GetShardIteratorRequest();
  17. getShardIteratorRequest.setStreamName(streamName);
  18. getShardIteratorRequest.setShardId(shard.getShardId());
  19. getShardIteratorRequest.setShardIteratorType("TRIM_HORIZON");
  20. final GetShardIteratorResult getShardIteratorResult = kClient.getShardIterator(getShardIteratorRequest);
  21. String shardIterator = getShardIteratorResult.getShardIterator();
  22. // Create a new getRecordsRequest with an existing shardIterator
  23. // Set the maximum records to return to 1
  24. GetRecordsRequest getRecordsRequest = new GetRecordsRequest();
  25. getRecordsRequest.setShardIterator(shardIterator);
  26. getRecordsRequest.setLimit(10);
  27. final GetRecordsResult result = kClient.getRecords(getRecordsRequest);
  28. // Put the result into record list. The result can be empty.
  29. return result.getRecords();
  30. }
  31. private static void printKinesisRecords(List<Record> records) {
  32. for (Record record : records) {
  33. System.err.println("RECORD: " + StandardCharsets.UTF_8.decode(record.getData()).toString());
  34. }
  35. }
  36. }
展开查看全部
oknwwptz

oknwwptz2#

在添加flinkkinesisconsumer作为源之前,请添加以下行: System.setProperty("com.amazonaws.sdk.disableCbor", "true") System.setProperty("org.apache.flink.kinesis.shaded.com.amazonaws.sdk.disableCbor", "true") 这与导出env变量的效果相同,但将其放入代码中可以减少设置环境所花费的时间。

whhtz7ly

whhtz7ly3#

我们需要通过env var禁用cbor和cert检查,并在同一控制台中启动flink

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

相关问题