@Override
  public void consume() throws Exception {
    try {
      networkQueue.poll(responseInfo -> {
        PutRequest putRequest = (PutRequest) responseInfo.getRequestInfo().getRequest();
        BlobId blobId = putRequest.getBlobId();
        int correlationId = putRequest.getCorrelationId();

        try {
          if (responseInfo.getError() != null) {
            logger.info("Error for correlation id {} blob id {} {} ", correlationId, blobId, responseInfo.getError());
            return;
          }
          InputStream serverResponseStream = new NettyByteBufDataInputStream(responseInfo.content());
          PutResponse putResponse = PutResponse.readFrom(new DataInputStream(serverResponseStream));
          responseInfo.release();
          putResponse.release();
          logger.info("received success response and decoded response successfully for correlation id {} {}",
              correlationId, blobId);
        } catch (Exception e) {
          logger.error("Error while processing response", e);
        }
      });
    } catch (ShutDownException e) {
      logger.info("Shutting down consumer as shutdown exception received");
      throw e;
    } catch (Exception e) {
      logger.error("Error while consuming", e);
    }
  }
Polls {@link #networkQueue} and passes the function which decodes response
@throws ShutDownException if it encounters {@link ShutDownException}
ambry-tools/src/main/java/com/github/ambry/tools/perf/serverperf/PutLoadProducerConsumer.java