[jira] [Created] (FLUME-2769) Elastic Search Sink throws exceptions on messages that look like JSON

["Adam Gent (JIRA)" <ji...@apache.org>]

Adam Gent created FLUME-2769:
--------------------------------

             Summary: Elastic Search Sink throws exceptions on messages that look like JSON
                 Key: FLUME-2769
                 URL: https://issues.apache.org/jira/browse/FLUME-2769
             Project: Flume
          Issue Type: Bug
          Components: Sinks+Sources
    Affects Versions: 1.6
            Reporter: Adam Gent


We have a message body that looks like JSON but is not and the elastic content type guesser gets confused:

{code}
24 Aug 2015 11:15:58,693 ERROR [SinkRunner-PollingRunner-DefaultSinkProcessor] (org.apache.flume.sink.elasticsearch.ElasticSearchSink.process:225)  - Failed to commit transaction. Transaction rolled back.
org.elasticsearch.common.jackson.core.JsonParseException: Unrecognized token 'Flush': was expecting ('true', 'false' or 'null')
 at [Source: [B@3ce85c92; line: 1, column: 7]
        at org.elasticsearch.common.jackson.core.JsonParser._constructError(JsonParser.java:1487)
        at org.elasticsearch.common.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:518)
        at org.elasticsearch.common.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3323)
        at org.elasticsearch.common.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2482)
        at org.elasticsearch.common.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:801)
        at org.elasticsearch.common.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:697)
        at org.elasticsearch.common.xcontent.json.JsonXContentParser.nextToken(JsonXContentParser.java:51)
        at org.apache.flume.sink.elasticsearch.ContentBuilderUtil.addComplexField(ContentBuilderUtil.java:60)
        at org.apache.flume.sink.elasticsearch.ContentBuilderUtil.appendField(ContentBuilderUtil.java:47)
        at org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer.appendBody(ElasticSearchLogStashEventSerializer.java:87)
        at org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer.getContentBuilder(ElasticSearchLogStashEventSerializer.java:79)
        at org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer.getContentBuilder(ElasticSearchLogStashEventSerializer.java:73)
        at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.addEvent(ElasticSearchTransportClient.java:164)
        at org.apache.flume.sink.elasticsearch.ElasticSearchSink.process(ElasticSearchSink.java:189)
        at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
        at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
        at java.lang.Thread.run(Thread.java:745)
{code}


It appears that this was also a previous regression:

{code:java}
  public static void addComplexField(XContentBuilder builder, String fieldName,
      XContentType contentType, byte[] data) throws IOException {
    XContentParser parser =
      XContentFactory.xContent(contentType).createParser(data);
    parser.nextToken();
    // Add the field name, but not the value.
    builder.field(fieldName);
    try {
      // This will add the whole parsed content as the value of the field.
      builder.copyCurrentStructure(parser);
    } catch (JsonParseException ex) {
      // If we get an exception here the most likely cause is nested JSON that
      // can't be figured out in the body. At this point just push it through
      // as is, we have already added the field so don't do it again
      builder.endObject();
      builder.field(fieldName, new String(data, charset));
    } finally {
      if (parser != null) {
        parser.close();
      }
    }
  }
{code}

Notice how the {{try}} statement is not around the {{nextToken}} which is now throwing the exception.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)