Kinesis 2.2.11 java 无法创建消费者

Question

我需要将 Kinesis 库迁移到 2.2.11 版本，所以我按照教程进行操作：https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html

我需要运行我的消费者应用程序的多个实例，因此每个实例都需要有一个唯一的应用程序名称，以便在 DynamoDb 中拥有一个单独的租约表。 初始化消费者 Kinesis 时运行 DynamoDBLeaseRefresher.createLeaseTableIfNotExists，它检查是否需要为此应用程序名称创建一个新表，如果找不到则创建一个。 所以执行了2个操作：

1. DescribeTable - 它返回表信息或抛出 ResourceNotFoundExecption，
2. 如果需要 - CreateTable。

我的问题在于 DescribeTable 方法。当我正在寻找一个现有的表时，它会毫无问题地返回它。但是当我正在寻找一个不存在的表时，它会抛出 ResourceNotFoundExecption -> 到目前为止一切都很好。不幸的是，它随后被包装起来，现在是：

java.util.concurrent.CompletionException：software.amazon.awssdk.core.exception.SdkClientException：无法执行 HTTP 请求：software.amazon.awssdk.awscore.exception.AwsServiceException$Builder.extendedRequestId(Ljava/lang/String ;)Lsoftware/amazon/awssdk/awscore/exception/AwsServiceException$Builder;

而预期 ResourceNotFoundException 的应用程序会得到不同的结果并崩溃。 包装的异常消息有点误导：“无法执行 HTTP 请求”，因为请求已执行并返回正确的消息：“找不到资源”。

有趣的是它有时会起作用，异常没有被包装，执行了 CreateTable 操作并且消费者正常启动。

我现在已经为它做了一个解决方法，我只是在初始化 LeaseCoordinator 之前创建表，所以它总是获取现有的表。

这是我的代码：

public KinesisStreamReaderService(String streamName, String applicationName, String regionName) {

KinesisAsyncClient kinesisClient = KinesisAsyncClient.builder()
      .credentialsProvider(EnvironmentVariableCredentialsProvider.create())
      .region(Region.of(connectionProperties.getRegion()))
      .httpClientBuilder(createHttpClientBuilder())
      .build();

DynamoDbAsyncClient dynamoClient = DynamoDbAsyncClient.builder().region(Region.of(regionName)).build();
    CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(Region.of(regionName)).build();

  //  if(!dynamoDbTableExists(dynamoClient, applicationName)) {
  //    createDynamoDbTable(dynamoClient, applicationName);
  //  }

    ConfigsBuilder configsBuilder = new ConfigsBuilder(streamName, applicationName, kinesisClient,
      dynamoClient, cloudWatchClient, workerId(), KinesisReaderProcessor::new);
    configsBuilder.retrievalConfig().initialPositionInStreamExtended(
      InitialPositionInStreamExtended.newInitialPosition(
        InitialPositionInStream.LATEST));

    scheduler = new Scheduler(
      configsBuilder.checkpointConfig(),
      configsBuilder.coordinatorConfig(),
      configsBuilder.leaseManagementConfig(),
      configsBuilder.lifecycleConfig(),
      configsBuilder.metricsConfig(),
      configsBuilder.processorConfig(),
      configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient))
    );
  }

  private void createDynamoDbTable(DynamoDbAsyncClient dynamoClient, String applicationName) {
    log.info("Creating new lease table: {}", applicationName);
    CompletableFuture<CreateTableResponse> createTableFuture = dynamoClient
      .createTable(CreateTableRequest.builder()
        .provisionedThroughput(ProvisionedThroughput.builder().readCapacityUnits(10L).writeCapacityUnits(10L).build())
        .tableName(applicationName)
        .keySchema(KeySchemaElement.builder().attributeName("leaseKey").keyType(KeyType.HASH).build())
        .attributeDefinitions(AttributeDefinition.builder().attributeName("leaseKey").attributeType(
          ScalarAttributeType.S).build())
        .build());
    try {
      CreateTableResponse createTableResponse = createTableFuture.get();
      log.debug("Created new lease table: {}", createTableResponse.tableDescription().tableName());
    } catch (InterruptedException | ExecutionException e) {
      throw new DataStreamException(e.getMessage(), e);
    }
  }

  private boolean dynamoDbTableExists(DynamoDbAsyncClient dynamoClient, String tableName) {
    CompletableFuture<DescribeTableResponse> describeTableResponseCompletableFutureNew = dynamoClient
      .describeTable(DescribeTableRequest.builder()
        .tableName(tableName).build());
    try {
      DescribeTableResponse describeTableResponseNew = describeTableResponseCompletableFutureNew
        .get();
      return nonNull(describeTableResponseNew);
    } catch (InterruptedException | ExecutionException e) {
      log.info(e.getMessage(), e);
    }
    return false;
  }

  private static String workerId() {
    String workerId;
    try {
      workerId = format("%s_%s", getLocalHost().getCanonicalHostName(), randomUUID().toString());
    } catch (UnknownHostException e) {
      workerId = randomUUID().toString();
    }
    return workerId;
  }

  @Override
  public void read(Consumer<String> consumer) {
    this.consumer = consumer;
    scheduler.run();
  }

  private class KinesisReaderProcessor implements ShardRecordProcessor {

    private String shardId;

    @Override
    public void initialize(InitializationInput initializationInput) {
      this.shardId = initializationInput.shardId();
      log.info("Initializing record processor for shard: {}", shardId);
    }

    @Override
    public void processRecords(ProcessRecordsInput processRecordsInput) {
      log.debug("Checking shard {} for new records", shardId);
      List<KinesisClientRecord> records = processRecordsInput.records();
      if (!records.isEmpty()) {
        log.debug("Processing {} records from kinesis stream shard {}", records.size(), shardId);
        records.forEach(record -> {
          String json = UTF_8.decode(record.data()).toString();
          log.info(json);
          consumer.accept(json);
        });
      }
    }

    @Override
    public void leaseLost(LeaseLostInput leaseLostInput) {
      log.info("Record processor has lost lease, terminating");
    }

    @Override
    public void shardEnded(ShardEndedInput shardEndedInput) {
      try {
        shardEndedInput.checkpointer().checkpoint();
      } catch (ShutdownException | InvalidStateException e) {
        log.error(e.getMessage(), e);
      }
    }

    @Override
    public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
      try {
        shutdownRequestedInput.checkpointer().checkpoint();
      } catch (ShutdownException | InvalidStateException e) {
        log.error(e.getMessage(), e);
      }
    }

  }

}

我是否缺少调度程序的一些配置或其他什么？为什么有时会起作用？ 谢谢

编辑： 问题是调用 DynamoDBLeaseRefresher.tableStatus() 中的这段代码来检查表是否存在：

DescribeTableResponse result;
    try {
      try {
        result = 
 (DescribeTableResponse)FutureUtils.resolveOrCancelFuture(this.dynamoDBClient.describeTable(request), this.dynamoDbRequestTimeout);
      } catch (ExecutionException var5) {
        throw exceptionManager.apply(var5.getCause());
      } catch (InterruptedException var6) {
        throw new DependencyException(var6);
      }
    } catch (ResourceNotFoundException var7) {
      log.debug("Got ResourceNotFoundException for table {} in leaseTableExists, returning false.", this.table);
      return null;
    }

在我的情况下，如果找不到表，它应该得到 ResourceNotFoundException，但正如我所说，期望在到达适当的 catch 块之前被包装到 CompletionException 并被捕获在此处的代码中：

catch (ExecutionException var5) {
        throw exceptionManager.apply(var5.getCause());

在尝试初始化 LeaseCoordinator 时，这在循环中发生了 20 次，然后停止尝试初始化连接。 （如上所述，它偶尔会起作用，但这让我更加陌生） 使用我的解决方法，它只需要 1 次尝试初始化

Answer 1

您不需要手动创建租用表 - DynamoDBLeaseCoordinator 将在初始化时创建一个如果不存在并等待它存在：

    @Override
    public void initialize() throws ProvisionedThroughputException, DependencyException, IllegalStateException {
        final boolean newTableCreated =
                leaseRefresher.createLeaseTableIfNotExists(initialLeaseTableReadCapacity, initialLeaseTableWriteCapacity);
        if (newTableCreated) {
            log.info("Created new lease table for coordinator with initial read capacity of {} and write capacity of {}.",
                    initialLeaseTableReadCapacity, initialLeaseTableWriteCapacity);
        }
        // Need to wait for table in active state.
        final long secondsBetweenPolls = 10L;
        final long timeoutSeconds = 600L;
        final boolean isTableActive = leaseRefresher.waitUntilLeaseTableExists(secondsBetweenPolls, timeoutSeconds);
        if (!isTableActive) {
            throw new DependencyException(new IllegalStateException("Creating table timeout"));
        }
    }

我认为，在您的情况下，问题在于它最终会被创建，您可能应该定期检查直到表格出现 - 就像 DynamoDBLeaseCoordinator#initialize() 那样。