FanOutConsumerRegistration does not handle initial registration properly

[lbourdages]

When booting a consumer using enhanced fan-out for the first time, I get the following two errors repeatedly:

ERROR s.a.k.r.f.FanOutConsumerRegistration - Status of StreamConsumer <consumer_name_here>, was not ACTIVE after all retries. Was instead CREATING.
ERROR s.a.kinesis.coordinator.Scheduler - Worker.run caught exception, sleeping for 1000 milli seconds!
java.lang.IllegalStateException: Status of StreamConsumer <consumer_name_here, was not ACTIVE after all retries. Was instead CREATING.

The number of times the errors get logged varies depending on the time it takes for the registration to become active but it is around 10 in my tests.

This occurs because of the following function in software.amazon.kinesis.retrieval.fanout.FanOutConsumerRegistration:

    private void waitForActive() throws DependencyException {
        ConsumerStatus status = null;

        int retries = maxDescribeStreamConsumerRetries;

        while (!ConsumerStatus.ACTIVE.equals(status) && retries > 0) {
            status = describeStreamConsumer().consumerDescription().consumerStatus();
            retries--;
        }

        if (!ConsumerStatus.ACTIVE.equals(status)) {
            final String message = String.format(
                    "Status of StreamConsumer %s, was not ACTIVE after all retries. Was instead %s.",
                    streamConsumerName, status);
            log.error(message);
            throw new IllegalStateException(message);
        }
    }

There is no sleep of any kind in the while loop. Thus, the checks will be performed as fast as possible, being only slowed down by the network latency and the speed of the kinesis API. With the default setting (10 retries), it will instantaneously throw the IllegalStateException which aborts the creation of the worker and starts it over. This should not happen as it is not an anomalous condition; an example where a similar case is handled correctly would be in software.amazon.kinesis.leases.dynamodb.DynamoDBLeaseCoordinator#initialize where the code patiently waits until the lease table is created in DynamoDB

I was able to prevent error logs by setting the maxDescribeStreamConsumerRetries value to an absurdly high number (such as 10000) - but it only works because the describeStreamConsumer calls get throttled and this induces backoff.

The fix seems fairly easy: all that's required is to add a sleep between the retries. There are really only two cases - either the consumer wasn't registered, in which case it's best to just sleep, or it is already registered and in that case the status should already be active and thus the sleep won't have any impact.

[sahilpalvia]

Closing this issue out. The PR has been merged, and the fix will be available in the next release.

Please feel free to reopen tt if problem persists.