private static void waitStreamActivation ( AmazonKinesis consumer , String streamName , long streamCreationTimeoutMillis ) { DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest ( ) . withStreamName ( streamName ) . withLimit ( 1 ) ; DescribeStreamResult describeStreamResult = null ; String streamStatus = null ; long endTime = System . currentTimeMillis ( ) + streamCreationTimeoutMillis ; do { try { describeStreamResult = consumer . describeStream ( describeStreamRequest ) ; streamStatus = describeStreamResult . getStreamDescription ( ) . getStreamStatus ( ) ; if ( ACTIVE_STREAM_STATUS . equals ( streamStatus ) ) { break ; } Thread . sleep ( 100 ) ; } catch ( ResourceNotFoundException | LimitExceededException ignored ) { // ignored } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; throw new AwsKinesisException ( "Thread interrupted while waiting for stream activation" , args -> args . add ( "streamName" , streamName ) , e ) ; } } while ( System . currentTimeMillis ( ) < endTime ) ; if ( describeStreamResult == null || streamStatus == null || ! streamStatus . equals ( ACTIVE_STREAM_STATUS ) ) { throw new AwsKinesisException ( "Stream never went active" , args -> args . add ( "streamName" , streamName ) . add ( "streamCreationTimeoutMillis" , streamCreationTimeoutMillis ) ) ; } }
Waits that the stream has been created .