Occasional "Null timestamp in input" during realtime ingestion

[gaodayue]

We use druid 0.10.0 and tranquility-kafka to do realtime ingestion. Occasionally, we see NPE in realtime task log, and the error message looks fairly strange.

io.druid.java.util.common.parsers.ParseException: Unparseable timestamp found!
        at io.druid.data.input.impl.MapInputRowParser.parse(MapInputRowParser.java:75) ~[druid-api-0.10.0.jar:0.10.0]
        at io.druid.segment.realtime.firehose.EventReceiverFirehoseFactory$EventReceiverFirehose.addAll(EventReceiverFirehoseFactory.java:192) ~[druid-server-0.10.0.jar:0.10.0]
Caused by: java.lang.NullPointerException: Null timestamp in input: {timestamp=1505203571000, reach=0, passImpression=1, passReach=0, frequencyFilter=0, click=0, impres...
        at io.druid.data.input.impl.MapInputRowParser.parse(MapInputRowParser.java:67) ~[druid-api-0.10.0.jar:0.10.0]
        ... 54 more

From the error message, it's obvious that the timestamp column "timestamp" does exist and has a valid value, but Druid says "Null timestamp in input" anyway.

[gaodayue]

Related code in TimestampSpec.java

// remember last value parsed
private ParseCtx parseCtx = new ParseCtx();

public DateTime extractTimestamp(Map<String, Object> input)
{
  return parseDateTime(input.get(timestampColumn));
}

public DateTime parseDateTime(Object input)
{
  DateTime extracted = null;
  if (input != null) {
    if (input.equals(parseCtx.lastTimeObject)) {
      extracted = parseCtx.lastDateTime;
    } else {
      ParseCtx newCtx = new ParseCtx();
      newCtx.lastTimeObject = input;
      extracted = timestampConverter.apply(input);
      newCtx.lastDateTime = extracted;
      parseCtx = newCtx;
    }
  }
  return extracted;
}

In this case, input of parseDateTime can't be null (timestamp column exists!), and timestampConverter.apply never returns null. So it looks like parseDateTime can't return null but it did happen occasionally.

[gaodayue]

It seems that TimestampSpec is used inside "/push-events" endpoint of EventReceiverFirehose, so there can be concurrent usage of it. But the implementation is not thread-safe, there is a race condition of ParseCtx.

[gianm]

Fixed by #4791.