private < T extends Operation & AttributeUpdaterOperation > CompletableFuture < Void > processAttributeUpdaterOperation ( T operation , TimeoutTimer timer ) { Collection < AttributeUpdate > updates = operation . getAttributeUpdates ( ) ; if ( updates == null || updates . isEmpty ( ) ) { // No need for extra complicated handling. return this . durableLog . add ( operation , timer . getRemaining ( ) ) ; } return Futures . exceptionallyCompose ( this . durableLog . add ( operation , timer . getRemaining ( ) ) , ex -> { // We only retry BadAttributeUpdateExceptions if it has the PreviousValueMissing flag set. ex = Exceptions . unwrap ( ex ) ; if ( ex instanceof BadAttributeUpdateException && ( ( BadAttributeUpdateException ) ex ) . isPreviousValueMissing ( ) ) { // Get the missing attributes and load them into the cache, then retry the operation, exactly once. SegmentMetadata segmentMetadata = this . metadata . getStreamSegmentMetadata ( operation . getStreamSegmentId ( ) ) ; Collection < UUID > attributeIds = updates . stream ( ) . map ( AttributeUpdate :: getAttributeId ) . filter ( id -> ! Attributes . isCoreAttribute ( id ) ) . collect ( Collectors . toList ( ) ) ; if ( ! attributeIds . isEmpty ( ) ) { // This only makes sense if a core attribute was missing. return getAndCacheAttributes ( segmentMetadata , attributeIds , true , timer ) . thenComposeAsync ( attributes -> { // Final attempt - now that we should have the attributes cached. return this . durableLog . add ( operation , timer . getRemaining ( ) ) ; } , this . executor ) ; } } // Anything else is non-retryable; rethrow. return Futures . failedFuture ( ex ) ; } ) ; }
Processes the given AttributeUpdateOperation with exactly one retry in case it was rejected because of an attribute update failure due to the attribute value missing from the in - memory cache .