Apache Druid

`LIKE` on our largest clusters is like a foot-bazooka that can blow up at anytime.

do you allow user send sql directly to druid? or you can do some filter before sending to druid?

We can probably add custom logic on Superset side.

In the recesses of my mind I seem to have a memory of <@U0344FW86DD> a few years back saying something about LIKE %x and LIKE x% having different performance characteristics… but I have drunk a lot of coffee since then :P

yeah `LIKE 'x%'` reads a range of the index without checking individual values, and `LIKE '%x'` or `LIKE '%x%'` reads the entire index and checks the pattern against every distinct value

so on high-cardinality columns, `LIKE 'x%'` is substantially faster than `LIKE '%x'` or `LIKE '%x%'`

on low-card columns they're all about the same

ok, that’s similar to MySQL, however none of these use-cases are using the bitmap index, right?

I think that the like operation is ran on the dictionary encoded (that's why it's faster/safe to stop scanning then the like is on the start) and then, the logic to use or not the bitmap index *may* vary depending on the number of results, if any
if the number of results is too high, druid may spending too much time during bitmaps AND's and a seq-scan could be better in some cases (but also, which struct could be good to check the existence of the matched values? another map lookup or a b-tree with the values IDs? )

but I don't know if the druid even do any different matching execution depending on the number of matches

although on high-card columns that still involves, for non-prefix `LIKE`, applying the `LIKE` pattern to each distinct value in the dictionary (so we know which bitmaps to pull)

it can still take some time for those high-card columns, even though we're using an index

I had no idea you can ship request logs like this. We did it the hard way by writing custom tailer lololol

• Adding a link for future ref: <https://druid.apache.org/docs/latest/operations/request-logging.html>

I was able to fix the `301: Moved Permanently` error.

However, I am still getting this error `Event too large to emit`  continuously.

```2022-11-11T08:43:43,806 ERROR [DruidSchema-Cache-0] org.apache.druid.java.util.emitter.core.HttpPostEmitter - Event too large to emit (1,559,214 &gt; 1,047,552): {"feed":"druidRequestLogFeed","timestamp":"2022-11-11T08:43:41.200Z","service":"druid-blue/broker","remoteAddr":"","sql":null,"queryStats":{"query/time":2586,"query/bytes":-1,"success":true,"identity":"allowAll"},"query":{"queryType":"segmentMetadata","dataSource":{"type":"table","name":"hello"},"intervals":{"type":"segments","segments":[{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":"2022-01-29T15:43:23.133Z","part":0},{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":"2022-01-29T15:43:23.133Z","part":1},{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":"2022-01-29T15:43:23.133Z","part":2},{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":"2022-01-29T15:43:23.133Z","part":3},{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":"2022-01-29T15:43:23.133Z","part":4},{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":"2022-01-29T15:43:23.133Z","part":5},{"itvl":"2022-01-26T00:00:00.000Z/2022-01-26T01:00:00.000Z","ver":" ...```
Any pointers how to solve this issue?

Seeing error because of this - <https://github.com/apache/druid/blob/druid-0.18.1/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java#L237>

Based on the codebase, it looks like you can control this by setting maxBatchSize. See: `class BaseHttpEmittingConfig`

```druid.emitter.http.maxBatchSize=2095104```
Tried to increase this setting to 2 MB. However, this didn't solve the `Event too large to emit` problem

I think you supposed to lower it so that it fits

<@U03149ALHN3>
```2022-11-11T08:43:43,806 ERROR [DruidSchema-Cache-0] org.apache.druid.java.util.emitter.core.HttpPostEmitter - Event too large to emit (1,559,214 &gt; 1,047,552):```
Here, Isn't Druid sending events of size  1,559,214 bytes but max permitted is only 1,047,552. Didn't understand why lower would solve the problem?

Yes. It seems like <https://github.com/apache/druid/blob/3e172d44abef43a25409260b51a797a1cef5f958/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java#L64|MAX_EVENT_SIZE is fixed to 1MB> making it a hard limit for a single event regardless of the maxBatchSize setting.

Then in that case, is there any way to overcome such `Event too large to emit` issues?

I've been thinking about that, I'm not sure.

Sounds like you have to do it the hard way by writing custom tailer.

In the past I successfully used this project for non trivial sized Cassandra. Too bad the project is dead. <https://github.com/Stratio/cassandra-lucene-index|https://github.com/Stratio/cassandra-lucene-index>

I seem to remember someone did hook up to Lucene for GIS ---- not heard of anyone else… one moment!

Aha
<https://www.druidforum.org/tag/lucene>

another note— you can do poor mans' full text search today with a procedure like this:

1) tokenize your free form text field during ingest: for example convert `"Here is a sentence."` to `["here", "sentence"]` (i.e. remove punctuation, split on whitespace, remove common words like `is` and `a`, ingest as array of strings). if you load this into a column named `msg_tokens` then druid will individually index each token.

2) when searching, tokenize your search string the same way, and use selector filters on `msg_tokens` (or `MV_CONTAINS` in SQL). for example: to search for `"Here is a sentence"`, use `WHERE MV_CONTAINS(msg_tokens, 'here') AND MV_CONTAINS(msg_tokens, 'sentence')`. this is super fast, since we can leverage the index for the individual tokens.

that being said, proper full text search using something designed for that specifically would be rad

do you think I can perform no.1 using the embedded JS? perform tokenization and create a new column based on the result?

i'm reluctant to encourage anyone to enable JS for any purpose due to the security concerns. let's say you don't care about that, though. then the answer is i'm not sure since i don't remember exactly what JS is capable of :slightly_smiling_face:. in particular i don't recall if JS transformation fns can generate arrays, or if they can only generate strings/numbers

so… maybe?

i think you could _almost_ do it with the builtin expressions: <https://druid.apache.org/docs/latest/misc/math-expr.html>

`string_to_array` splits the string, although i think you might really want split-on-regex to really properly tokenize (remove all whitespace and punctuation)

once it's an array, you can remove stop words using `map` and `replace`

or, you could do the transformation upstream using Lucene's tokenizer <https://lucene.apache.org/core/7_3_1/core/org/apache/lucene/analysis/standard/StandardTokenizer.html>

speaking of lucene :slightly_smiling_face:

the join itself will be streamed, but the collection of subqueries A, B, C will not (they will be fully collected in memory before the join starts)

:book:  Quick ad that there’s a new free course on Druid logging now available in the Druid Faculty :slightly_smiling_face: (kudos to <@U030L4P6UDA>!)
<https://learn.imply.io/apache-druid-logs>

When we had our major outage, likely due to `LIKE` queries. I random sampled N Historical pods and run jstack on them.

All of the processing threads are stuck and they all stuck the same way.

Either a groupBy:
```"groupBy_JoinDataSource{left=problematic-table, right=InlineDataSource{signature={__time:LONG, d0:STRING, a0:LONG}}, rightPrefix='j0.', condition=(substring("problematic_column", 0, 3) == "j0.d0"), joinType=INNER, leftFilter=null}_[2022-09-26T00:00:00.000Z/2022-09-27T00:00:00.000Z]" #502 daemon prio=5 os_prio=0 cpu=684108891.95ms elapsed=1966818.51s tid=0x00007f17a0115c60 nid=612 runnable  [0x00007e9b2e327000]
   java.lang.Thread.State: RUNNABLE
    at java.lang.StringLatin1.regionMatchesCI(java.base@18.0.2.1/StringLatin1.java:394)
    at java.lang.String.regionMatches(java.base@18.0.2.1/String.java:2226)
    at java.lang.String.equalsIgnoreCase(java.base@18.0.2.1/String.java:1962)
    at java.lang.Boolean.parseBoolean(java.base@18.0.2.1/Boolean.java:149)
    at org.apache.druid.math.expr.Evals.asBoolean(Evals.java:72)
    at org.apache.druid.math.expr.ExprEval$StringExprEval.asBoolean(ExprEval.java:932)
    at org.apache.druid.segment.filter.ExpressionFilter$2.matches(ExpressionFilter.java:208)
    at org.apache.druid.segment.filter.NotFilter$1.matches(NotFilter.java:71)
    at org.apache.druid.segment.filter.AndFilter$1.matches(AndFilter.java:205)
    at org.apache.druid.segment.join.PostJoinCursor.advanceToMatch(PostJoinCursor.java:71)
    at org.apache.druid.segment.join.PostJoinCursor.advanceUninterruptibly(PostJoinCursor.java:100)
    at org.apache.druid.segment.join.PostJoinCursor.advance(PostJoinCursor.java:92)```
or a topN:
```"topN_problematic-table_[2022-06-18T00:00:00.000Z/2022-06-19T00:00:00.000Z]" #505 daemon prio=5 os_prio=0 cpu=24162945.76ms elapsed=1966818.48s tid=0x00007f17a008c420 nid=615 runnable  [0x00007e9b2e023000]
   java.lang.Thread.State: RUNNABLE
   at java.util.HashMap.getNode(java.base@18.0.2.1/HashMap.java:577)
   at java.util.HashMap.get(java.base@18.0.2.1/HashMap.java:556)
   at org.apache.druid.math.expr.InputBindings$6.get(InputBindings.java:144)
   at org.apache.druid.math.expr.IdentifierExpr.eval(IdentifierExpr.java:131)
   at org.apache.druid.math.expr.BinaryBooleanOpExprBase.eval(BinaryOperatorExpr.java:173)
   at org.apache.druid.math.expr.BinOrExpr.eval(BinaryLogicalOperatorExpr.java:399)
   at org.apache.druid.math.expr.BinOrExpr.eval(BinaryLogicalOperatorExpr.java:397)
   at org.apache.druid.math.expr.BinOrExpr.eval(BinaryLogicalOperatorExpr.java:397)```

Also, I saw about 7-8 GC threads running with high CPU time at the bottom of jstack. Not sure if that information is useful or not.

So first question, is there a way to know what the topN is actually doing? I cannot tell which Boolean operation it does.

Second question, are the threads supposed to have such high cpu time? That number doesn’t look right.
Does druid reap the processing thread after it is done working?

Third question is related to my previous question.
It doesn’t look like Historical clean its own processing threads even when Broker already timed out the query.

In theory if there are no bugs, even if there’s a brutal `LIKE` query running, our 5 minutes Broker timeout should have killed the query and the query should have been killed at Historical level as well, right?

The easiest way to replicate this behavior is by using Druid’s full text search on a huge table.
You will see that all of Historical processing threads would hang.

This happened on JVM 18 G1 GC, which already have superior performance compared to older G1 GC versions.

the time range of our `LIKE` queries are usually between 1 month to 1 quarter.

Inside `DruidProcessingModule` , I don’t see that `PrioritizedExecutorService` inside `getProcessingExecutorPool` can be configured to be cancellable after X timeout.

Should `PrioritizedExecutorService` implement `invokeAll` so that any `Runnable` task can be killed after X timeout (and then have a global setting for timeout)?

I see the `awaitTermination` method, but I don’t see where is it being called. But anyway, it looks like that’s for shutting down the entire pool.

I see that this is how the pool is actually being used: `queryProcessingPool.submitRunnerTask`

I see, it relied on `QueryTimeoutException` to stop.
and one way to set the timeout is via the context:
```                  QueryContexts.hasTimeout(query) ?
                      future.get(QueryContexts.getTimeout(query), TimeUnit.MILLISECONDS) :
                      future.get()```

Then… this begs a question, if the Historical JVM is too busy collecting so much garbage (i.e. when it loads so many rows), will `QueryTimeoutException` actually be raised?

Regarding the question:
&gt; So first question, is there a way to know what the topN is actually doing? I cannot tell which Boolean operation it does.
Can you please share the topN query that is being run please. Also, I am not super familiar with jstack, however for the topN’s trace, should there be more lines (tracing back to Thread.run())?

it’s a bit hard to find the exact topN query that matches the trace in jstack.

hard to say exactly what's going on here from the thread dumps, but it doesn't seem to me to be LIKE related

the first dump you posted is some kind of expression filter after a join. could be anything; definitely seeing the query itself would be helpful

do you have request logging enabled? you might be able to find it that way <https://druid.apache.org/docs/latest/operations/request-logging.html>

a little pitch for Imply :wink: one of our products, Clarity, is all about gathering these kind of details. the workflow there is:

1) we show you all the query IDs you ran, sorted by total amount of CPU time used

2) use this to track down the query IDs that used the most CPU time during the problematic time frame

3) check your request logs for those specific query IDs; take some action

the second thread dump, btw, is also an expression of some sort; again unfortunately hard to tell what it is or where it's getting called in the topN logic. some more of the stack trace would illuminate where it's getting called in the topN logic, but likely won't tell us what the expression actually is

In the few random experiments I performed on our Druid setup, however, ICONTAINS_STRING performed about 4-5 times better for a result set that's was limited to 100k records with the `LIMIT` clause.

wow, good to know. i better update my information :slightly_smiling_face:

this sounds like you historical capacity is too small

Yes - this sounds like some sort of resource contention. It's hard to say without looking at the cluster what the resource is, but task heap or task slots are likely culprits.

you won't have parse exceptions due to resource constraints — that's likely really some issue in your data or your ingestion command

you can have rows _appear_ to drop due to resource constraints though; like didip said, this can happen if historicals are 'full'. you can see this in the Services tab of the web console. if you are in this situation, the apparently-dropped rows will come back if resources are increased

hello , this definitely happened, if I reduce the number of workers on the middle manager or manually ingest the individual task one by one from console, I don’t have any issues. I will try to reproduce it and share the logs.

ok - that will be helpful, appreciate it!

Could you share your ingestion spec and the failed task log?

My ingestion spec is here:
```{
  "type": "kinesis",
  "spec": {
    "dataSchema": {
      "dataSource": "data_source",
      "metricsSpec": [
        {
          "name": "ex_count",
          "type": "count"
        }
        
      ],
      "granularitySpec": {
        "segmentGranularity": "hour",
        "queryGranularity": "minute",
        "rollup": true,
        "type": "uniform"
      },
      "dimensionsSpec": {
        "dimensions": [
          "ex.dimension"
      },
      "timestampSpec": {
        "column": "at",
        "format": "posix"
      }
    },
    "ioConfig": {
      "stream": "&lt;kinesis-stream-name&gt;",
      "deaggregate": true,
      "useEarliestSequenceNumber": true,
      "inputFormat": {
        "type": "json",
        "flattenSpec": {
          "useFieldDiscovery": true,
          "fields": [
            
            {
              "expr": "$.message.ex_field",
              "name": "ex_field",
              "type": "path"
            }
          ]
        }
      },
      "endpoint": "kinesis.&lt;kinesis-region&gt;.<http://amazonaws.com|amazonaws.com>",
      "taskCount":"&lt;taskCount&gt;",
      "replicas":"&lt;replicas&gt;",
      "lateMessageRejectionStartDateTime": "2022-11-10T08:15Z",
      "type": "kinesis"
    },
    "tuningConfig": {
      "resetOffsetAutomatically": false,
      "skipSequenceNumberAvailabilityCheck":true,
      "type": "kinesis",
      "reportParseExceptions": false,
      "logParseExceptions": true,
      "intermediatePersistPeriod": "PT10M",
      "maxRowsInMemory": 75000
    }
  }
}```

Unfortunately we have had trouble getting the task logs before. When we go to the taks logs we only get garbage heap info like:
```  [GC Worker End (ms): Min: 9088.7, Avg: 9088.7, Max: 9088.7, Diff: 0.0]
   [Code Root Fixup: 0.0 ms]
   [Code Root Purge: 0.0 ms]
   [Clear CT: 0.1 ms]
   [Other: 0.4 ms]
      [Choose CSet: 0.0 ms]
      [Ref Proc: 0.1 ms] ...```

Not sure about why the task log is giving you that. But a few items regarding your ingestion spec:
• I don't see a `taskDuration`, probably a good idea to set this to 1 hour, or maybe 2. This means that each task will stop after that time  to output segments and another task will be started to continue reading from the same shards.
• I have not seen `lateMessageRejectionStartDateTime` before, usually it is done with `lateMessageRejectionPeriod` which you set to a duration like `PT1H` such that messages arriving with a timestamp that is more than an hour before the task is created are rejected. Is this what you mean to do? 

Thanks <@U030K4UM3H7> I did try `lateMessageRejectionPeriod` and it worked but I was worried that it would cut off some data, and thought that `lateMessageRejectionStartDateTime` would cut off data intially, but not afterwards.

However, i’m guessing that we probably should expect much data to arrive over  full hour late, so `lateMessageRejectionPeriod` should usually be acceptable?

It seems like the first job that fails, fails with overlord logs like:
```2022-11-14T08:06:30,766 INFO [org.apache.druid.metadata.SqlSegmentsMetadataManager-Exec--0] org.apache.druid.metadata.SqlSegmentsMetadataManager - Polled and found 28,486 segments in the database
2022-11-14T08:06:36,375 INFO [qtp1329043305-79] org.apache.druid.indexing.overlord.TaskLockbox - Added task[index_kinesis_data_source_6a6dc7bae0c9a5b_fkbelhbe] to TaskLock[TimeChunkLock{type=EXCLUSIVE, groupId='index_kinesis_data_source', dataSource='data_source', interval=2022-11-11T21:00:00.000Z/2022-11-11T22:00:00.000Z, version='2022-11-14T08:06:36.263Z', priority=75, revoked=false}]
2022-11-14T08:06:36,375 INFO [qtp1329043305-79] org.apache.druid.indexing.overlord.MetadataTaskStorage - Adding lock on interval[2022-11-11T21:00:00.000Z/2022-11-11T22:00:00.000Z] version[2022-11-14T08:06:36.263Z] for task: index_kinesis_data_source_6a6dc7bae0c9a5b_fkbelhbe
2022-11-14T08:06:36,386 ERROR [qtp1329043305-79] org.apache.druid.indexing.common.actions.SegmentAllocateAction - Could not allocate pending segment for rowInterval[2022-11-11T21:49:00.000Z/2022-11-11T21:50:00.000Z], segmentInterval[2022-11-11T21:00:00.000Z/2022-11-11T22:00:00.000Z].
2022-11-14T08:06:37,047 INFO [NodeRoleWatcher[PEON]] org.apache.druid.curator.discovery.CuratorDruidNodeDiscoveryProvider$NodeRoleWatcher - Node[<http://stage-druid-middle-manager-1:8100>] of role[peon] went offline.
2022-11-14T08:06:37,049 INFO [ServerInventoryView-0] org.apache.druid.client.BatchServerInventoryView - Server Disappeared[DruidServerMetadata{name='stage-druid-middle-manager-1:8100', hostAndPort='stage-druid-middle-manager-1:8100', hostAndTlsPort='null', maxSize=0, tier='_default_tier', type=indexer-executor, priority=0}]
2022-11-14T08:06:38,271 INFO [Curator-PathChildrenCache-1] org.apache.druid.indexing.overlord.RemoteTaskRunner - Worker[stage-druid-middle-manager-1:8091] wrote FAILED status for task [index_kinesis_data_source_6a6dc7bae0c9a5b_fkbelhbe] on [TaskLocation{host='stage-druid-middle-manager-1', port=8100, tlsPort=-1}]
2022-11-14T08:06:38,271 INFO [Curator-PathChildrenCache-1] org.apache.druid.indexing.overlord.RemoteTaskRunner - Worker[stage-druid-middle-manager-1:8091] completed task[index_kinesis_data_source_6a6dc7bae0c9a5b_fkbelhbe] with status[FAILED]```

My best explanation of this log is that
1. A late arriving task gets picked up.
2. The late arriving task can not get properly allocated
3. The task not getting properly allocated kills the running task
4. Next tasks get failed as well

The problem with late data is that it will create another segment for the corresponding time interval, which will likely already have been submitted to deep storage which just means you end up creating many small segments. In order to avoid that, you setup a threshold `lateMessageRejectionPeriod` which can be whatever you want, 1 hour or 6 hours or more, it's up to you.
I think the taskDuration is more important so that segments are output to deep storage at least once an hour and a new task starts building new segments at that point. I'm still not sure what the source your failure is though. Each task will request a segment allocation from the overlord when it sees a timestamp that doesn't fall into the intervals for the segments that it is currently building. Long running tasks would tend to have many more segments in progress, so controlling the task duration might help.
It is always a good practice to follow up a streaming ingestion with a compaction or auto-compaction job to consolidate all those potentially small segments that created by the parallel tasks and/or late arriving messages.

these "Could not allocate segment for row" errors can occur if you have mixed granularities in the same datasource… does this seem likely in your situation?

we've recently adjusted the logic here to be more robust, as well; you can try updating to 24.0 if you are not already on it. if you _are_ on it then i hope we figure out what is going on in your situation, since i'd like to fix whatever it is!

<@U0344FW86DD> we have a compaction task which I think might changes granularity, but that would be after 10 days. The segment here on 11/11 is only 3 days older than the one thats trying to run on 11/4 and should be using the same granularity.

In any case though, adding `lateMessageRejectionPeriod`
seems to make the tasks run smoothly. The data is also looking good.

We are on *0.18.1* so the recent fixes would not apply.

<@U04BB1B9MPA> since the `lateMessageRejectionPeriod` seems to solve it, Is it likely that you were getting timestamps that were older than 10 days which were trying to allocate a segment in one granularity that overlapped with one of the compacted granularity segments?

<@U030K4UM3H7> I don’t think that is the case, because we had `lateMessageRejectionStartDateTime` applied which was set to only 4 days back from the error (11/10). Also the segment that seems to be giving trouble is from 11/11:
```Could not allocate pending segment for rowInterval[2022-11-11T21:49:00.000Z/2022-11-11T21:50:00.000Z], segmentInterval[2022-11-11T21:00:00.000Z/2022-11-11T22:00:00.000Z]```
]

What granularity are the compacted segments on?

I'll give this a shot. I'm not an expert. Seems to me like the most likely explanation is that this is a network issue. Are you able to use curl successfully from the same server where you are running this?

Curl always works with sending request to HTTP end point. But for some reason when druid is posting the metrics to HTTP end point, it fails with `org.eclipse.jetty.io.EofException: Early EOF` exception.

Please note that it doesn't happen for all requests. For some requests, I am able to process them, and for some I get the exception.

I feel like druid is closing the stream prematurely before it can be read by HTTP end point

`druid.emitter.http.minHttpTimeoutMillis=100` Setting this property minimized the number of `org.eclipse.jetty.io.EofException: Early EOF` exception.

Looks like druid is closing the connection way too early than needed.

Also I am not sure this is 100% sure good solution. I am still seeing few of them.

I am still seeing few of ehse exceptions.

If it is timeout related it is likely happening here: <https://github.com/apache/druid/blob/8296123d895db7d06bc4517db5e767afb7862b83/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java#L664|https://github.com/apache/druid/blob/8296123d895db7d06bc4517db5e767afb7862b83/core[…]va/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java>

`lastBatchFillTimeMillis` is set based on `druid.emitter.http.minHttpTimeoutMillis` .
Perhaps your HTTP server is processing slowly? Have you tried a much larger setting, like 500 or 1000?

Reverting from 22.08-iap to previous version of Druid (22.04-iap) fixed the issue.

Speaking to others more knowledgable than me, they suggest this could be a permission issue with kafka/Confluent Kafka and that is likely solved by defining the consumer group in the<https://druid.apache.org/docs/24.0.0/development/extensions-core/kafka-supervisor-reference.html#more-on-consumerproperties| consumer properties >and/or ensuring consumer group has the correct ACLs

The default query timeout is set at `druid.server.http.defaultQueryTimeout=300000` (in millis = 5 minutes) which can be set on historicals, brokers and MMs.
You can also set it in the <https://druid.apache.org/docs/24.0.0/querying/query-context.html|query context> with `timeout` .
These values are governed by `druid.server.http.maxQueryTimeout`.
There is also `druid.broker.http.readTimeout=PT15M` which defines the timeout for reading results from a historical into the broker.
What settings do you have on these? Your timeout is occurring in 1 minute, right?

Thanks for your response, <@U030K4UM3H7>!
Correct our timeout occurs after a minute, as we set it in the query because our SLA for these types of queries is under 1 min.

On broker,
```druid.broker.http.readTimeOut=PT15M```
On Historical,
```druid.server.http.maxIdleTime=PT5M```

We do not have `druid.server.http.maxQueryTimeout`  on Broker or Historical.

Okay, so your question is why is it taking so long?

What is the query? Perhaps we can help optimize how it is resolving that type of query.

One common reason is that group by queries on large cardinality dimensions bottleneck in the merge step on the broker since this step is executed in a single thread. Depending on the nature of the query, there are techniques that help, like <https://druid.apache.org/docs/latest/ingestion/rollup.html|rollup> and <https://druid.apache.org/docs/latest/querying/sql-functions.html#approx_count_distinct|approximations>.

Correct, the question is, why are only some queries taking time and others are passing.. I will try to get a gist link for the query, but these type of queries have been running for years together… I think your point of large cardinality makes sense. I know for sure that the cardinality of my dimensions has increased recently. Let me check what can be done.
Thanks a lot for your pointers, I will keep the thread updated on the progress

Hi <@U03EXS199PD>. Do you mean you are attempting to use zookeeper as the metadata database for Druid? I don't think that is a supported configuration. Zookeeper is used in Druid for interprocess communication and for leader election, but I've never seen it used as a Metadata database component. For that you can use postgres, mysql. Or am I misunderstanding?

<@U030K4UM3H7> yeah it is for interprocess communication onIy... what metadata i  mean is here the segments information we store right that is why i just mentioned as metadata

I see. Okay, can you provide more context? How are you deploying it? Is this a fresh install? Does this occur on startup or when does it occur?

<@U030K4UM3H7> It was working for months now and all of sudden we started seeing this issue today... where in our zookeeper went into bad state and we lost quorum... This is the error that zookeeper is failing with
```Unable to load database on disk
java.io.IOException: Unreasonable length = 8522996```

we tried to restore from the backup it was functioning for 2 hours and then it started the issue again

<https://issues.apache.org/jira/browse/ZOOKEEPER-3593>

we are trying to understand is above issue was due to above jira

I am not sure what this bug is but I would try to use zookeeper as less as possible.
For that use http based load queues and http task runner. I think that would elevate the pressure on zookeeper.

Just to add to <@U0315MQKEHK>’s point. Those <https://druid.apache.org/docs/24.0.0/configuration/index.html#segment-management|settings> are:
`druid.serverview.type=http`
`druid.coordinator.loadqueuepeon.type=http`
`druid.indexer.runner.type=httpRemote`

We are getting the above issue and zookeeper is not coming up. Here Zookeeper is running as a Kubernetes service.

Is there anyway to bring up zookeeper  by resolving the below issue:
```2022-11-16 20:18:54,446 [myid:3] - INFO  [main:ZKAuditProvider@42] - ZooKeeper audit is disabled.
2022-11-16 20:18:54,606 [myid:3] - ERROR [main:QuorumPeer@1148] - Unable to load database on disk
java.io.IOException: Unreasonable length = 8523180```


By any chance, is your ingestion spec huge?

We had this same issue before. The huge payload is bumping against ZK’s default limit.

Do you deploy ZK with Bitnami Zookeeper Helm chart?

Step 1: Enlarge the limit on ZK itself:
```jvmFlags: "-Djute.maxbuffer=16777216"```
Step 2: Enlarge the limit on Druid side by setting the same thing on every JVM options:
```-Djute.maxbuffer=16777216```
Obviously you have to restart everything for the settings to take place.
Step 3: Problem solved.

<@U03149ALHN3> Should we keep the max jute buffer size for zookeeper and druid same ?
If it is same will that not effect zookeeper anything?

it’s translated to :
```{
    "queryType": "scan",
    "dataSource": {
        "type": "table",
        "name": "sp_campaigns_realtime_aggregation"
    },
    "intervals": {
        "type": "intervals",
        "intervals": [
            "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
        ]
    },
    "resultFormat": "compactedList",
    "columns": [
        "__time",
        "accountId"
       
    ],
    "legacy": false,
    "context": {
        "sqlOuterLimit": 1001,
        "sqlQueryId": "e4a54e89-fb46-42d7-ab05-ce14b0efb60d"
    },
    "granularity": {
        "type": "all"
    }
},signature={__time:LONG, accountType:LONG,
})```

the fix to query is trivial:
```select __time, accountId 
from sp_campaigns_realtime_aggregation 
where 
__time = TIMESTAMP '2022-11-16 14:00:00'```

Similar query with `&gt;` sign works normally
```select __time, accountId 
from sp_campaigns_realtime_aggregation 
where 
__time &gt; '2022-11-16 14:00:00'```

I looked into this for a bit, and don't have a solution yet, but made some progress on figuring out the cause

Raised a github issue with findings: <https://github.com/apache/druid/issues/13400>

What does the log inside MM say? It supposed to say “successfully deleted <hdfs://xyz>” file

Hi <@U03149ALHN3>, thank you for your response. The above lines are extract from the MM log itself. I do not see a line stating it is deleting some files from HDFS

I dont have access to my YARN right now, will check and update the thread

It looks like it hasn't even tried to delete anything

Is it possible your datasource does not have any unused segments in the interval you provided?

Hi <@U0344FW86DD>, correct, it is not deleting at all.

I verified, my datasource has a retention of `P3D`, and I see underlying segments in HDFS deep storage from 2021.

the interval in the job `2020-01-01T00:00:00.000Z/2020-12-31T00:00:00.000Z` does not include 2021 -- if you did not do a different job including 2021 as well, perhaps this is why

Sorry, my bad,  I should have specified that clearly.

I had two separate jobs for 2020 and 2021; and in underlying HDFS I see both 2020 and 2021

in that case i am not sure what happened! in the most recent log for the 2021 task, do you see any attempts to delete files?

You should really debug this inside your YARN. YARN will say if the applicationId that supposed to delete is working or not.

Thanks for your pointers, <@U03149ALHN3> and <@U0344FW86DD>!
I did some investigation from the metadata MySQL table as well, and here are my two observations -
1. I had to specify a smaller time range for clearing the underlying HDFS segments. Since the number of segments in the HDFS directory was too large, even when running `hdfs dfs -ls` , I had to specify extra memory to get the output using `export _JAVA_OPTIONS="-Xmx8g` . Looks like when Druid’s Kill process was also trying to access the HDFS location, it was running into memory issues. When I specified smaller windows, the kill task was able to delete the HDFS segments. 
2. The kill process did not delete all the segments in my specified interval, because in the MySQL `druid_tasklocks` table, I could still see the listing of these segments. Some segments dated back to 2020, and 2021. (My datasources’ retention is set to P3D). 

is this kill job manual? The automatic kill job supposed to kill 1 day only

Correct, <@U03149ALHN3>, this kill job was manually triggered.

i see i see.
You will run into same problem with compaction as well. In general, you have to be careful with really large time range

We automate kill with our airflow job by submitting 1 day at a time to avoid this issue.

Got it, do we have any recommendations on the time range we should use?

We usually do things 1 day at a time, even though the actual time range is large.

it’s best to automate that using something like airflow because it requires some logic: chunking, retry, backoff, etc. We do this for backfill, kill, and compaction.

makes sense.. Long time back we also had cron scripts that would issue kill tasks, but somehow they got lost in transition from one infra to the other. Thanks again for all your pointers! They really helped!

On another note, <@U03149ALHN3>, <@U0344FW86DD> - Where do we specify feature requests? Please let me know if this would make sense -

• From the Druid console, when we click on `Delete unused segments (issue kill task)`  for a datasource, it starts  `api_issued_kill*`  task, and the interval for this task is selected from year 1000 to 3000. 
• But for the data source for which I started this task, has continuous real time ingestion, and batch ingestion. So there is always a lock on some time range between the years 1000 and 3000. As a consequence of which, the kill task stays in `PENDING` state. 
• So, the feature request here being, can we specify the interval when we want to issue the kill task from the console? Thinking about the recent conversation in this thread, we also have to be mindful of the time ranges. 


What specifically do you mean by "multi region"?

Setup 2 Druid clusters but they are both backed by the same S3 and PG.
And then I can use DNS/LB trickery to swing traffic between the two.

I am investigating absolutely zero downtime for Druid cluster at the moment…

If Cluster A’s Historical and MM use tierA and Cluster B’s Historical and MM use tierB…
I think it ~cloud~ work.
LMAO, could work.

well, if you have they connected to the same ZK, they will be just one cluster still

I think ZK would be non-issue, even if you have a dedicated Zookeeper per Kubernetes cluster.

if you have separate ZK, each will think that they have a coordinator leader and may corrupt the database, but as long you only set up a historical, it's fine for a fast- recovery until you kill the other ZK/coordenators

because internal traffic (brokered by ZK) never escapes to outside.

and Historical segment files don’t contain cluster information, which is a great design! :+1:

With the kubernetes extensions instead of ZK you can control which cluster owns the leader via kubernetes APIS.

I think the main issue is to not ever have 2 coordenators as laeder running

having a second 'dr' cluster with just historicals could indeed work, I think, if they use the same ZK and have a different tier in order to not cause each query routed on the 'active' cluster goes across a potential expensive network round-trip

and for the MM in the 'dr' cluster, connected to the same ZK, and add the task replica &gt; 1 to have it HA but I don't remember if there's tiers for MM

but I think that's useless bc if the database or ZK fails on the active region, the dr MM would not be able to commit and serve queries when the task finsihes

MM has category, which can be abused in similar manner :stuck_out_tongue:
Even better, it can be updated dynamically from the Overlord Dynamic Config :smile:

MM is less of a concern for me because the MM cluster can be down on the DR cluster.
Historical is much more important because data sync takes super long time.

It's not that crazy :slightly_smiling_face:

I've seen people do something similar for blue/green deploys: launch a whole new tier and then terminate the old one

Just keep in mind that the clusters will be linked in some ways and so you still have a chance of correlated failure (for example if the shared metadata store or shared deep storage has problems)

I am running 0.23 with some local bug fixes to the k8s extension.  We also automatically cycle out the coordinator pods every 3 hours to ensure leadership is fresh.  There is definitely another bug I have to track down (time allowing) that stabilizes it so that I can submit the fix upstream.

Cycling the pods works around the stability issue.

The other fixes I am using are pretty small.

One thing I noticed is that the logs quickly fill up with errors, but the error is actually more like a debug statement using error level, so it looks like things are very broken, but in fact everything is fine.

I'm going to discuss a lot of this in my druid summit talk

as far as I can remeber, it was a null point exception when downloading some metadata from the k8s

I suffered the same NPE like <@U038XPQFJHM> last time I tried it. But to be honest, I didn’t try very hard because it looked like it’s not ready.

I am pretty sure I fixed that NPE in my internal fork.  As soon as I am happy with the stability I'll PR it upstream.

Following that I'll upgrade to the lates kubernetes client also.

i haven’t tested it extensively, but in the limited amount i did it was a bit smaller but also slower to read (at least in the ‘hot’ use case where segments are not reading from disk)

its available in druid if you’d like to experiment, <https://github.com/apache/druid/blob/master/docs/ingestion/ingestion-spec.md#indexspec>

Damn it, did not see it and as a result it is not exposed in the UI (you can always just type it in!).

<@U030C8H59T8> could you please review this PR: <https://github.com/apache/druid/pull/13388> it is trying to true up the UI to the docs

one task cannot read from two different topics. You could merge the topics using kstreams and read them using one task in druid and use filters while querying. Depending on the kafka producer you are using you may be able to send the different streams to different partitions in kafka and read them using fewer tasks in druid

they would use more space, because they're buffering temporary stuff

the amount of extra space used should be about the same order of magnitude as the amount of data being ingest

note that for native batch (`index_parallel` or `compact`) you can reduce this usage by setting `druid.processing.intermediaryData.storage.type = deepstore`

in MSQ (SQL-based ingest) we are adding something similar soon

[we already have half of it implemented: using deep storage for stage outputs. but we're adding deep storage for sorting too and then we'll fully release the feature]

This is the error on the server:
```2022-11-18T14:48:38,836 ERROR [qtp857732012-81] org.apache.druid.indexing.overlord.sampler.SamplerExceptionMapper - Failed to sample data: Failed to get Avro schema: 79
org.apache.druid.indexing.overlord.sampler.SamplerException: Failed to sample data: Failed to get Avro schema: 79
	at org.apache.druid.indexing.overlord.sampler.InputSourceSampler.sample(InputSourceSampler.java:246) ~[druid-indexing-service-24.0.0.jar:24.0.0]
	at org.apache.druid.indexing.seekablestream.SeekableStreamSamplerSpec.sample(SeekableStreamSamplerSpec.java:117) ~[druid-indexing-service-24.0.0.jar:24.0.0]
	at <http://org.apache.druid.indexing.overlord.sampler.SamplerResource.post|org.apache.druid.indexing.overlord.sampler.SamplerResource.post>(SamplerResource.java:43) ~[druid-indexing-service-24.0.0.jar:24.0.0]
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_275]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_275]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_275]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_275]
	at com.sun.jersey.spi.container.JavaMethodInvokerFactory$1.invoke(JavaMethodInvokerFactory.java:60) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.model.method.dispatch.AbstractResourceMethodDispatchProvider$TypeOutInvoker._dispatch(AbstractResourceMethodDispatchProvider.java:185) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.model.method.dispatch.ResourceJavaMethodDispatcher.dispatch(ResourceJavaMethodDispatcher.java:75) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.uri.rules.HttpMethodRule.accept(HttpMethodRule.java:302) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.uri.rules.ResourceClassRule.accept(ResourceClassRule.java:108) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.uri.rules.RightHandPathRule.accept(RightHandPathRule.java:147) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.uri.rules.RootResourceClassesRule.accept(RootResourceClassesRule.java:84) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1542) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.application.WebApplicationImpl._handleRequest(WebApplicationImpl.java:1473) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1419) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.server.impl.application.WebApplicationImpl.handleRequest(WebApplicationImpl.java:1409) ~[jersey-server-1.19.4.jar:1.19.4]
	at com.sun.jersey.spi.container.servlet.WebComponent.service(WebComponent.java:409) ~[jersey-servlet-1.19.4.jar:1.19.4]```

It means there is some problem contacting your schema registry, like an IO error of some sort

Unfortunately the actual IO exception is not attached to the parse error you get! This patch fixes it so you would get the original error: <https://github.com/apache/druid/pull/13403>

I would suggest double checking your `avroBytesDecoder` config and verifying that your registry actually does have a valid schema for id `79` in it

If that doesn't lead you to a solution, then that original error would be very useful, so you could consider applying the patch and building a version of Druid with that

Alternatively you can try running the Confluent schema registry client yourself, with the same config, and see what error you get when trying to retrieve schema id `79`. It is likely that you would get the same error from the schema registry client as Druid is encountering.

You can try to get the schema using SchemaRegistry REST API, like:
```curl -k -X GET <https://schemaregistry.organization.com:8081/schemas/ids/79>```
and you can add the basic auth. too to curl if the SchemaRegistry requires authentication.

we see index_kafka tasks that succeded in log, i.e.
```2022-11-21T13:08:34,252 INFO [task-runner-0-priority-0] org.apache.druid.indexing.worker.executor.ExecutorLifecycle - Task completed with status: {
  "id" : "index_kafka_sp_campaigns_realtime_aggregation_eef1b68744abc48_jfncjiei",
  "status" : "SUCCESS",
  "duration" : 1798848,
  "errorMsg" : null,
  "location" : {
    "host" : null,
    "port" : -1,
    "tlsPort" : -1
  }
}```
however task status if FAILED
with errorMsg:
  “errorMsg”: “No task in the corresponding pending completion taskGroup[4] succeeded before completion timeout ela...”

our completionTimeout was 30mins(increased it to 60 mins)
but there is gap in data which we would like to fill

For a successful task even though the handoff failed, the segment would've been published to deep storage and would be eligible for loading through the run rules coordinator duty.
could you check if there is a gap in your data loaded in druid? It would be done via UI in datasources tab or using API (<https://druid.apache.org/docs/latest/operations/api-reference.html#segment-loading-by-datasource>)

image.png

Hi <@U031A2GTWEM>
Are you talking about following ?
“Availability detail” under data sources

Hi Michael! yes, there's availability % as well in that tab

In our case we didn’t have any availability issues (or at least no indication of this) .
Just number of ingestion tasks failed because of `completionTimeout,`  which <@U047P93LEUT> mentioned, still not sure why it may take so much time.

oh I see, I thought there was availability concern as well due to this message :
&gt;  but there is gap in data which we would like to fill
I think I misunderstood that

Regarding more investigation on the long time for handoff, you can probably check the runtime of `RunRules` coordinator duty around that time (it is emitted as the metric `coordinator/time`) and see if it was high as well.
If there are lot of segments being handed-off around the same, we've seen issues around their completion. Multiple fixes are being done in the upcoming release to reduce handoff time.

thanks <@U031A2GTWEM> for the info, we will check coordinator/time metric

we see high times for tasks
I’ve found in coordinator logs some locking with kill tasks that we schedule to remove unused segments
```2022-11-21T11:54:15,763 INFO [TaskQueue-Manager] org.apache.druid.indexing.overlord.TaskLockbox - Cannot create a new taskLockPosse for request[TimeChunkLockRequest{lockType=EXCLUSIVE, groupId='kill_sp_campaigns_realtime_aggregation_after_dedup_pjjhnijp_2022-11-14T00:00:00.000Z_2022-11-21T00:00:00.000Z_2022-11-21T03:00:01.515Z', dataSource='sp_campaigns_realtime_aggregation_after_dedup', interval=2022-11-14T00:00:00.000Z/2022-11-21T00:00:00.000Z, preferredVersion='null', priority=0, revoked=false}] because existing locks[[TaskLockPosse{taskLock=TimeChunkLock{type=EXCLUSIVE, groupId='index_kafka_sp_campaigns_realtime_aggregation_after_dedup', dataSource='sp_campaigns_realtime_aggregation_after_dedup', interval=2022-11-20T23:00:00.000Z/2022-11-21T00:00:00.000Z, version='2022-11-20T23:00:00.599Z', priority=75, revoked=false}, taskIds=[index_kafka_sp_campaigns_realtime_aggregation_after_dedup_2899dc64f6e52cd_cpfinpbf, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_5cccc69339458cf_mhoddmpl, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_a173f06e2e4bc07_hpbcbmki, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_3bcd3eee475f061_lndfdjfb, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_4e232d8b7824f50_nejlbefp, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_bc56833a08847ae_imcnnool, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_6f07318c19267f8_oieefadb, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_7bf9902c552596f_hcnhlbbk, index_kafka_sp_campaigns_realtime_aggregation_after_dedup_57cc76904e3f9a9_kdedccld]}]] have same or higher priorities```


We have cron that submits kill tasks to cleanup data periodically, so we most probably caused this “deadlock”
by setting end of interval to be within retention window. I’m going to change this cron so it will submit kill tasks so that interval wont intersect with retained window

Thanks for the logs - they show that the kill task wasn't able to get the lock on the interval `2022-11-14T00:00:00.000Z/2022-11-21T00:00:00.000Z` since kafka ingestion was writing to that interval.
It doesn't indicate any deadlock, just that the system has decided not to provide write lock to two tasks simultaneously. Also, druid always prefers ingestion tasks over kill tasks for locks.
If you wish to avoid these conflicts proactively, then as you mentioned you can ensure that the kill interval is not overlapping the ingestion one

it's a bug of some sort!

a few of these were fixed recently, so if you are not on the latest release (24.0) I would suggest upgrading

if you see this even on 24.0 please let us know; some more logs around this (especially if they show the historical exiting) would be helpful in debugging

There is a metric `druid_jvm_mem_used` that has a property `memKind`  which has possible values `'heap'` and `'nonheap'` and I _believe_ using `nonheap` gets you the off heap usage but I can't concretely confirm that.

Spot on.  That metric (jvm/mem/used) has  dimension "memkind".  This screenshot from Clarity, Imply's APM tool, shows simple analysis you can do with that data

I've seen that metric but I don't believe it includes the offHeap memory usage. That metric exists for all node types, but stays at around 190MB even on my indexer tasks where offHeap memory is used during kafka ingest. The amount of nonheap memory does not increase as expected during ingest/merging.

After the restart, are the supervisors in Druid still running? Are they healthy? You can see this in the Druid Console in the Ingestion View.

```{
  "dataSource": "",
  "stream": "",
  "partitions": 1,
  "replicas": 1,
  "durationSeconds": 3600,
  "activeTasks": [
    {
      "id": "",
      "startingOffsets": {
        "0": 11305
      },
      "startTime": "2022-11-22T16:14:00.921Z",
      "remainingSeconds": 1304,
      "type": "ACTIVE",
      "currentOffsets": {
        "0": 11305
      },
      "lag": {
        "0": -9753
      }
    }
  ],
  "publishingTasks": [],
  "latestOffsets": {
    "0": 1552
  },
  "minimumLag": {
    "0": -9753
  },
  "aggregateLag": 0,
  "offsetsLastUpdated": "2022-11-22T16:52:06.402Z",
  "suspended": false,
  "healthy": true,
  "state": "RUNNING",
  "detailedState": "RUNNING",
  "recentErrors": []
}```

Do you see the real-time segments for the datasource in the Segments view?
The broker depends on the announcement of segments to recognize the data that is available. Normally (or by default) this happens through Zookeeper. Could there be something wrong with zookeeper after the restart?
Another thought is that you can change this to use the HTTP communication instead by setting the `druid.serverview.type=http` instead of `batch` for<https://druid.apache.org/docs/24.0.0/configuration/index.html#segment-discovery| segment discovery> in the Broker.

I've noticed that when the system restarts, kafka resets the offset. While, druid keeps the old offset.

Oh, I see. You can use the "hard reset" from the supervisor's Actions menu  to reset the offsets.  <https://druid.apache.org/docs/24.0.0/operations/web-console.html#supervisors-and-tasks|These are the relevant docs>.

you can set the `size` parameter during ingestion to control size/accuracy

the table on this page has info about memory requirements <https://datasketches.apache.org/docs/Theta/ThetaSize.html>

we just haven’t added a jdk17 docker yet, everything should work fine if using to build