code: @Override
      public Tuple getNext() throws IOException
      {
          try {
              if (reader == null || !reader.nextKeyValue()) {
                  return null;
              }
  
              final Object value = reader.getCurrentValue();
  
              if (value instanceof SmileEnvelopeEvent) {
                  final SmileEnvelopeEvent envelope = (SmileEnvelopeEvent) value;
                  final JsonNode data = (JsonNode) envelope.getData();
  
                  final Tuple tuple = factory.newTuple(data.size());
                  int i = 0;
                  for (final GoodwillSchemaField field : schema.getSchema()) {
                      final JsonNode node = data.get(field.getName());
                      tuple.set(i, getJsonValue(field.getType(), node));
                      i++;
                  }
  
                  return tuple;
              }
              else {
                  throw new IOException(String.format("Expected SmileEnvelopeEvent, not %s", value.getClass()));
              }
          }
          catch (NullPointerException e) {
              String splitInfo = "<no split info>";
              if(split != null) {
                 splitInfo = split.toString();
              }
              log.error(String.format("Corrupt Smile file (%s), ignoring the rest of the input",splitInfo), e);
          }
          catch (com.fasterxml.jackson.core.JsonParseException e) {
              String splitInfo = "<no split info>";
              if(split != null) {
                 splitInfo = split.toString();
              }
              log.error(String.format("Corrupt Smile file (%s), ignoring the rest of the input",splitInfo), e);
              return null;
          }
          catch (InterruptedException e) {
              Thread.currentThread().interrupt();
          }
  
          return null;
      }
docs: Retrieves the next tuple to be processed. Implementations should NOT reuse
  tuple objects (or inner member objects) they return across calls and
  should return a different tuple object in each call.
  
  @return the next tuple to be processed or null if there are no more tuples
  to be processed.
  @throws java.io.IOException if there is an exception while retrieving the next
  tuple