library/src/main/java/com/kdgregory/aws/utils/kinesis/KinesisUtil.java
/**
     *  Creates a stream and waits for it to become available.
     *
     *  @param  client      The AWS client used to make requests.
     *  @param  streamName  The name of the stream.
     *  @param  numShards   The number of shards to create.
     *  @param  timeout     The number of milliseconds to wait for the stream to become available (this
     *                      also controls the number of attempts that we'll make to create the stream
     *                      if throttled).
     *
     *  @return The final status retrieved. If this is <code>null</code> it means that thes stream
     *          was not created. Caller should only use the stream if it's <code>ACTIVE</code>/
     *
     *  @throws IllegalArgumentException if thrown by the underlying calls (ie, this is not caught).
     */
Creates a stream and waits for it to become available.
@param  client      The AWS client used to make requests.
@param  streamName  The name of the stream.
@param  numShards   The number of shards to create.
@param  timeout     The number of milliseconds to wait for the stream to become available (this
also controls the number of attempts that we'll make to create the stream
if throttled).

@return The final status retrieved. If this is null it means that thes stream
was not created. Caller should only use the stream if it's ACTIVE

@throws IllegalArgumentException if thrown by the underlying calls .
Creates a stream and waits for it to become available . @param client The AWS client used to make requests . @param streamName The name of the stream . @param numShards The number of shards to create . @param timeout The number of milliseconds to wait for the stream to become available ( this also controls the number of attempts that we ' ll make to create the stream if throttled ) . @return The final status retrieved . If this is null it means that thes stream was not created . Caller should only use the stream if it ' s ACTIVE @throws IllegalArgumentException if thrown by the underlying calls .
public static StreamStatus createStream(AmazonKinesis client, String streamName, int numShards, long timeout)
    {
        logger.debug("createStream: " + streamName + ", with " + numShards + " shards");
        long currentSleep = 100;
        long abortAt = System.currentTimeMillis() + timeout;
        while (System.currentTimeMillis() < abortAt)
        {
            CreateStreamRequest request = new CreateStreamRequest()
                                              .withStreamName(streamName)
                                              .withShardCount(numShards);
            try
            {
                client.createStream(request);
                break;
            }
            catch (LimitExceededException ex)
            {
                if (CommonUtils.sleepQuietly(currentSleep)) return null;
                currentSleep *= 2;
            }
            catch (ResourceInUseException ex)
            {
                logger.warn("createStream called with existing stream: " + streamName);
                break;
            }
        }
        return waitForStatus(client, streamName, StreamStatus.ACTIVE, (abortAt - System.currentTimeMillis()));
    }
public static StreamStatus createStream ( AmazonKinesis client , String streamName , int numShards , long timeout ) { logger . debug ( "createStream: " + streamName + ", with " + numShards + " shards" ) ; long currentSleep = 100 ; long abortAt = System . currentTimeMillis ( ) + timeout ; while ( System . currentTimeMillis ( ) < abortAt ) { CreateStreamRequest request = new CreateStreamRequest ( ) . withStreamName ( streamName ) . withShardCount ( numShards ) ; try { client . createStream ( request ) ; break ; } catch ( LimitExceededException ex ) { if ( CommonUtils . sleepQuietly ( currentSleep ) ) return null ; currentSleep *= 2 ; } catch ( ResourceInUseException ex ) { logger . warn ( "createStream called with existing stream: " + streamName ) ; break ; } } return waitForStatus ( client , streamName , StreamStatus . ACTIVE , ( abortAt - System . currentTimeMillis ( ) ) ) ; }
Creates a stream and waits for it to become available.
Creates a stream and waits for it to become available .
// the documentation isn't very clear on this exception; it's thrown when // the stream already exists