Apache Druid

first thing i would check is are the services starting up? like do the main containers have logs that indicate the services are running?

if so then it is probably a health check issue, perhaps to do with connectivity

if not then that's the first thing to tackle

Hi <@U0344FW86DD>. Thank you for the response. The issue with the pods were fixed by redeploying the chart.
Now, we are deploying Druid on a service mesh cluster, on gke. I see co-ordinator is failing.

Yes, taskCount = # of shards per Kafka / Kinesis as a best practice

hmm. even with 10 shards -&gt; 1 task, a GB heap is intended to be enough. it would just go slower than 10 tasks; the idea is you should be scaling up your tasks for perf reasons not memory reasons

do you know what is in the 10 GB (do you have any heap dumps, for example)?

could indicate something we need to improve

I don’t have any heap dumps, though I suspect these would be very useful. I’ll look into this into the future. Though just recently, I discovered that reducing the recordsPerFetch from default 4000 -&gt; 10 and recordBufferSize from 10000 -&gt; 100 allowed for a 1GB heap

We use KPL aggregation on our records and I’m lead to believe the internal buffer for records between the fetch threads and the indexer thread is causing the OOMs

ah, maybe that's it! those, i bet, are per-shard

oh, and you're also saying something else, that your messages are huge because of aggregation

By “those” you mean the internal record buffers?

Messages are max 1MB; they still abide to Kinesis shard limits

Though the average size of our messages are 500KB

are you using the `deaggregate` setting in your druid kinesis supervisor?

nvm, those settings are not per-shard. just checked. so i think you're right that it's deaggregation related

Yeah, my earlier read through the source-code revealed the recordBufferSize counts an aggregated record as 1 count, rather than counting the individual exploded records. Which makes it consistent with recordsPerFetch I suppose, but it was counter to my initial thoughts

raised a patch to set the defaults better, and clarify the docs: <https://github.com/apache/druid/pull/13539>

Last starting log : `2022-12-09T14:35:27,100 INFO [main] org.apache.druid.sql.calcite.schema.SegmentMetadataCache - SegmentMetadataCache initialized in [755,644] ms.`

and `druid.sql.planner.metadataSegmentCacheEnable` is set to False (default value)

But I don’t want broker to load segments, only segment metadata 

Broker only loads the segment metadata always.
I find it suspicious that broker is taking 12 mins just to initialise.

```|`druid.broker.segment.awaitInitializationOnStart`|Boolean|Whether the Broker will wait for its view of segments to fully initialize before starting up. If set to 'true', the Broker's HTTP server will not start up, and the Broker will not announce itself as available, until the server view is initialized. See also `druid.sql.planner.awaitInitializationOnStart`, a related setting.|true|```


You can set this property to false. The downside is that broker might give incorrect results while the whole metadata is loaded.

Do you see exceptions like
`Metadata refresh failed, trying again soon.` in the logs ?

Are other services also restarted at the same time ?

&gt;  Are other services also restarted at the same time ?

We would be logging the exception. Those exception stack traces would point to what is going on.

Here is what I found in `/var/log/syslog/` :
```Dec 12 15:40:12 druid-query run-druid[29103]: Exception in thread "BrokerServerView-5" java.util.concurrent.RejectedExecutionException: Task org.apache.druid.client.BrokerServerView$$Lambda$307/624230821@5f2810c9 rejected from java.util.concurrent.ThreadPoolExecutor@5717b890[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 44035]
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.util.concurrent.Executors$DelegatedExecutorService.execute(Executors.java:668)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.BrokerServerView.runTimelineCallbacks(BrokerServerView.java:397)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.BrokerServerView.serverAddedSegment(BrokerServerView.java:293)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.BrokerServerView.access$000(BrokerServerView.java:67)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.BrokerServerView$1.segmentAdded(BrokerServerView.java:149)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.FilteringSegmentCallback.segmentAdded(FilteringSegmentCallback.java:52)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.BatchServerInventoryView.lambda$addSingleInventory$2(BatchServerInventoryView.java:278)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at org.apache.druid.client.BatchServerInventoryView.lambda$runSegmentCallbacks$0(BatchServerInventoryView.java:239)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
Dec 12 15:40:12 druid-query run-druid[29103]: #011at java.lang.Thread.run(Thread.java:750)```


This might look like a concurrency issue. Would need to full broker log to confirm .
I would suggest creating a github ticket with the broker logs so we can look at it

You could increase it to 15 and see if there is any improvement

Alright thanks Karan, will create a Github issue soon

Weird that we never have this problem.
But by default we set http.numThreads = 500 everywhere.

Didip these are not the netty threads which you might set to such a big number

These are the threads which go to each historical and get the segments served by that historical so that the broker can plan stuff

Hello, after running some queries on a large period, my brokers crashed, but it seems that the query was still running on my historicals.
Once I saw that my brokers have crashed, I’ve restarted them and it took a long time. But I’ve noticed that once the CPU usage on historicals dropped, the brokers started. It looks like the fact that historicals were busy, brokers could not start

Hi. Do you see this with 24.0.x as well or are you on an older version of Druid?

<@U0319HD8HEC> I am on druid-0.18.1.
Would you know if there is a way to stop this from happening without upgrading?

Would there be a parameter to space out druid task creation for instance, so they all wouldn’t be synced up, creating the spikes when they are created

I ask this because more tasks seems to help speed up the ingestion, but I am not sure why.

I have tried increasing recordsPerFetch very high with a lower level tasks, but this does not seem to help.

For instance with these params, my stream did not lag
```"taskCount": "17",
"recordsPerFetch": 150,
"fetchDelayMillis": 6500```
While these params did lag
```"taskCount": "6",
"recordsPerFetch": 450,
"fetchDelayMillis": 6500```
despite there being more total throughput implied by my choice of taskCount, recordsPerFetch and fetchDelayMillis. I also was not receiving too many throughput errors in either case. Also we are not seeing any cpu/memory issues on the middle manager.

Could there be another parameter serving as a bottleneck for pulling data through a single task?

The latest event being ingested is falling minutes behind the current time

druid.ingest.events.messageGap &gt; 1 Minute and grows

Also i’m comparing this to the existing tranquility  ingestion, and the event count for a given hour is behind the tranquility ingestion

<@U0344FW86DD> when u get a chance will u be able to help

what kind of failures do you see….a timeout or some other error?

with unions the actual QPS at the historical will be more than 30 (30*number of queries in each union query). Can you paste your broker and historical run time properties here?

Hi,
I'm using druid batch ingestion using s3 for almost 100m records, but batch ingestion is running very slow.
Can you please help, what kind of performance setting we need to apply to make it more fast.
Thanks

around 61 files, each file is around 150MB

Hi <@U04EXJWJE10>,
Can you tell us a bit about the size of your cluster?
• Share your Middle Manager configuration. 
• How many middle managers do you have and what resources in terms of CPUs and Memory do they have available? 
• What is the worker.capacity set to 
• What are your settings for `druid.indexer.runner.javaOptsArray`? 
In addition to all that, and as an alternate path which is usually better, are you on Apache Druid 24.0.x ? One of the great new features is <https://druid.apache.org/docs/latest/multi-stage-query/index.html|SQL based batch ingestion>, where you can use a SQL INSERT or REPLACE statement to ingest external data and it has better performance than the traditional index_parallel batch ingestion. Rollup becomes a SQL aggregation query such as:

```INSERT INTO "kttm_rollup"

WITH kttm_data AS (
SELECT * FROM TABLE(
  EXTERN(
    '{"type":"http","uris":["<https://static.imply.io/example-data/kttm-v2/kttm-v2-2019-08-25.json.gz>"]}',
    '{"type":"json"}',
    '[{"name":"timestamp","type":"string"},{"name":"agent_category","type":"string"},{"name":"agent_type","type":"string"},{"name":"browser","type":"string"},{"name":"browser_version","type":"string"},{"name":"city","type":"string"},{"name":"continent","type":"string"},{"name":"country","type":"string"},{"name":"version","type":"string"},{"name":"event_type","type":"string"},{"name":"event_subtype","type":"string"},{"name":"loaded_image","type":"string"},{"name":"adblock_list","type":"string"},{"name":"forwarded_for","type":"string"},{"name":"language","type":"string"},{"name":"number","type":"long"},{"name":"os","type":"string"},{"name":"path","type":"string"},{"name":"platform","type":"string"},{"name":"referrer","type":"string"},{"name":"referrer_host","type":"string"},{"name":"region","type":"string"},{"name":"remote_address","type":"string"},{"name":"screen","type":"string"},{"name":"session","type":"string"},{"name":"session_length","type":"long"},{"name":"timezone","type":"string"},{"name":"timezone_offset","type":"long"},{"name":"window","type":"string"}]'
  )
))

SELECT
  FLOOR(TIME_PARSE("timestamp") TO MINUTE) AS __time,
  session,
  agent_category,
  agent_type,
  browser,
  browser_version,
  MV_TO_ARRAY("language") AS "language", -- Multi-value string dimension
  os,
  city,
  country,
  forwarded_for AS ip_address,

  COUNT(*) AS "cnt",
  SUM(session_length) AS session_length,
  APPROX_COUNT_DISTINCT_DS_HLL(event_type) AS unique_event_types
FROM kttm_data
WHERE os = 'iOS'
GROUP BY 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11
PARTITIONED BY HOUR
CLUSTERED BY browser, session```

The <https://druid.apache.org/docs/latest/multi-stage-query/reference.html#sql-reference|EXTERN table function >can access any batch source that traditional batch ingestion supports, so you can use `"type":"s3"` and in the input format type you can use  `"type":"parquet"`. This should be directly transferable from you current ioConfig and inputFormat settings in your current batch ingestion.

<@U04EXJWJE10> any luck in figuring out what was making it slow?

Hi <@U030K4UM3H7>,
Thanks for the suggestion, I need to try out the sql ingestion, I'll get back to you asap

<@U030K4UM3H7> <@U03BJAMQ4H5>
Middle Manager Conf

``` druid_node_type: 'middleManager'
  druid_worker_capacity: 10
  DRUID_XMX: 5G
  DRUID_XMS: 1G
  DRUID_MAXDIRECTMEMORYSIZE: 2g
  druid_processing_buffer_sizeBytes: '200000000'
  druid_processing_numMergeBuffers: 2
  druid_processing_numThreads: 2
  druid_server_http_numThreads: 250
  druid_indexer_runner_javaOptsArray: '["-server", "-Xms512m", "-Xmx1g", "-XX:MaxDirectMemorySize=2g", "-Duser.timezone=UTC", "-Dfile.encoding=UTF-8", "-XX:+ExitOnOutOfMemoryError", "-Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager"]'
  druid_indexer_fork_property_druid_processing_buffer_sizeBytes: '200000000'
  druid_indexer_fork_property_druid_processing_numMergeBuffers: 2
  druid_indexer_fork_property_druid_processing_numThreads: 2
  druid_indexer_fork_property_druid_server_http_numThreads: 70
  #new changes
  druid_indexer_task_baseDir: "/opt/druid/var/druid/task_baseDir"
  druid_indexer_task_baseTaskDir: "/opt/druid/var/druid/baseTaskDir"
  druid_indexer_task_restoreTasksOnRestart: true
  druid_processing_tmpDir: "/opt/druid/var/druid/tmpDir"
  druid_indexer_storage_type: "metadata"
  druid_realtime_cache_useCache: true
  druid_realtime_cache_populateCache: true
  druid_cache_type: caffeine
  #groupby
  druid_query_groupBy_maxMergingDictionarySize: "100000000"
  druid_query_groupBy_maxOnDiskStorage: "10000000000"```


Thanks <@U04EXJWJE10>. Just to confirm the fit, what resources in terms of CPU/Memory does each Middle Manager have?
A few items caught my attention about the config:
• the -xms and -xmx values are recommended to be set to the same value for both middle manager and peons (javaOptsArray).
• The middle manager process itself does not need a lot of memory, so you can probably lower that to 1G or less and increase the heap for the peons which will help during ingestion reducing the need to spill to disk.
• Think of splitting available memory among the peons, you have 10 peons configured - with 1G heap and 2G max direct memory. Take a look at the <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html#middlemanager|guidelines here> which would indicate that max direct memory with you settings, would be more like `(druid.processing.numThreads + druid.processing.numMergeBuffers + 1) * druid.processing.buffer.sizeBytes` for the forked process = 1G, and more heap will help, so you could invert them heap=2G and maxDirect = 1G 
• that implies a total of 3G * 10 peons = 30G - if you have more, add more to the heap.
• worker capacity set to 10 implies that you have 11 CPUs available to the MM. If you have less CPUs, reduce this number and redistribute the memory among the number of peons you end up with.
• If you go down the SQL-Based Ingestion path, there's a really good explanation of <https://druid.apache.org/docs/latest/multi-stage-query/concepts.html#memory-usage|how memory is used in the peons/workers here>.
Finally in a little shameless self promotion, I wrote this blog post yesterday speaking of using <https://sergioferragut.substack.com/p/rollup-ingestion-in-druid-using-sql|SQL to do Rollup ingestion, including approximations>. I hope you find it useful.

Hi <@U03D7RPAT0A>, Select query has been deprecated in favor of <https://druid.apache.org/docs/latest/querying/scan-query.html|Scan queries>. Pagination on scan queries can be done using `limit` and `offset` parameters. It does not have the `"fromnext":"true"` functionality that existed in the Select query type though.

I use very older version of druid and it does not support scan and we are backfilling data using select API call . Observation , we use while loop for select query with pagination specs + limit and exit from loop once we get empty result set but this approach is failing for one of the corner case with multiple partitions for a time interval and endless loop is running because don't get empty result set .

Can anyone suggest select query support pagination with multiple partitions ?

<@U03D7RPAT0A> In a previous thread you mentioned the installation you have is running 0.9.1. Is that correct? I ask because there looks to be enhancements to the select query with pagination based on a breakage in 0.9.1.x related to this <https://github.com/apache/druid/pull/2480|bug>...<https://groups.google.com/g/druid-user/c/M5i51SfDJL0|more information is here>

<https://github.com/apache/druid/pull/2480#issuecomment-235729318|There is a comment in the thread that offers a workaround that may help>.
`The work around is to take the pagingIdentifiers returned by Druid and strip them down to only the last (latest) segment. That is returned to the web app, which includes it in the request for the next page just like the current implementation. But when the service receives that request, it parses out the starting time of the segment and uses that as the start date on the interval for the query.` 

Let me know if that helps.

Thanks <@U03S0JXTM4N>. I can relate , this is a similar case as per above shared . i use druid version": "0.9.1.1" Seems this bug is not fixed in 0.9.1.1 version . Also tried to strip down to only the latest segment in my code but it does not work . threshold distribute the offset with multiple segments and it can be sum of multiple segments but offset length is not fix while distributing.

In my case -  {'test-ds_2021-10-13T00:00:00.000Z_2021-10-13T08:01:00.000Z_2022-02-07T05:09:25.413Z': 9388, 'test-ds_2021-10-13T00:00:00.000Z_2021-10-13T08:01:00.000Z_2022-02-07T05:09:25.413Z_1': 5610}

test-ds_2021-10-13T00:00:00.000Z_2021-10-13T08:01:00.000Z_2022-02-07T05:09:25.413Z
test-ds_2021-10-13T00:00:00.000Z_2021-10-13T08:01:00.000Z_2022-02-07T05:09:25.413Z_1

Hi,
Can we use spark or hive table to write data into druid?
Thanks

no. with spark you need to either post to kafka or write out a file and then ingest into druid. With hive you will ingest from hdfs

That is correct, they are marked as unused

Thank you Ian. I will dig to find out why they are not removed them DeepStorage then.
Have a good one :slightly_smiling_face:

it is a kill tasks job to remove unused segments from deep storage. You said `And when I issue a kill task to remove them from DeepStorage, nothing happens.` . there is something going wrong there. How are you issuing the kill task?

Hello Vadim, 
I’m issuing it from the console 

does this datasource have streaming ingestion?

I think you are running into this issue: <https://apachedruidworkspace.slack.com/archives/C030K0Z7S1Z/p1669222099825209>

it was fixed here: <https://github.com/apache/druid/pull/13431> and will be released in the upcoming 25.0 release

the issue is that the console make a kill task with interval `1000/3000` (so all time). Since that interval overlaps with your streaming ingestion, those tasks just sit there pending waiting for streaming ingestion to stop which never happens

right now the workaround is to issue a kill task yourself with an interval that does not include the future

image.png

from the `Submit JSON task` dialog in the Ingestion view

I’ve noticed this issue on the intervals. 

I’ve tried to delete on the interval <tel:2021-01-01/2022-01-01|2021-01-01/2022-01-01> but still no deep storage deletion 

strange, do you see that kill task get marked as SUCCESS in is status?

Could you please share the task log for that kill task? I would expect to see:

Not at work right now. Will do tomorrow first hour 

No problem. Would love to get to the bottom of this.

Capture d’écran 2022-12-16 à 11.09.48.png

Capture d’écran 2022-12-16 à 11.09.13.png

Hello Vadim,
I’ll try to remove segments from deesptorage using :
```{
  "type": "kill",
  "dataSource": "beam-realtime",
  "interval": "2022-09-14/2022-09-15"
}```

And here are the indexer logs I got :
```2022-12-16T10:15:20,216 INFO [WorkerTaskManager-NoticeHandler] org.apache.druid.indexing.worker.WorkerTaskManager - Task[kill_beam-realtime_emdmldij_2022-09-14T00:00:00.000Z_2022-09-15T00:00:00.000Z_2022-12-16T10:15:20.189Z] started.
2022-12-16T10:15:20,216 INFO [[kill_beam-realtime_emdmldij_2022-09-14T00:00:00.000Z_2022-09-15T00:00:00.000Z_2022-12-16T10:15:20.189Z]-threading-task-runner-executor-5] org.apache.druid.segment.loading.SegmentLocalCacheManager - Using storage location strategy: [LeastBytesUsedStorageLocationSelectorStrategy]
2022-12-16T10:15:20,269 INFO [threading-task-runner-executor-5] org.apache.druid.indexing.overlord.ThreadingTaskRunner - Removed task directory: /opt/druid-data/task/kill_beam-realtime_emdmldij_2022-09-14T00:00:00.000Z_2022-09-15T00:00:00.000Z_2022-12-16T10:15:20.189Z
2022-12-16T10:15:20,274 INFO [WorkerTaskManager-NoticeHandler] org.apache.druid.indexing.worker.WorkerTaskManager - Task [kill_beam-realtime_emdmldij_2022-09-14T00:00:00.000Z_2022-09-15T00:00:00.000Z_2022-12-16T10:15:20.189Z] completed with status [SUCCESS].```

I don’t know if that matters, but I’m using Indexers and not MM

Screenshot 2022-12-15 at 9.47.51 PM.png

I don't think t's possible to define the credentials in input source for azure. Do you have a possibility to set this up via properties: <https://druid.apache.org/docs/latest/development/extensions-core/azure.html>

^yeah i think only the s3 input source supports overriding credentials with a new set of keys/iam role

```druid.azure.account=....
druid.azure.container=....
druid.azure.sharedAccessStorageToken=....```


<@U037DQV43D1> but these credentials config is for setting deep storage right ? I want to use azure blob storage as input source (to read the files and ingest in druid) ?

Or does this mean same Azure storage account has to be only used both as input source and as deep storage as well ?

Yes, I think for now you should use the same storage account for both input source and deep storage

I think it's worth opening an issue on github for a closer feature parity between azure input source and AWS input source.I can see situations where there's overhead to move data between different storage accounts just to be able to ingest it into druid.

Sure <@U037DQV43D1> Will add the issue on github.

<@U043WQK8XDJ> You might have some credentials that are visible. Suggest you to scrub them off.

I am not sure how to do that, could you please let me know how/where to do the same

should i be adding like this for clarity?
```druid.extensions.loadList=["druid-hdfs-storage"]```
i couldn't find clarity emitter extension name in the doc here though- <https://druid.apache.org/docs//0.15.1-incubating/development/extensions.html#core-extensions>

I have the following:
```druid.monitoring.monitors=["org.apache.druid.java.util.metrics.JvmMonitor","org.apache.druid.java.util.metrics.JvmThreadsMonitor"]

druid.emitter.composing.emitters=["logging","clarity"]

druid.emitter=composing
druid.emitter=clarity```

you have to add that in
```druid.extensions.loadList=["clarity-emitter"]```

ok thanks, let me try this, will take a while for deployment process, will acknowledge back

BTW, if you are using Clarity, you should log a ticket with Imply Support as well

<@U04CGQ1P68Z> after deploying this change, still i am getting error in druid-historical node.  added this config - druid.extensions.loadList=["druid-s3-extensions", "mysql-metadata-storage", "druid-datasketches", "druid-prometheus-emitter", "clarity-emitter"]
```com.tubemogul.druid.query.zerofilledtopn.ZeroFilledTopNQueryModule -&gt; com.google.inject.multibindings.MapBinder$RealMapBinder)
  while locating java.util.Map&lt;java.lang.Class&lt;? extends org.apache.druid.query.Query&gt;, org.apache.druid.query.QueryToolChest&gt;
    for the 1st parameter of org.apache.druid.query.MapQueryToolChestWarehouse.&lt;init&gt;(MapQueryToolChestWarehouse.java:35)
  while locating org.apache.druid.query.MapQueryToolChestWarehouse
  while locating org.apache.druid.query.QueryToolChestWarehouse
    for the 1st parameter of org.apache.druid.server.QueryLifecycleFactory.&lt;init&gt;(QueryLifecycleFactory.java:52)
  at org.apache.druid.server.QueryLifecycleFactory.class(QueryLifecycleFactory.java:52)
  while locating org.apache.druid.server.QueryLifecycleFactory
    for the 1st parameter of org.apache.druid.server.QueryResource.&lt;init&gt;(QueryResource.java:113)
  at org.apache.druid.server.QueryResource.class(QueryResource.java:78)
  while locating org.apache.druid.server.QueryResource
  while locating org.apache.druid.server.metrics.QueryCountStatsProvider
    for the 1st parameter of org.apache.druid.server.metrics.QueryCountStatsMonitor.&lt;init&gt;(QueryCountStatsMonitor.java:40)
  while locating org.apache.druid.server.metrics.QueryCountStatsMonitor
  at org.apache.druid.server.metrics.MetricsModule.getMonitorScheduler(MetricsModule.java:92) (via modules: com.google.inject.util.Modules$OverrideModule -&gt; com.google.inject.util.Modules$OverrideModule -&gt; org.apache.druid.server.metrics.MetricsModule)
  at org.apache.druid.server.metrics.MetricsModule.getMonitorScheduler(MetricsModule.java:92) (via modules: com.google.inject.util.Modules$OverrideModule -&gt; com.google.inject.util.Modules$OverrideModule -&gt; org.apache.druid.server.metrics.MetricsModule)
  while locating org.apache.druid.java.util.metrics.MonitorScheduler
  at org.apache.druid.server.metrics.MetricsModule.configure(MetricsModule.java:77) (via modules: com.google.inject.util.Modules$OverrideModule -&gt; com.google.inject.util.Modules$OverrideModule -&gt; org.apache.druid.server.metrics.MetricsModule)
  while locating org.apache.druid.java.util.metrics.MonitorScheduler annotated with @com.google.inject.name.Named(value=ForTheEagerness)

5 errors
	at com.google.inject.internal.Errors.throwCreationExceptionIfErrorsExist(Errors.java:470)
	at com.google.inject.internal.InternalInjectorCreator.injectDynamically(InternalInjectorCreator.java:184)
	at com.google.inject.internal.InternalInjectorCreator.build(InternalInjectorCreator.java:110)
	at com.google.inject.Guice.createInjector(Guice.java:99)```

1. What’s your deployment environment like ?
2. What version are you running ?
3. I assume this is on-prem Clarity.  Would SaaS Clarity be an option for you ?

1. What’s your deployment environment like ? - K8
2. What version are you running ? 0.15.1-incubating apis
3. I assume this is on-prem Clarity.  Would SaaS Clarity be an option for you ? in production with ec2 instane it works, but with k8 deployment we are getting this error


A client has similar issue ( K8s as well )  and they resolved it thru Support. So you may want to consider that route if you are an Imply customer

If they are available on the historicals, are they queryable?, perhaps you could try reingesting them.

yes they are available in historical and we can query them

we r doing the delta load and it's keeping the intermediate copies which metadata is aware but we are not able to delete them

<https://medium.com/nmc-techblog/data-retention-and-deletion-in-apache-druid-74ffd12398a8>

In your question you said "Is it possible to move data from historical to deep storage"? In general any data in a historical should already be in deep storage - that is where historicals load their data from. If you find yourself in a situation where a historical has some data that is not in deep storage then you are in deep deep shit. The typical way to get into that situation is if an intern (who will not be named!) messes up an `aws s3` cli command and helpfully cleans your deep storage for you. I pray that you are not actually in that situation. If you are say so and I will try to help as best as I can.

in the second part of your question you mentioned "We have stale segments in deep storage (metadata is not aware of them) and the kill command is not taking care of cleaning." which suggests that you want to clean your deep storage and kill tasks are not working for you. If this is actually where you are then lets all be grateful it is not the previous thing I posted. :stuck_out_tongue:

we are in second part of situation, druid_segments where used='false' gives the positive number(which shows we have versions of the segments which came with delta load) so we are trying to clean them from deep storage. For example we a dataset with one segment but in deep storage we see 200plus files for it. we tried all possibilities of kill command but the kill command is not working

Also do you have realtime ingestion in this datasource?

we have both batch and stream , this case batch injestion one, below is the jsaon  the last segment is on 4-1 so we are killing until 3-31. We are submitting this to the coordinator through curl {
   "type": "kill",
   "dataSource": "abc",
   "interval" : "2022-03-01/2022-03-31"
}

Can you share the logs for that kill task?

we were able to run it successfully after initating it from coordinator dynamic config

Configure the coordinator/runtime.properties with the following
```druid.coordinator.kill.on=true 
druid.coordinator.kill.durationToRetain=P30D 
druid.coordinator.kill.maxSegments=100
druid.coordinator.kill.period=P1D```

• druid.coordinator.kill.on, has to be set to true to trigger the kill task.
• druid.coordinator.kill.durationToRetain, data to be retained in the Deep storage.
• druid.coordinator.kill.maxSegments, maximum/total segments to be removed per kill task.
• druid.coordinator.kill.period, this determines how often to run the kill task.


When we updated our parameters above, nothing was running

so our question is druid.coordinator.kill.on=true  only works when we do through coordinator

Just read through the thread . The property `druid.coordinator.kill.on` will only take effect on coordinator restart

We got that kiran, The point i want to mention is even though u have that property u have to  have the job through coordinator otherwise these parameters are of no use.

anyone have idea, where we specify the no of parallel sessions to run with kill job?

&gt; the job through coordinator otherwise these parameters are of no use.
How are you launching the kill task job ?

They are sending a kill task via curl (see above in thread)

Ah I see.
Yeah we would need to kill task job logs.

&gt;  anyone have idea, where we specify the no of parallel sessions to run with kill job?
I donot understand what you mean by this. Do you want to control how many kill tasks can run in parallel ?

I donot think we can control that. Is there a reason you would want to control it though. I guess after the initial pruning, where it might launch a lot of tasks, things would become stable. No ?

<@U038HLPD5TJ> this is quite an old version of druid. What are you trying to extract out of the broker logs ?
FWIW we do store the startTime on the router :
<https://github.com/apache/druid/blob/0.21.1/server/src/main/java/org/apache/druid/server/AsyncQueryForwardingServlet.java#L354>
to calculate the query/time.

In the broker and the historicals query metrics are reported here : <https://github.com/apache/druid/blob/ef006fd2571228984b04b95f3a3e2b03d930dce3/server/src/main/java/org/apache/druid/server/QueryLifecycle.java#L302> Idk if that answers your question

So yes we do store the start time of the query on the broker and the historical but that query is the "local" query

pasted the full query here FWIW:
```{"queryType":"topN","dataSource":{"type":"union","dataSources":[{"type":"table","name":"dsp_perf"},{"type":"table","name":"dsp_hha_impressions"}]},"virtualColumns":[],"dimension":{"type":"LegacyDimensionSpec","dimension":"line_id","outputName":"line_id","outputType":"STRING"},"metric":{"type":"numeric","metric":"advertiser_cost"},"threshold":10000,"intervals":{"type":"LegacySegmentSpec","intervals":["2022-12-15T05:00:00.000Z/2022-12-16T04:59:59.000Z"]},"filter":{"type":"and","fields":[{"type":"or","fields":[{"type":"selector","dimension":"order_id","value":"455853","extractionFn":null}]},{"type":"or","fields":[{"type":"selector","dimension":"valid","value":"1","extractionFn":null},{"type":"selector","dimension":"valid","value":null,"extractionFn":null},{"type":"selector","dimension":"valid","value":"true","extractionFn":null}]},{"type":"not","field":{"type":"or","fields":[{"type":"selector","dimension":"event_type","value":"4","extractionFn":null}]}},{"type":"or","fields":[{"type":"selector","dimension":"seat_id","value":"3331","extractionFn":null}]}]},"granularity":{"type":"all"},"aggregations":[{"type":"doubleSum","name":"advertiser_cost","fieldName":"advertiser_cost","expression":null}],"postAggregations":[],"context":{"priority":0,"queryId":"topN_union_request_1","timeout":20000},"descending":false}```

<@U0315MQKEHK> this is why <@U038HLPD5TJ> want to confirm/extract the start time from request log, sorry for being a bit spammy but I want to create a separate thread for the timed out issue just for visibility.

on historical:
```druid.server.http.numThreads=2000
druid.processing.numThreads=31```
we’ve been seeing this on larger clusters but can reproduce the behavior in a cluster with only 4 historical boxes.

Do you know what the cpu utilisation is on the historicals when this issue occurs?

For historical CPU was underutilized most of the time and no spike during the period.  eg. &lt; 10% for cpu.sys.pct/cpu.user.pct

The sub queries in a union query run back to back, not in parallel.

that kind of make sense, but still can explain for the first subquery it fire 16 secs later after receive the top level query

what is the CPU utilization on the broker looking like? Also what is your qps? If the union queries have multiple queries then one connection is open per sub query to the historical.

I suspect the issue is on the broker side though since it took 16 seconds to launch the query.
You can track : jetty/numOpenConnections on the broker.
<https://github.com/apache/druid/blob/master/server/src/main/java/org/apache/druid/server/initialization/jetty/JettyServerModule.java#L540>
The newer versions in druid have more metrics around jetty .

If the numOpenConnections is &gt; druid.server.http.numThreads (of the broker) then you might be connection bottlenecked and see why the queries are taking so long to respond.

Maybe the answer is external as in new q's on the system and we need more brokers to serve them.

<@U0315MQKEHK> thanks for the responses.  we also suspect some thread/connection/queue bottlenecks there, so checking the numOpenConnections might be a good idea.  Some questions  - Within what range the number is proper/acceptable? What config control that number?  what happen to those requests that cannot get open connection?

Regarding the idea of adding more broker, we haven’t give it some thoughts, as even when we tested with qps = 1 for 60 mins we can still see this issues started to happen after 10 - 20 mins

another qq, is `druid.broker.http.numConnections` the config control that the number of open connection? if so we have experimented with 500 while numThread set to 600, so it should always less than http thread unless it’s not getting honored

&gt;  Some questions  - Within what range the number is proper/acceptable? What config control that number?  
I think if `numOpenConnections` are constantly less than &lt; `druid.server.http.numThreads`  we should be good.
&gt; what happen to those requests that cannot get open connection?
Idk I have not tried it. I think the connection might be rejected but that would be more of a jetty thing.


&gt; Regarding the idea of adding more broker, we haven’t give it some thoughts, as even when we tested with qps = 1 for 60 mins we can still see this issues started to happen after 10 - 20 min
Since your query is a group by/topn query you might be using merge buffers which might become a bottleneck on the broker. You might want to increase the number of merge buffers and see if that helps. `druid.processing.numMergeBuffers`

&gt; `druid.broker.http.numConnections`
I think this is more related to broker to historical/indexer. How many historicals/taskslots  does queries generally get served by would help decide that number.
So for one jetty thread, ie "numThread" we can go out and reach out to 10 historicals and 5 peons so would require atleast 15 connections

so the main question how to investigate those issues, how to understand what’s blocking from beeing published in time?

A common reason for handoff to fail is if your historicals are full. You could see that on the Services tab of the console

thanks <@U030W50CNAC>, just to clarify you mean full in a sense of disk space configured?
we further investigated the problem and found it’s connected somehow to coordinator(we running it with overlord)
Restart of leader coordinator helped to release handoffs. As some metric that helped us is that number of jetty connections started to grow on coordinator(probably all other processes tried to call some endpoints of coordinator).
The question what metric I can check to understand what happened to coordinator. Looking in logs of coordinator didn’t revealed something special
Gc/Heap usage wasn’t high (same as before problem happened)

It sounds like your coordinator is taking a long time in one of it's cycles which means it's not handing off segments quick enough.

Do you have metrics enabled (<https://druid.apache.org/docs/latest/operations/metrics.html#jetty>)?

There is a metric for coordinator/global/time which will have a dimension called dutyGroup which will tell you how long per cycle the coordinator is spending on each duty. Hopefully it should be obvious once looking at those metrics to see which duty is problematic...

Hi <@U03HF9GLL1X> yes we have coordinator/global/time, but it’s reported through prometheus emmiter(so there are some differences)
I haven’t noticed something special in coordinator global times(what is normal here?)
I don’t have dimension by duty(fixing it now)
took
```max by (hostname)  (histogram_quantile(1, druid_coordinator_global_time_bucket{job="druid-spd", hostgroup=~".*$hostgroup"}))```
and got something like that
when green line stopped and yellow line started - is the switch between coordinators due to restart. So if I understand it correctly at problem time, coordinator cycle took up to 10seconds(which I believe not that big)

Apologies for not answering your question directly, but one thing you might consider is switching to http based task announcement. It is specifically designed to make concerns of znode size a thing of the past. The setting `druid.indexer.runner.type=httpRemote` is considered to be experimental as of the current release although in the upcoming 25 release (the RC for it is about to come out!) it will be the default. See <https://github.com/apache/druid/pull/13092> . I can tell you that we have been testing `httpRemote` extensively at Imply and it is the bee's knees.

Is there a significant delta between functionality/bugfixes for httpRemote between 24.0.1 and 25 (besides it being default)? looking through the closed PRs on the repo seems to suggest the answer is no. I'd be open to switching to that if there wasn't a lot of drift

to the best of my knowledge: no. It has actually been pretty stable for a while. We just had a hard time finding the person-power to push it into stable (like update the docs and stuff). The reason it is happening now is that MSQ is going GA and MSQ tends to make these huge task payloads so this really helps

Ok that makes sense. I can give this a try. Thanks!

in your example you would have to make 4 lookups, `id_to_street`  , `id_to_city`, etc

How do you guys tuned the number of merge buffers vs processing threads in historical?
In the past, we encountered some bugs when the number of processing threads is bigger than merge buffers.
So, currently our settings is 1:1.

Also, is the number of processing threads 1:1 to the number of CPU in the pods/bare metal?

Vision Druid - Zookeeper Warning Logs.rtf

Vision Druid - HTTP Client Request Logs.rtf

Could you share the Druid version that cluster is running?

wow that is old! I remember the 0.14 release line fondly because that was the first release with the new web console

I see you are using Druid SQL `druid/v2/sql` in the 0.14 line DruidSQL was still considered an experimental feature. I know that since then there were a lot of improvements to some edge cases in SQL parsing. Are there any specific queries that you notice causing the high CPU usage on these brokers?

I will try to get those queries (those are triggered by other team) but currently brokers are not responding queries at all. even those that we used to get in 1 second.  and brokers still reporting 100% CPU

It is possible that some query with an edge case parsing issue (in your version) is locking up a thread and one at a time all threads get jammed 

Are there queries with a large number of filter expressions like in and IN?

Also you must have been running 0.14 for a while . Did something change recently?

Our stakeholder (owner of the queries) recently (~past 2 months) added a new query, that polls for the previous 30d worth of data

That is the only known change - potentially we are hitting a max data retention/performance threshold on our cluster?

Seems this is what they are running:
```"""
New Engagement.
Every 10 minutes.
"""

END = datetime.utcnow()
START = END - timedelta(days=30)

# Now
client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    having=Having(type="greaterThan", aggregation="count", value=5),
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="now"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
    context={"timeout": 300_000_000},
)

client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    having=Having(type="greaterThan", aggregation="count", value=16),
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="now"),
            Filter(dimension="pzncon_event", value="seen"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
    context={"timeout": 300_000_000},
)

# PCC
client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    having=Having(type="greaterThan", aggregation="count", value=5),
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watchid"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
    context={"timeout": 300_000_000},
)

client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    having=Having(type="greaterThan", aggregation="count", value=16),
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watch_id"),
            Filter(dimension="pzncon_event", value="seen"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
    context={"timeout": 300_000_000},
)

"""
Engagement.
Every 10 minutes.
"""

END = datetime.utcnow()
START = END - timedelta(hours=24)

# Now
client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="now"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="now"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

# PCC
client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watchid"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

client.groupby(
    datasource="vision_conformed",
    granularity="all",
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watchid"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

"""
Popular.
Every 10 minutes.
"""

END = datetime.utcnow()
DURATION = timedelta(hours=1)

# PCC
client.groupby(
    datasource="vision_conformed",
    granularity={"type": "duration", "duration": duration.total_seconds() * 1000},
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watchid"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

# Now
client.groupby(
    datasource="vision_conformed",
    granularity={"type": "duration", "duration": duration.total_seconds() * 1000},
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watchid"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

"""
Trending.
Every 10 minutes.
"""

END = datetime.utcnow()
SLOTS = 5
DURATION = timedelta(hours=1)

# Now
client.groupby(
    datasource="vision_conformed",
    granularity={"type": "duration", "duration": duration.total_seconds() * 1000},
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="now"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)

# PCC
client.groupby(
    datasource="vision_conformed",
    granularity={"type": "duration", "duration": duration.total_seconds() * 1000},
    dimensions=["pzncon_content_id"],
    aggregations={"count": doublesum("count")},
    filter=Filter(
        type="and",
        fields=[
            Filter(dimension="pzncon_content_id_type", value="watchid"),
            Filter(dimension="pzncon_event", value="consumed"),
        ],
    ),
    intervals=f"{start.strftime('%Y-%m-%dT%H:%M:%S')}/{end.strftime('%Y-%m-%dT%H:%M:%S')}",
)```

What is the cardinality of the `pzncon_content_id` dimension?

It's not necessarily unique per each row - whenever anyone sees/consumes a content title within a production client app, a new row would be added

i.e. different individuals could consume the same content title

(can definitely discuss further during our sync shortly - <https://disney.zoom.us/j/5025748165?pwd=U3JsUjBMNXduMmlaTXFGRG4wbjlCdz09>)