private void processResponse ( Response < ByteArray , Object > response , Pipeline pipeline ) { if ( response == null ) { logger . warn ( "RoutingTimedout on waiting for async ops; parallelResponseToWait: " + numNodesPendingResponse + "; preferred-1: " + ( preferred - 1 ) + "; quorumOK: " + quorumSatisfied + "; zoneOK: " + zonesSatisfied ) ; } else { numNodesPendingResponse = numNodesPendingResponse - 1 ; numResponsesGot = numResponsesGot + 1 ; if ( response . getValue ( ) instanceof Exception && ! ( response . getValue ( ) instanceof ObsoleteVersionException ) ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "PUT {key:" + key + "} handling async put error" ) ; } if ( response . getValue ( ) instanceof QuotaExceededException ) { /** * TODO Not sure if we need to count this Exception for * stats or silently ignore and just log a warning. While * QuotaExceededException thrown from other places mean the * operation failed, this one does not fail the operation * but instead stores slops. Introduce a new Exception in * client side to just monitor how mamy Async writes fail on * exceeding Quota? * */ if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Received quota exceeded exception after a successful " + pipeline . getOperation ( ) . getSimpleName ( ) + " call on node " + response . getNode ( ) . getId ( ) + ", store '" + pipelineData . getStoreName ( ) + "', master-node '" + pipelineData . getMaster ( ) . getId ( ) + "'" ) ; } } else if ( handleResponseError ( response , pipeline , failureDetector ) ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "PUT {key:" + key + "} severe async put error, exiting parallel put stage" ) ; } return ; } if ( PipelineRoutedStore . isSlopableFailure ( response . getValue ( ) ) || response . getValue ( ) instanceof QuotaExceededException ) { /** * We want to slop ParallelPuts which fail due to * QuotaExceededException. * * TODO Though this is not the right way of doing things, in * order to avoid inconsistencies and data loss, we chose to * slop the quota failed parallel puts. * * As a long term solution - 1) either Quota management * should be hidden completely in a routing layer like * Coordinator or 2) the Server should be able to * distinguish between serial and parallel puts and should * only quota for serial puts * */ pipelineData . getSynchronizer ( ) . tryDelegateSlop ( response . getNode ( ) ) ; } if ( logger . isDebugEnabled ( ) ) { logger . debug ( "PUT {key:" + key + "} handled async put error" ) ; } } else { pipelineData . incrementSuccesses ( ) ; failureDetector . recordSuccess ( response . getNode ( ) , response . getRequestTime ( ) ) ; pipelineData . getZoneResponses ( ) . add ( response . getNode ( ) . getZoneId ( ) ) ; } } }
Process the response by reporting proper log and feeding failure detectors