code: @Override
      public void execute(Tuple input)
      {
          try
          {
              super.writeTuple(input, this.tupleMapper);
          }
          catch (Exception ex)
          {
              String messageFormat = "Tuple write failed. InputTuple={0}";
              String errorMessage = MessageFormat.format(messageFormat, input.toString());
              logger.warn(errorMessage, ex);
          }
  
          getCollector().ack(input);
      }
docs: {@inheritDoc}