code: @Override
  	public void onAdd(Response response) {
  		ensureInitialized();
  
  		KeyValuePair keyValuePair = new KeyValuePair(KafkaConstants.RESPONSE_TYPE, gson.toJson(response));
  		String message = gson.toJson(keyValuePair);
  		
  		try {
  			kafkaSender.send(buildTopicNames(response), message);
  		} catch (InterruptedException | ExecutionException e) {
  			logger.error("Failed to send add response message to output topic.", e);
  		}
  	}
docs: This implementation of onAdd watches for responses, and immediately sends them out to the
  appropriate exchange/queue for routing to the necessary client application(s)