library/src/main/java/com/kdgregory/aws/utils/kinesis/KinesisUtil.java
/**
     *  Updates a stream's retention period and waits for it to become available again. This
     *  function looks at the current retention period and chooses either the "increase" or
     *  "decrease" operation. Calling with the same retention period is a no-op (other than
     *  calls to retrieve the stream status).
     *
     *  @param  client      The AWS client used to make requests.
     *  @param  streamName  The name of the stream.
     *  @param  retention   The new retention period (hours)
     *  @param  timeout     The maximum amount of time (millis) that we'll spend trying to
     *                      update the stream or wait for it to become available.
     *
     *  @return The stream status. After successful update this will be ACTIVE. It will be
     *          <code>UPDATING</code> if the update operation completed but the stream was
     *          still updating when the timeout expired. If the update call timed out or
     *          the stream does not exist, will return <code>null</code>.
     *
     *  @throws IllegalArgumentException if thrown by underlying call. The AWS docs don't
     *          give specific reasons for this exception, but one cause is specifying a
     *          retention period outside the allowed 24..168 hours.
     */
Updates a stream's retention period and waits for it to become available again. This
function looks at the current retention period and chooses either the "increase" or
"decrease" operation. Calling with the same retention period is a no-op (other than
calls to retrieve the stream status).

@param  client      The AWS client used to make requests.
@param  streamName  The name of the stream.
@param  retention   The new retention period (hours)
@param  timeout     The maximum amount of time (millis) that we'll spend trying to
update the stream or wait for it to become available.

@return The stream status. After successful update this will be ACTIVE. It will be
UPDATING if the update operation completed but the stream was
still updating when the timeout expired. If the update call timed out or
the stream does not exist, will return null.

@throws IllegalArgumentException if thrown by underlying call. The AWS docs don't
give specific reasons for this exception, but one cause is specifying a
retention period outside the allowed 24..168 hours.
Updates a stream ' s retention period and waits for it to become available again . This function looks at the current retention period and chooses either the " increase " or " decrease " operation . Calling with the same retention period is a no - op ( other than calls to retrieve the stream status ) . @param client The AWS client used to make requests . @param streamName The name of the stream . @param retention The new retention period ( hours ) @param timeout The maximum amount of time ( millis ) that we ' ll spend trying to update the stream or wait for it to become available . @return The stream status . After successful update this will be ACTIVE . It will be UPDATING if the update operation completed but the stream was still updating when the timeout expired . If the update call timed out or the stream does not exist will return null . @throws IllegalArgumentException if thrown by underlying call . The AWS docs don ' t give specific reasons for this exception but one cause is specifying a retention period outside the allowed 24 .. 168 hours .
public static StreamStatus updateRetentionPeriod(AmazonKinesis client, String streamName, int retention, long timeout)
    {
        logger.debug("updateRetentionPeriod: " + streamName + " to " + retention + " hours");
        long timeoutAt = System.currentTimeMillis() + timeout;
        int currentSleep = 100;
        while (System.currentTimeMillis() < timeoutAt)
        {
            StreamStatus initialStatus = waitForStatus(client, streamName, StreamStatus.ACTIVE, timeout);
            if (initialStatus == null)
            {
                return null;
            }
            DescribeStreamRequest describeRequest = new DescribeStreamRequest().withStreamName(streamName);
            StreamDescription description = describeStream(client, describeRequest, (timeoutAt - System.currentTimeMillis()));
            if (description == null)
            {
                return null;
            }
            if (getStatus(description) != StreamStatus.ACTIVE)
            {
                continue;
            }
            int currentRetentition = description.getRetentionPeriodHours().intValue();
            if (currentRetentition == retention)
            {
                return StreamStatus.ACTIVE;
            }
            try
            {
                if (currentRetentition > retention)
                {
                    DecreaseStreamRetentionPeriodRequest request = new DecreaseStreamRetentionPeriodRequest()
                                                                   .withStreamName(streamName)
                                                                   .withRetentionPeriodHours(retention);
                    client.decreaseStreamRetentionPeriod(request);
                    break;
                }
                else
                {
                    IncreaseStreamRetentionPeriodRequest request = new IncreaseStreamRetentionPeriodRequest()
                                                                   .withStreamName(streamName)
                                                                   .withRetentionPeriodHours(retention);
                    client.increaseStreamRetentionPeriod(request);
                    break;
                }
            }
            catch (LimitExceededException ex)
            {
            }
            catch (ResourceInUseException ex)
            {
            }
            catch (ResourceNotFoundException ex)
            {
                logger.warn("updateRetentionPeriod called for nonexistent stream: " + streamName);
                return null;
            }
            CommonUtils.sleepQuietly(currentSleep);
            currentSleep *= 2;
        }
        return waitForStatus(client, streamName, StreamStatus.ACTIVE, (timeoutAt - System.currentTimeMillis()));
    }
public static StreamStatus updateRetentionPeriod ( AmazonKinesis client , String streamName , int retention , long timeout ) { logger . debug ( "updateRetentionPeriod: " + streamName + " to " + retention + " hours" ) ; long timeoutAt = System . currentTimeMillis ( ) + timeout ; int currentSleep = 100 ; while ( System . currentTimeMillis ( ) < timeoutAt ) { StreamStatus initialStatus = waitForStatus ( client , streamName , StreamStatus . ACTIVE , timeout ) ; if ( initialStatus == null ) { return null ; } DescribeStreamRequest describeRequest = new DescribeStreamRequest ( ) . withStreamName ( streamName ) ; StreamDescription description = describeStream ( client , describeRequest , ( timeoutAt - System . currentTimeMillis ( ) ) ) ; if ( description == null ) { return null ; } if ( getStatus ( description ) != StreamStatus . ACTIVE ) { continue ; } int currentRetentition = description . getRetentionPeriodHours ( ) . intValue ( ) ; if ( currentRetentition == retention ) { return StreamStatus . ACTIVE ; } try { if ( currentRetentition > retention ) { DecreaseStreamRetentionPeriodRequest request = new DecreaseStreamRetentionPeriodRequest ( ) . withStreamName ( streamName ) . withRetentionPeriodHours ( retention ) ; client . decreaseStreamRetentionPeriod ( request ) ; break ; } else { IncreaseStreamRetentionPeriodRequest request = new IncreaseStreamRetentionPeriodRequest ( ) . withStreamName ( streamName ) . withRetentionPeriodHours ( retention ) ; client . increaseStreamRetentionPeriod ( request ) ; break ; } } catch ( LimitExceededException ex ) { } catch ( ResourceInUseException ex ) { } catch ( ResourceNotFoundException ex ) { logger . warn ( "updateRetentionPeriod called for nonexistent stream: " + streamName ) ; return null ; } CommonUtils . sleepQuietly ( currentSleep ) ; currentSleep *= 2 ; } return waitForStatus ( client , streamName , StreamStatus . ACTIVE , ( timeoutAt - System . currentTimeMillis ( ) ) ) ; }
Updates a stream's retention period and waits for it to become available again.
Updates a stream ' s retention period and waits for it to become available again .
// this is certainly not the most efficient way to do things, but the code is simple // and nobody is going to be calling this method in a tight loop // probable timeout while waiting for stream to become active // there is a chance that status changes between the last describe and this one // but we'll simply let the update call fail rather than use an inner test // stream has magically disappeared between two calls or we've timed out // race condition, someone else is changing stream so wait until they're done // nothing to see here, move along // fall out to the end of the loop // fall out to the end of the loop // stream doesn't exist so no point in trying to do anything more