Kinesis 2.2.11 java unable to create consumer

user_14058953 0
Viewed 641

I need to migrate to Kinesis library to version 2.2.11 so I followed the tutorial: https://docs.aws.amazon.com/streams/latest/dev/kcl-migration.html

I need to run multiple instances of my consumer app, so every one of them needs to have an unique application name in order to have a separate lease table in DynamoDb. When initializing the consumer Kinesis runs DynamoDBLeaseRefresher.createLeaseTableIfNotExists which checks if a new table needs to be created for this application name and creates one if it cannot be found. So 2 operations are performed:

  1. DescribeTable - it returns the table info or throws a ResourceNotFoundExecption,
  2. if needed - CreateTable.

The problem for me is with the DescribeTable method. When I am looking for an existing table it returns it with no problem. But when I am looking for a non-existent table it throws the ResourceNotFoundExecption -> so far so good. Unfortunately it then gets wrapped and is now:

java.util.concurrent.CompletionException: software.amazon.awssdk.core.exception.SdkClientException: Unable to execute HTTP request: software.amazon.awssdk.awscore.exception.AwsServiceException$Builder.extendedRequestId(Ljava/lang/String;)Lsoftware/amazon/awssdk/awscore/exception/AwsServiceException$Builder;

and the app expecting ResourceNotFoundException gets something different instead and crashes. The wrapped exception message is a bit misleading: "Unable to execute HTTP request" since the request was performed and returned the proper message: "Resource not found".

Funny thing is that it sometimes works, the exception does not get wrapped, the CreateTable operation is performed and the consumer starts properly.

I have made a workaround for it for now where I just create the table before the initialization of the LeaseCoordinator, so it always gets the existing table.

here is my code:

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);
      }
    }

  }

}

Am I missing some configuration for the scheduler or something? Why is it sometimes working? Thanks

Edit: The problem is this block of code in DynamoDBLeaseRefresher.tableStatus() is invoked to check if the table exists:

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;
    }

and in my case it should get ResourceNotFoundException if the table is not found, but as I said the expection gets wrapped to CompletionException before it reaches the appropriate catch block and is caught in the code here:

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

This is happening 20 times in the loop while trying to Initialize the LeaseCoordinator and then just stops trying to initialize the connection. (As mentioned above it works occasionally, but that makes it even stranger to me) With my workaround it only needs 1 try to get initialized

amazon-kinesis amazon-dynamodb java
1 Answers

You don't need to create a lease table manually - DynamoDBLeaseCoordinator will create one if not exists on initialization and wait until it exists:


    @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"));
        }
    }

The issue in your case, I think, is that it's eventually created and you probably should periodically check until table appears - like DynamoDBLeaseCoordinator#initialize() does.

user_140583740
Related