compact task throws NullPointerException

[pdeva]

2018-04-09 20:05:03,829 ERROR i.d.i.o.ThreadPoolTaskRunner [task-runner-0-priority-0] Exception while running task[CompactionTask{id=compact_pctile-hour_2018-04-09T20:04:50.734Z, type=compact, dataSource=pctile-hour}]
java.lang.NullPointerException
	at io.druid.indexing.common.task.CompactionTask.lambda$createDataSchema$1(CompactionTask.java:299) ~[druid-indexing-service-0.12.0.jar:0.12.0]
	at java.util.stream.MatchOps$1MatchSink.accept(MatchOps.java:90) ~[?:1.8.0_161]
	at java.util.ArrayList$ArrayListSpliterator.tryAdvance(ArrayList.java:1359) ~[?:1.8.0_161]
	at java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:126) ~[?:1.8.0_161]
	at java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:498) ~[?:1.8.0_161]
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:485) ~[?:1.8.0_161]
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[?:1.8.0_161]
	at java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:230) ~[?:1.8.0_161]
	at java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:196) ~[?:1.8.0_161]
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_161]
	at java.util.stream.ReferencePipeline.allMatch(ReferencePipeline.java:454) ~[?:1.8.0_161]
	at io.druid.indexing.common.task.CompactionTask.createDataSchema(CompactionTask.java:299) ~[druid-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.common.task.CompactionTask.createIngestionSchema(CompactionTask.java:232) ~[druid-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.common.task.CompactionTask.run(CompactionTask.java:184) ~[druid-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:444) [druid-indexing-service-0.12.0.jar:0.12.0]
	at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:416) [druid-indexing-service-0.12.0.jar:0.12.0]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_161]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_161]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_161]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_161]

State of segments:

Task specification:

{
  "type": "compact",
  "dataSource": "pctile-hour",
  "interval": "2018-04-08T00:00:00.000Z/2018-04-09T00:00:00.000Z"
}

Payload:

{
  "task": "compact_pctile-hour_2018-04-09T20:04:50.734Z",
  "payload": {
    "id": "compact_pctile-hour_2018-04-09T20:04:50.734Z",
    "resource": {
      "availabilityGroup": "compact_pctile-hour_2018-04-09T20:04:50.734Z",
      "requiredCapacity": 1
    },
    "dataSource": "pctile-hour",
    "interval": "2018-04-08T00:00:00.000Z/2018-04-09T00:00:00.000Z",
    "segments": null,
    "tuningConfig": null,
    "context": null,
    "groupId": "compact_pctile-hour_2018-04-09T20:04:50.734Z",
    "dimensionsSpec": null
  }
}

Druid 0.12.0

[gianm]

I think the NPE here is caused by the fact that not all segments have rollup set (it's a relatively newer feature) but the compaction task calls index.getMetadata().getRollup() and casts it to a boolean primitive.

In addition to this issue, the compaction task should not assume that index.getMetadata() exists at all, since it won't for segments created by older versions.

It should degrade as gracefully as it can in these situations. Some questions:

• Should the task exit with SUCCESS or FAILED if it cannot figure out how to compact the segments?
• How will the auto compacter handle this?

/cc @jihoonson

[pdeva]

the part i dont understand is why this segment is not compactable. i have another datasource with query granularity set to minute and it compacts fine.

I am pasting the supervisor spec for this (pctile-hour) datasource:

{
  "type": "kafka",
  "dataSchema": {
    "dataSource" : "pctile-hour",
    "parser": {
      "type": "string",
      "parseSpec": {
        "format": "json",
        "timestampSpec": {
          "column": "timestamp",
          "format": "auto"
        },
        "dimensionsSpec": {
          "dimensions": ["accountid", "appid", "jvmid", "metrictype", "name"]
        }
      }
    },
    "metricsSpec": [
      {
        "name": "count",
        "type": "count"
      },
      {
        "name": "value",
        "fieldName": "value",
        "type": "quantilesDoublesSketch",
        "k": 512
      }
    ],
    "granularitySpec": {
      "type": "uniform",
      "segmentGranularity": "DAY",
      "queryGranularity": "HOUR"
    }
  },
  "tuningConfig": {
    "type": "kafka",
    "maxRowsPerSegment": 5000000
  },
  "ioConfig": {
    "topic": "dupercentile",
    "consumerProperties": {
      "bootstrap.servers": "xxx"
    },
    "taskCount": 1,
    "replicas": 2,
    "taskDuration": "PT24H"
  }
}

Note that i had updated our quantiles sketch extension manually with the bugfix update, so that is not relevant here.

Is there anything that needs to be modified in the spec above to allow druid to compact it without issues?

[jihoonson]

@pdeva thank you for the report! I'm making a patch for this bug.

Druid stores metadata for each segment. The metadata contains some meta information of a segment like timestamp spec, query granularity, a flag to indicate the segment is rolled up, and so on. This information is stored in a separate meta.smoosh file when a segment is stored.

The problem is that most fields of metadata can be null. null in metadata means that Druid can't determine that information. For example, let me suppose we have two segments to merge into a single segment. If one of them was rolled up (isRollup = true) but another wasn't (isRollup = false), isRollup is set to null for the new merged segment because we can't decide it.

compactionTask uses metadata to get the information about aggregators and isRollup of input segments. So, you can avoid this issue by using indexTask instead of compactionTask which requires to specify the whole dataSchema in the task spec.

• Should the task exit with SUCCESS or FAILED if it cannot figure out how to compact the segments?
• How will the auto compacter handle this?

@gianm good point. I think the task complete state should be FAILED to let users know the compaction failed.

This kind of bug will make the compactor to try to compact the same interval forever. Currently, since this will occupy only one task slot, remaining intervals will be compacted if users have enough task slots for compaction. However, we need something to handle this kind of infinite compaction failures in a nicer way. Maintaining a blacklist for failed intervals might be a solution. Users can reset the blacklist once they fix their problems. What do you think?

[pdeva]

@jihoonson

if one of them was rolled up (isRollup = true) but another wasn't (isRollup = false)

my question is, given my above supervisor spec, under what condition would isRollup be set to false or null?
from the looks of the supervisor spec, its fairly clear about rolling every segment to day level granularity.

[jihoonson]

@pdeva are you running only kafka index tasks for the same dataSource? I think you might be better to debug what changed it. You can check which segment has a null or false isRollup in metadata, and once you find such a segment, you can check which task or operation changed it from logs.

[pdeva]

@jihoonson yes only kafka indexing was used. nothing else.

i got the metadata.

{
  "queryType":"segmentMetadata",
  "dataSource":"pctile-hour",
  "intervals":["2018-04-08T00:00:00.000Z/2018-04-09T00:00:00.000Z"]
}

result:

[
    {
        "id": "pctile-hour_2018-04-08T00:00:00.000Z_2018-04-09T00:00:00.000Z_2018-04-07T23:59:01.257Z",
        "intervals": [
            "2018-04-08T00:00:00.000Z/2018-04-09T00:00:00.000Z"
        ],
        "columns": {
            "__time": {
                "type": "LONG",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": null,
                "minValue": null,
                "maxValue": null,
                "errorMessage": null
            },
            "accountid": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 44,
                "minValue": "52c7e28eebb26a86fb1505b5",
                "maxValue": "5ac81949e4b03fa758538302",
                "errorMessage": null
            },
            "appid": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 328,
                "minValue": "53c3f5dd0cf23a3c0bb94a8f",
                "maxValue": "5ac81ed3547efa8194ae0570",
                "errorMessage": null
            },
            "count": {
                "type": "LONG",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": null,
                "minValue": null,
                "maxValue": null,
                "errorMessage": null
            },
            "jvmid": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 471,
                "minValue": "54ca377db3690af75ba4cf8e",
                "maxValue": "5aca1ee8547efa8194ae0dc0",
                "errorMessage": null
            },
            "metrictype": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 1,
                "minValue": "TXN",
                "maxValue": "TXN",
                "errorMessage": null
            },
            "name": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 4743,
                "minValue": "/ (GET)",
                "maxValue": "/{product: ide|rm|ci}/componentDetails/list (GET)",
                "errorMessage": null
            },
            "value": {
                "type": "quantilesDoublesSketch",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": null,
                "minValue": null,
                "maxValue": null,
                "errorMessage": null
            }
        },
        "size": 0,
        "numRows": 64336,
        "aggregators": null,
        "timestampSpec": null,
        "queryGranularity": null,
        "rollup": null
    },
    {
        "id": "pctile-hour_2018-04-08T00:00:00.000Z_2018-04-09T00:00:00.000Z_2018-04-07T23:59:01.257Z_1",
        "intervals": [
            "2018-04-08T00:00:00.000Z/2018-04-09T00:00:00.000Z"
        ],
        "columns": {
            "__time": {
                "type": "LONG",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": null,
                "minValue": null,
                "maxValue": null,
                "errorMessage": null
            },
            "accountid": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 41,
                "minValue": "52c7e28eebb26a86fb1505b5",
                "maxValue": "5ac81949e4b03fa758538302",
                "errorMessage": null
            },
            "appid": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 317,
                "minValue": "53c3f5dd0cf23a3c0bb94a8f",
                "maxValue": "5ac81ed3547efa8194ae0570",
                "errorMessage": null
            },
            "count": {
                "type": "LONG",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": null,
                "minValue": null,
                "maxValue": null,
                "errorMessage": null
            },
            "jvmid": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 447,
                "minValue": "54ca377db3690af75ba4cf8e",
                "maxValue": "5acaabca547efa8194ae0e77",
                "errorMessage": null
            },
            "metrictype": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 1,
                "minValue": "TXN",
                "maxValue": "TXN",
                "errorMessage": null
            },
            "name": {
                "type": "STRING",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": 3930,
                "minValue": "/ (GET)",
                "maxValue": "/{product: ide|rm|ci}/componentDetails/list (GET)",
                "errorMessage": null
            },
            "value": {
                "type": "quantilesDoublesSketch",
                "hasMultipleValues": false,
                "size": 0,
                "cardinality": null,
                "minValue": null,
                "maxValue": null,
                "errorMessage": null
            }
        },
        "size": 0,
        "numRows": 45479,
        "aggregators": null,
        "timestampSpec": null,
        "queryGranularity": null,
        "rollup": null
    }
]

seems both shards have rollup set to null

[pdeva]

investigating further:

• all segments created by Kafka Indexing Service (KIS) have rollup set to null
• doing a compact on another data source (ingested by KIS) pctile-minute works fine, even though its segments have rollup: null too.
• the task that KIS creates does have rollup: true set, however the generated segments have rollup: null. Here is an example of the task submission from the logs

2018-04-09 14:28:16,476 INFO i.d.i.w.e.ExecutorLifecycle [main] Running with task: {
  "type" : "index_kafka",
  "id" : "index_kafka_pctile-hour_00e3999bb72c9c8_oakmcogn",
  "resource" : {
    "availabilityGroup" : "index_kafka_pctile-hour_00e3999bb72c9c8",
    "requiredCapacity" : 1
  },
  "dataSchema" : {
    "dataSource" : "pctile-hour",
    "parser" : {
      "type" : "string",
      "parseSpec" : {
        "format" : "json",
        "timestampSpec" : {
          "column" : "timestamp",
          "format" : "auto"
        },
        "dimensionsSpec" : {
          "dimensions" : [ "accountid", "appid", "jvmid", "metrictype", "name" ]
        }
      }
    },
    "metricsSpec" : [ {
      "type" : "count",
      "name" : "count"
    }, {
      "type" : "quantilesDoublesSketch",
      "name" : "value",
      "fieldName" : "value",
      "k" : 512
    } ],
    "granularitySpec" : {
      "type" : "uniform",
      "segmentGranularity" : "DAY",
      "queryGranularity" : "HOUR",
      "rollup" : true,
      "intervals" : null
    },
    "transformSpec" : {
      "filter" : null,
      "transforms" : [ ]
    }
  },
  "tuningConfig" : {
    "type" : "KafkaTuningConfig",
    "maxRowsInMemory" : 75000,
    "maxRowsPerSegment" : 5000000,
    "intermediatePersistPeriod" : "PT10M",
    "basePersistDirectory" : "/tmp/1522938445282-0",
    "maxPendingPersists" : 0,
    "indexSpec" : {
      "bitmap" : {
        "type" : "concise"
      },
      "dimensionCompression" : "lz4",
      "metricCompression" : "lz4",
      "longEncoding" : "longs"
    },
    "buildV9Directly" : true,
    "reportParseExceptions" : false,
    "handoffConditionTimeout" : 0,
    "resetOffsetAutomatically" : false,
    "segmentWriteOutMediumFactory" : null
  },
  "ioConfig" : {
    "type" : "kafka",
    "baseSequenceName" : "index_kafka_pctile-hour_00e3999bb72c9c8",
    "startPartitions" : {
      "topic" : "dupercentile",
      "partitionOffsetMap" : {
        "0" : 1215429707
      }
    },
    "endPartitions" : {
      "topic" : "dupercentile",
      "partitionOffsetMap" : {
        "0" : 9223372036854775807
      }
    },
    "consumerProperties" : {
      "bootstrap.servers" : "xx"
    },
    "useTransaction" : true,
    "pauseAfterRead" : false,
    "minimumMessageTime" : null,
    "maximumMessageTime" : null,
    "skipOffsetGaps" : false
  },
  "context" : {
    "checkpoints" : "{\"0\":{\"0\":1215429707}}",
    "IS_INCREMENTAL_HANDOFF_SUPPORTED" : true
  },
  "groupId" : "index_kafka_pctile-hour",
  "dataSource" : "pctile-hour"
}

i feel this deserves another issue for KIS

[drcrallen]

NOTE: the null comes from segment metadata query, the descriptor.json in deep storage might be a better indicator on what is in storage.

[jihoonson]

@drcrallen thanks. It looks that rollup is not stored at all in that file.

@pdeva thanks for digging this. I'll check further.

[pdeva]

@jihoonson @drcrallen please move this discussion to #5620
if #5613 is mergred, i can close this issue.