docs: @Override
      public void onEvent(MessageEvent event) throws Exception {
          incomingMessages.mark();
  
          final Message msg = event.getMessage();
          if (msg == null) {
              LOG.debug("Skipping null message.");
              return;
          }
          LOG.debug("Processing message <{}> from OutputBuffer.", msg.getId());
  
          final Set<MessageOutput> messageOutputs = outputRouter.getStreamOutputsForMessage(msg);
          msg.recordCounter(serverStatus, "matched-outputs", messageOutputs.size());
  
          final Future<?> defaultOutputCompletion = processMessage(msg, defaultMessageOutput);
  
          final CountDownLatch streamOutputsDoneSignal = new CountDownLatch(messageOutputs.size());
          for (final MessageOutput output : messageOutputs) {
              processMessage(msg, output, streamOutputsDoneSignal);
          }
  
          // Wait until all writer threads for stream outputs have finished or timeout is reached.
          if (!streamOutputsDoneSignal.await(configuration.getOutputModuleTimeout(), TimeUnit.MILLISECONDS)) {
              LOG.warn("Timeout reached. Not waiting any longer for stream output writer threads to complete.");
          }
  
          // now block until the default output has finished. most batching outputs will already been done because their
          // fast path is really fast (usually an insert into a queue), but the slow flush path might block for a long time
          // this exerts the back pressure to the system
          if (defaultOutputCompletion != null) {
              Uninterruptibles.getUninterruptibly(defaultOutputCompletion);
          } else {
              LOG.error("The default output future was null, this is a bug!");
          }
  
          if (msg.hasRecordings()) {
              LOG.debug("Message event trace: {}", msg.recordingsAsString());
          }
  
          outputThroughput.inc();
  
          LOG.debug("Wrote message <{}> to all outputs. Finished handling.", msg.getId());
  
          event.clearMessages();
      }