Apache Druid

How it was removed / deleted except 1 which is stream today ?

is it anything to do with worker capacity / slots ?

Check for the retention rule in the Datasource page for the specific datasource.

I have cluster default rule only one : loadforever rule with replicant set to 1 (as single machine) and tier set to _default_tier

Then retention should not have deleted the rows

check for the coordinator logs, and historical for any clue on the deleted segments

what is the task duration of the kafka task? If the task failed for somereason before committing the records then you will see this

Hi vijay, where I can see the task duration ?  And for long hours yesterday till go home I could see the records in datasource, it mean it was committed right ?  or still in not yet commit stage ?

the records will show up in the datasource though they are still in the ingestion task memeory. So if the task fails before writing segments to deep storage then a new task has to start from the earlier kafka offset

you can see task duration in ingestion spec "taskDuration": "PT3600S"

so I guess the slot? worker capacity will be be trouble ?

and the slot / task would shared to others? I see PENDING state by today ? to that source

mostly duration was 1:00:06 shows under Tasks in web console

worker capacity may be a problem as you need atleast two slots to run 1 ingestion task. The task will shut down in hour and handover to another but both tasks will run together for some time when the handover happens

how many slots do you have and how many tasks?

so within limited worker capacity, but having task duration set higher can manage ?

or must to increase slots / worker capacity ?

Vijay, In another project I have less tasks more worker capacity where I not find this issue...

But I like to understand and learn the reason.

how many worker slots do you have for running 8 tasks? You should have 10-12 slots so that atleast some slots are available for task cycling

okay, yes currently I have only 7 worker capacity  (based on recommendation number of cpu - 1)

generally if data shows up and then mysteriously disappears, it's often a problem with handoff

<https://druid.apache.org/docs/latest/design/architecture.html#indexing-and-handoff>

often these are issues with metadata store or deep storage configuration, so i would double-check that stuff

you want to make sure deep storage is shared between all servers and metadata store is shared between all possible leading masters (Coordinator/Overlord)

if you just have one server this is probably not the problem — so in that case i'd suggest looking at the ingest task logs to see if there are errors, or looking at `sys.segments` table to see if you have any unavailable segments

that could indicate a problem getting them loaded up.

Hi Gian, Thank you for you details...  Sorry delay msg as am not well for a week.

Well Gian,
1. I tried (in mean time) by increasing the worker capacity and it's working till now. (to ensure all the tasks / data-source have enough slots to process and commits)
2. But as per recommendation of druid, worker capacity should =&gt; (no., of cpu - 1) 
 3.   Don't know what other would caused disappears !

it could be the historical vanished or had some other problem?

Where did you set druid.lookup.enableLookupSyncOnStartup, and did you restart services so it would take effect?   If that all checks out, you might also look at increasing <https://druid.apache.org/docs/latest/querying/lookups.html#saving-configuration-across-restarts|druid.lookup.numLookupLoadingThreads> if needed.

We are using docker for local development, so I set the properties in the 'environment' file, and i can find the following in the coordinator log files

2023-03-10 15:37:53 Setting druid.lookup.lookupTierIsDatasource=true in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.lookup.enableLookupSyncOnStartup=true in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.storage.type=local in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.metadata.storage.connector.user=druid in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.indexer.logs.directory=/opt/shared/indexing-logs in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.zk.service.host=zookeeper in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.indexer.fork.property.druid.processing.buffer.sizeBytes=256MiB in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.storage.storageDirectory=/opt/shared/segments in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.metadata.storage.connector.password=FoolishPassword in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.emitter.logging.logLevel=debug in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties
2023-03-10 15:37:53 Setting druid.lookup.numLookupLoadingThreads=4 in /tmp/conf/druid/cluster/master/coordinator-overlord/runtime.properties

Could try increasing  <https://druid.apache.org/docs/latest/querying/lookups.html#saving-configuration-across-restarts|druid.lookup.numLookupLoadingThreads>, see whether it helps.

2023-03-09 19:38:09 Caused by: org.apache.druid.java.util.common.ISE: Lookup [departments] not found
2023-03-09 19:38:09 at org.apache.druid.query.lookup.RegisteredLookupExtractionFn.lambda$ensureDelegate$0(RegisteredLookupExtractionFn.java:151) ~[druid-processing-0.23.0.jar:0.23.0]
2023-03-09 19:38:09 at java.util.Optional.orElseThrow(Optional.java:290) ~[?:1.8.0_275]
2023-03-09 19:38:09 at org.apache.druid.query.lookup.RegisteredLookupExtractionFn.ensureDelegate(RegisteredLookupExtractionFn.java:151) ~[druid-processing-0.23.0.jar:0.23.0]
2023-03-09 19:38:09 at org.apache.druid.query.lookup.RegisteredLookupExtractionFn.apply(RegisteredLookupExtractionFn.java:122) ~[druid-processing-0.23.0.jar:0.23.0]
2023-03-09 19:38:09 at org.apache.druid.query.expression.LookupExprMacro$1LookupExpr.eval(LookupExprMacro.java:91) ~[druid-processing-0.23.0.jar:0.23.0]

it's possible there is some issue loading the lookup — maybe it's available on the query server `:38888` but not the ingest task. i would suggest checking the ingest task log for clues

(web console -&gt; Ingestion tab -&gt; view log for task)

image.png

This is what appears in the ingestion wizard, however if I remove that and just edit the spec, then the 'submit' works and fails afterwards

2023-03-10T16:41:15,608 INFO [task-runner-0-priority-0] org.apache.druid.server.coordination.BatchDataSegmentAnnouncer - Announcing segment[druid-agent-states_2023-03-10T15:00:00.000Z_2023-03-10T16:00:00.000Z_2023-03-10T16:37:08.286Z] at new path[/druid/segments/172.18.0.11:8100/172.18.0.11:8100_indexer-executor__default_tier_2023-03-10T16:41:15.607Z_19858ac271b64299b8d15f133b7809650]
2023-03-10T16:41:15,657 ERROR [task-runner-0-priority-0] org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner - Encountered exception in run() before persisting.
org.apache.druid.java.util.common.ISE: Lookup [departments] not found
	at org.apache.druid.query.lookup.RegisteredLookupExtractionFn.lambda$ensureDelegate$0(RegisteredLookupExtractionFn.java:151) ~[druid-processing-0.23.0.jar:0.23.0]
	at java.util.Optional.orElseThrow(Optional.java:290) ~[?:1.8.0_275]
	at org.apache.druid.query.lookup.RegisteredLookupExtractionFn.ensureDelegate(RegisteredLookupExtractionFn.java:151) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.query.lookup.RegisteredLookupExtractionFn.apply(RegisteredLookupExtractionFn.java:122) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.query.expression.LookupExprMacro$1LookupExpr.eval(LookupExprMacro.java:91) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.transform.ExpressionTransform$ExpressionRowFunction.eval(ExpressionTransform.java:103) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.transform.Transformer$TransformedInputRow.getRaw(Transformer.java:207) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.incremental.IncrementalIndex.toIncrementalIndexRow(IncrementalIndex.java:580) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.incremental.IncrementalIndex.add(IncrementalIndex.java:515) ~[druid-processing-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.realtime.plumber.Sink.add(Sink.java:184) ~[druid-server-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.realtime.appenderator.StreamAppenderator.add(StreamAppenderator.java:276) ~[druid-server-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.realtime.appenderator.BaseAppenderatorDriver.append(BaseAppenderatorDriver.java:411) ~[druid-server-0.23.0.jar:0.23.0]
	at org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver.add(StreamAppenderatorDriver.java:189) ~[druid-server-0.23.0.jar:0.23.0]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:652) [druid-indexing-service-0.23.0.jar:0.23.0]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.run(SeekableStreamIndexTaskRunner.java:265) [druid-indexing-service-0.23.0.jar:0.23.0]
	at org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask.run(SeekableStreamIndexTask.java:149) [druid-indexing-service-0.23.0.jar:0.23.0]
	at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:477) [druid-indexing-service-0.23.0.jar:0.23.0]
	at org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:449) [druid-indexing-service-0.23.0.jar:0.23.0]

However, the lookup exists, and is query-able, as you can see in my printscreen

hmm do you have `druid.lookup.enableLookupSyncOnStartup` set in your configs? The default is `true`, which is good for making sure they're available

If it's overridden to `false` for some reason you'd want to change that

I am guessing that for some reason the task didn't load the lookup straight away

```druid-s3-extensions], jars: httpcore-4.4.11.jar, httpclient-4.5.10.jar, jmespath-java-1.12.37.jar, commons-codec-1.13.jar, commons-logging-1.1.1.jar, jackson-dataformat-cbor-2.10.5.jar, aws-java-sdk-core-1.12.37.jar, ion-java-1.0.2.jar, aws-java-sdk-sts-1.12.37.jar, joda-time-2.10.5.jar, druid-s3-extensions-0.22.1.jar```

i haven't seen it before, some additional error details might be helpful

```Caused by: com.amazonaws.SdkClientException: Data read has a different length than the expected: dataLength=243754376; expectedLength=292634686; includeSkipped=true; in.getClass()=class com.amazonaws.services.s3.AmazonS3Client$2; markedSupported=false; marked=0; resetSinceLastMarked=false; markCount=0; resetCount=0
	at com.amazonaws.util.LengthCheckInputStream.checkLength(LengthCheckInputStream.java:151) ~[aws-java-sdk-core-1.12.37.jar:?]
	at com.amazonaws.util.LengthCheckInputStream.read(LengthCheckInputStream.java:109) ~[aws-java-sdk-core-1.12.37.jar:?]
	at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:90) ~[aws-java-sdk-core-1.12.37.jar:?]
	at com.amazonaws.services.s3.internal.S3AbortableInputStream.read(S3AbortableInputStream.java:125) ~[aws-java-sdk-s3-1.12.37.jar:?]
	at com.amazonaws.internal.SdkFilterInputStream.read(SdkFilterInputStream.java:90) ~[aws-java-sdk-core-1.12.37.jar:?]
	at com.google.common.io.CountingInputStream.read(CountingInputStream.java:62) ~[guava-16.0.1.jar:?]
	at org.apache.druid.data.input.impl.RetryingInputStream.read(RetryingInputStream.java:144) ~[druid-core-0.22.1.jar:0.22.1]
	at java.io.FilterInputStream.read(FilterInputStream.java:133) ~[?:1.8.0_342]
	at java.util.zip.InflaterInputStream.fill(InflaterInputStream.java:238) ~[?:1.8.0_342]
	at java.util.zip.InflaterInputStream.read(InflaterInputStream.java:158) ~[?:1.8.0_342]
	at java.util.zip.GZIPInputStream.read(GZIPInputStream.java:117) ~[?:1.8.0_342]
	at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284) ~[?:1.8.0_342]
	at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326) ~[?:1.8.0_342]
	at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178) ~[?:1.8.0_342]
	at java.io.InputStreamReader.read(InputStreamReader.java:184) ~[?:1.8.0_342]
	at java.io.BufferedReader.fill(BufferedReader.java:161) ~[?:1.8.0_342]
	at java.io.BufferedReader.readLine(BufferedReader.java:324) ~[?:1.8.0_342]
	at java.io.BufferedReader.readLine(BufferedReader.java:389) ~[?:1.8.0_342]
	at org.apache.commons.io.LineIterator.hasNext(LineIterator.java:96) ~[commons-io-2.11.0.jar:2.11.0]```

&gt; does it keep happening if you try again? 
It’s the single failure from multiple running tasks

Have you tried druid-operator to run the cluster in kubernetes? It supports to upscale/downscale services under utilization metrics. <https://github.com/druid-io/druid-operator/blob/master/examples/tiny-cluster-hpa.yaml|Example>

```      hpAutoscaler:
        maxReplicas: 10
        minReplicas: 1
        scaleTargetRef:
           apiVersion: apps/v1
           kind: StatefulSet
           name: druid-tiny-cluster-brokers
        metrics:
         - type: Resource
           resource:
             name: cpu
             target:
               type: Utilization
               averageUtilization: 50```
on this case everytime k8s will spawn new broker pods (up to `maxReplicas` = 10) when cpu usage of current ones is &gt; 50%. When the load is down, it shuts down no longer needed instances (but never less than `minReplicas` amount).

And if you using k8s you can even use <https://druid.apache.org/docs/latest/development/extensions-contrib/k8s-jobs.html|middlemanager-less cluster, >so it spins small pods when you have ingestions and they die as soon as the task is done.

I'm afraid we're not using kubernetes, just regular docker; switching is definitely on the table but not our case yet

<@U041X7X3VND> <@U04L6JH4H2M>
If you are running on k8s, you can run without MM <https://druid.apache.org/docs/latest/development/extensions-contrib/k8s-jobs.html>

<@U0322ET4ZGE> I mentioned this in my last message :point_up:

in general, tasks aren't kept around longer than they need to be — druid will shut them down when their work is done

is there a reason you need to shut them down early in step (2b)? (instead of wait for them to exit)

hey, sorry <@U0344FW86DD> I missed your message. so what I'm doing is shutting down the _*middlemanager containers*_ themselves once they are idle (0 tasks running out of X slots)

apparently that messes running tasks up due to later stages somehow using data from middlemanagers that participated in other earlier stages of that particular ingestion task

I verified that by checking the payloads of tasks failing for apparently no reason, and indeed they were referencing middlemanagers that I killed due to them being idle

so I took the naive approach of just waiting for all tasks to be completed to scale middlemanagers down (effectively killing the containers); we currently only have need for them during a relatively small chunk of time during the day, so it's kind of working for now

but I was kind of surprised about this unexpected "coupling" between specific nodes in between stages

ahhh, I see. Yes, MMs provide a shuffle service that is used for `index_parallel` subtasks; the data on the shuffle service outlives the task themselves (but does not outlive the `index_parallel` task)

with SQL-based ingestion using MSQ we redesigned the shuffle mesh to not use the MM service, so it's been decoupled. You can be sure that if the tasks are gone then the MM is no longer needed. You may find value in switching to MSQ for this reason (and others! it's more robust generally in other ways too)

that's here: <https://druid.apache.org/docs/latest/multi-stage-query/index.html>

Another option, if you'd like to stick with `index_parallel`: you can use `druid.processing.intermediaryData.storage.type = deepstore` to use remote storage (S3, etc) as the shuffle service for `index_parallel`. That would also make it safe to terminate the MMs once the tasks are gone

oh! both outstanding suggestions Gian, thank you very much!! I’ll go with deep storage first as the quick and easy solution, and definitely eventually migrate to MSQ (have been looking forward to it); thanks again!

Everything I search for tells me that in this version of druid everything should be working without issue.

the service account is annotated:
```apiVersion: v1
automountServiceAccountToken: true
kind: ServiceAccount
metadata:
  annotations:
    <http://eks.amazonaws.com/role-arn|eks.amazonaws.com/role-arn>: 'arn:aws:iam::11111:role/druid-s3'```
and the pod gets the appropriate env vars:
```        - name: AWS_STS_REGIONAL_ENDPOINTS
          value: regional
        - name: AWS_DEFAULT_REGION
          value: us-west-2
        - name: AWS_REGION
          value: us-west-2
        - name: AWS_ROLE_ARN
          value: 'arn:aws:iam::11111:role/druid-s3'
        - name: AWS_WEB_IDENTITY_TOKEN_FILE
          value: /var/run/secrets/eks.amazonaws.com/serviceaccount/token```

this was on me.  I didn't set the service account name for the historical service in the OIDC setup on AWS

I've been trying to find it. I found this: <https://druid.apache.org/docs/latest/design/coordinator.html#segment-availability>
Which seems to indicate there is a dropped segment lifetime before being reassigned to other historicals. But like you, I can't find a parameter with that behavior description. The nearest I found is:
`druid.segmentCache.dropSegmentDelayMillis` which says
```How long a process delays before completely dropping segment.```
But this is a historical property, not sure if the coordinator would also pick it up or if it could be set to something larger. Seems more like a delay in deleting local files after the instruction to drop is received.

I would expect it to be a coordinator configuration. I'll let you know if I find anything.

in general there is an assumption that if historicals go away then their data should be rebalanced starting immediately

however, you can slow down that process via `replicationThrottleLimit` (see <https://druid.apache.org/docs/latest/configuration/index.html#dynamic-configuration>), assuming other replicas exist elsewhere. you might find this helpful

also i wonder why your historical frequently crashes? if i was you i would try to fix that— it sounds annoying! maybe we can help with this issue too.

Solved.
Correct configuration should be like following
```druid.request.logging.delegate.type=slf4j
druid.request.logging.type=filtered
druid.request.logging.mutedQueryTypes=["segmentMetadata"]```
Maybe worth to add available types for `druid.request.logging.delegate.type` in documentation

thanks for letting us know how you solved it!

ChatGpt to the rescue :stuck_out_tongue:
```You need to include it in the list of "aggregations" for the outer query to make sure it is retained in the final result.```
So the following works:
```{
  "queryType": "groupBy",
  "dataSource": {
    "type": "join",
    "left": "table1",
    "right": {
      "type": "query",
      "query": {
        "queryType": "topN",
        "aggregations": [{
            "type": "HLLSketchBuild",
            "name": "tmp_hash",
            "fieldName": "hash_field"
          }]}}
  },
  "filter": {},
  "aggregations": [
    {
      "type": "doubleSum",
      "name": "tmp_hash",
      "fieldName": "b.tmp_hash"
    }
  ],
  ...
  }```

so basically the prefix “b.” is recognised even outside the datasource block.

wow chatgpt should get into our slack channel

GC of 67s is very large. How many threads in the historical and how much heap to you have?

-Xms50g -Xmx50g -XX:+UseG1GC -XX:MaxDirectMemorySize=150g

I would suggest focusing on the 67s full gc— that's wild!

if we can address that, then your other issues will be mitigated

some suggestion here about how to figure out what's causing the GCs (heap histogram or alloc profiling): <https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1676370760624409?thread_ts=1676314924.103779&amp;cid=C0309C9L90D>

Hi Khajjiar, is there a reason not to use SQL then?

i think the sql will be converted to a groupBy query in which we will use HLLSketch with lgk=12. which will fail on our cluster.
I am converting a groupBy query to a topN query.

Okay, fishing around for hacks to get the native query to not be a GroupBy ...

```with tmp as (
 select floor(__time to HOUR) Hr, 
        count(*) cnt 
   from "wikipedia_hr"
  group by 1
)
 select t.* 
   from tmp t
  cross join (select 1 val)
  where t.cnt &gt; 1200```
Adding a benign join on the final select forces a query type of "scan" ... I don't know if you can use your sketches with a scan query, but if not then I suggest following this train of logic and trying post-agg operations that will not allow the "having" filter to push back into your group by.

Let us know what you find out.  Thx.

topN doesn't support HAVING on aggregations b/c it isn't really compatible with the topN optimization of pushing down limits to the segment: HAVING logically applies before LIMIT, so if we push down LIMIT, we must also push down HAVING, but if we push down HAVING, then it cannot see the full aggregation results! (because they are not available at the individual segment level.)

there is sort of a logical incompatibility there

btw, i know you're trying topN because you had issues with groupBy performing. one thing i wanted to ask is, could you try your groupBy query with `mergeThreadLocal: true` in your context, using Druid 25?

It's an experimental parameter that changes groupBy's merge on historicals to be thread-local instead of using a shared buffer, which reduces thread contention. We found it useful in certain cases and are wondering if we should make this on by default. Knowing if it helps you would help us make that decision :slightly_smiling_face:

Thanks Gian. i would have loved to try that param with Druid 25. The issue is that i don’t have much control over the builds/version on our dev cluster. I would see whats possible over the weekend.

ah, okay. It's available on 24.0 too, if you use that

The coordinator by default attempts to distribute segments in a way that makes querying most efficient - this is the `cachingCost` balancer strategy
If you are more interested in disk space, try setting `druid.coordinator.balancer.strategy` to `diskNormalized`
See: <https://druid.apache.org/docs/latest/configuration/index.html#coordinator-operation>

Hey Shilpa – once upon a time I had a balance problem – and it was down to `maxSegmentsToMove` – you may want to check that too

Thanks <@U030MBK46BD>, <@U0306F95YR3>
Will try out the options

If it is a few specific queries and you have already gone through the general cluster tuning and segment partitioning/optimization exercises, then if you post your query here you should be able to get some help.

I personally know a fair amount about SQL level tuning ... I know some segment partitioning/optimization, but others here know more ...

I don't know of a specific "slow query perf tuning" guide available to the public ... but I suggest check out the Imply (Druid) YouTube channel, there are a lot of architecture tech-talk videos, some of which may cover aspects of tuning for query performance.

<https://support.imply.io/hc/en-us/articles/360033747953-Profiling-Druid-queries-using-flame-graphs|Flame graphs> might be a good starting point.

Imply (my employer) has also recently published a free course on <https://learn.imply.io/apache-druid-metrics|Druid metrics>, and I think one of the labs focuses on identifying a slow query.

Are you using exact count distinct or approximate count distinct in druid?

Not this query very query is running slow in druid

We need to get on call. Are you ok for call at 530 pm?

<@U04CZ1Z9EUU> One thought is that you are not filtering on __time which means that the query will need to examine all segments in order to produce the result. What relationship does "asdt" have with __time, if they are aligned, adding the filter also on __time should be a boost to performance. Also, how is the table's secondary partitioning?

Unused segments will be deep storage but not on historical 

thanks , so is there an overhead on the cluster to keep unused sgements ..?
with regards to the segments processing / life-cycle ..
when / why I need to use the "kill" to delete un-used segments .. as long as they not on historical ?

If you have thousand of unused segments then they will cause load on the metastore as these are still tracked in the metastore and also the coordinator keeps track of these segments. So if you know you will never need these segments then issuing the kill task reduces metastore and coordinator overhead

Screenshot 2023-03-14 at 15.28.56.png

I was trying to check the above consumer in kafka with command  below pic but am not seeing any group though druid is connected

Hi <@U03DSHS8Z3N>, Druid doesn't require a consumer group to maintain its place in each partition, it has its own logic for checkpointing. You can still specify a kafka consumer group in the `consumerProperties` inside the `ioConfig` of your supervisor spec, like this:
```"ioConfig": {
    "type": "kafka",
    "consumerProperties": {
      "bootstrap.servers": "my-kafka-broker:9092",
      "group.id": "my-consumer-group-id"
    },
    "topic": "my-kafka-topic" ```

More info <https://druid.apache.org/docs/latest/development/extensions-core/kafka-supervisor-reference.html#kafkasupervisorioconfig|here> and <https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#ak-consumer-configurations-for-cp|here>.

In relation to this, “boolean” is also converted to “VARCHAR”. I think that’s not a correct behavior.

If implicit conversion supposed to happen, “boolean” should have been automatically converted to “long”.

Is this still true?

&gt; Druid natively supports five basic column types: “long” (64 bit signed int), “float” (32 bit float), “double” (64 bit float) “string” (UTF-8 encoded strings and string arrays), and “complex” (catch-all for more exotic data types like json, hyperUnique, and approxHistogram columns).

wow, strange! Thankfully this is not an issue with SQL based ingestion

I am sure this was unintentional, I think it's a side-effect of the fact that `defaultImpl` (with jackson deserialization) is the devil. i bet we can fix it by removing `defaultImpl` from something or other, and possibly adding a deserializer for back-compat if needed

It's possible the one on `DimensionSchema` is the one you are referring to

Screenshot 2023-03-14 at 7.59.49 PM.png

Coordinator dynamic config -
```{
  "millisToWaitBeforeDeleting": 900000,
  "mergeBytesLimit": 524288000,
  "mergeSegmentsLimit": 100,
  "maxSegmentsToMove": 200,
  "percentOfSegmentsToConsiderPerMove": 100,
  "replicantLifetime": 15,
  "replicationThrottleLimit": 150,
  "balancerComputeThreads": 7,
  "emitBalancingStats": false,
  "killDataSourceWhitelist": [],
  "killAllDataSources": false,
  "killPendingSegmentsSkipList": [],
  "maxSegmentsInNodeLoadingQueue": 1000,
  "decommissioningNodes": [],
  "decommissioningMaxPercentOfMaxSegmentsToMove": 0,
  "pauseCoordination": false
}```

Can you check the coordinator logs to see why these segments are not getting loaded?We can click on the unavailable segments on the console which will show the segment id of them, then we can try to grep for those ids in the coordinator logs

If you are collecting error logs from Historicals, you may notice errors about segments failing to load, as well

That could be your problem— sometimes people encounter that if certain Historicals cannot contact deep storage, for example

Thanks a lot <@U030LNJJ13M>, and <@U0344FW86DD> for your pointers. We were able to identify the root cause of the issue which was - one historical box was not able to load segments because of disk issues. Taking that historical down stabilized the entire cluster, and all segments loaded correctly.

Two follow up questions on this behavior -
1. We were not able to see the segments trying to be loaded on the historical from the coordinator endpoint, `druid/coordinator/v1/loadqueue` . That historical had 0 segments to load, and 0 segments to drop. Wondering why would this happen?
2. This historical is part of a low priority tier, but there are other tiers that are part of some highly critical revenue barring system which are must for our operations. How can we avoid one historical's misbehaviors from impacting other tiers? The coordinator got too busy in dealing with this historical, which caused the entire cluster to suffer. 
Thanks!

&gt; That historical had 0 segments to load, and 0 segments to drop. Wondering why would this happen?
:point_up: just a guess, but it could be failing so fast that its queue was empty

&gt; This historical is part of a low priority tier, but there are other tiers that are part of some highly critical revenue barring system which are must for our operations. How can we avoid one historical's misbehaviors from impacting other tiers? The coordinator got too busy in dealing with this historical, which caused the entire cluster to suffer.
:point_up:  on this, we're working on improvements to the Coordinator algorithm and we may want to consider this case as part of that. /cc <@U0318CX1NH4> who has been looking into these improvements

&gt; just a guess, but it could be failing so fast that its queue was empty
Yes, this is possible.

&gt; How can we avoid one historical's misbehaviors from impacting other tiers?
The coordinator already handles each tier separately. Did you see any exceptions in the coordinator logs when this was happening?

&gt; The coordinator got too busy in dealing with this historical, which caused the entire cluster to suffer.
We are working to ensure that the coordinator doesn't keep trying to assign segments to unresponsive/failing historicals. But that said, this shouldn't have affected loading on other tiers.

common.runtime.zip

looks like it is trying to connect to localhost as the coordinator and not able to.  Is the coordinator located on the same node?  Does localhost resolve correctly?

coordinator is on different server and the query server can reach it (verified by netcat)

Was this run previously on zookeeper as a single node cluster?

all the servers are single node including zookeeper. The behavior is similar for fresh setup or retires. Have also tried shutting down all servies and deleting the path in zookeeper and bringing up all services.

hmmm, it should be discovering the path via zookeeper. <https://www.druidforum.org/t/router-is-failing-to-start-due-to-failure-to-connect-to-coordinator/8101>

yeah I did follow this thread. Couldnt seem to be working for me.

check that the router can talk to zookeeper as well

also that the router config has the correct
```druid.router.coordinatorServiceName```


there is connectivity between router and zookeeper. Also the router logs convey it.

service name looks correct to me. I hope I am checking at correct location

are you running the coordinator under the overlord process?

I am not sure, I have not done any other changes explicitly, just kept it the default.

do you have a separate process for overlord and for coordinator?

the issue is in the discovery process.  The router cannot seem to find the coordinator.  I would check the paths via the zkCli to make sure there are existing paths as it is configured.  It may be because it is being run as one service

okay. when i query zk I can see the path for both coordinator and overload

I will try to run them separately and check the behaviour

check coordinator log as well to see if any errors stand out regarding zookeeper

I do see exceptions in master intermittently when connecting to data server for the ports 8083 and 8091

your common.runtime.properties has druid.host=localhost.  Try removing that

after removing the services started fine except for router. but now all the services are showing this exception

I believe what is happening is each service is being registered as localhost.  That is now in zookeeper.  The default for druid.host is InetAddress.getLocalHost().getCanonicalHostName()

please try that and try cleaning out the zk directories as you did before

okay, what you are suggesting is stopping all services, cleaning zk and starting them back (no changes in common.runtime.properties)

you need to put in druid.host with the default value, localhost is for single node clusters

druid.host=InetAddress.getLocalHost().getCanonicalHostName()

you should just be able to comment out druid.host on all common.runtime.properties and it will use the default value

did you previously just remove the value?

I am unsure why it is not working.  It should have used the default when you removed from all common.runtime.properties files and cleaned out zookeeper

you could put the ip or hostname in on each one but I do not believe that is necessary

you mean druid.host to their respective ips?

setting them to their respective ips all services are up and running

<@U03TAQRMJPQ> Thank you for your time and expertise. I could not have made it work otherwise :clap:

1. not sure about memory limits specifically, but `druid.server.http.maxSubqueryRows` allows you to limit by row count.

The best way to do this is try to do the join after aggregation. Can you paste the sql query here?

in general, Druid queries use a controlled amount of memory, and if they need too much you will get a `Resource limit exceeded` error: <https://druid.apache.org/docs/latest/querying/querying.html#query-errors>

In MSQ it would be an INSERT/SELECT statement such as:

```insert into ds2
 select floor(__time to HOUR) as __time,
        emailid, 
        count(distinct tmp_hash) as num_hashes
   from ds1
  group by 1, 2
partitioned by HOUR```
The ingestion shouldn't care about the segment granularity of ds1, it is only source data.

Let me know if that doesn't do what you want.

i am on druid .22
Should the datasource already exist?
```Object 'ds2' not found```

till now i have only used ingestion tasks to create datasources. so not aware of this. sorry if its too basic. learning.

What you have is the correct result. If you query ds2 with approx count ds hll you will get the result you want. Creating a hll sketch in the ingestion will replace the column with the sketch

Screen Shot 2023-03-15 at 6.36.26 PM.png

With INSERT INTO the datasource does not need to exist already.

Druid 22 doesn't have MSQ ... to my knowledge MSQ was added in Druid 24.  So you may need to upgrade, then add the "druid-multi-stage-query" extension to enable it.   One easy way to tell that you have MSQ enabled is to look at the SQL engine options in the Query tab (see screenshot) ...

Hi Vijay, pls find the query here
```{
  "queryType": "topN",
  "dataSource": {
    "type": "join",
    "left": "myDataSource",
    "right": {
      "type": "query",
      "query": {
        "queryType": "topN",
        "dataSource": "myDataSource",
        "dimension": "emailid",
        "threshold": 50000,
        "granularity": "all",
        "filter": {},
        "aggregations": [{
            "type": "HLLSketchBuild",
            "name": "dev_hash_chg_freq",
            "fieldName": "dev_hash",
            "lgK": 12,
            "tgtHllType": "HLL_4",
            "round": true
          }],
        "metric": {
          "type": "numeric",
          "metric": "dev_hash_chg_freq"
        },
        "intervals": []
      }
    },
    "rightPrefix": "b.",
    "condition": "emailid == \"b.emailid\"",
    "joinType": "INNER"
  },
  "virtualColumns": [],
  "dimension": {
    "type": "default",
    "dimension": "host",
    "outputName": "host",
    "outputType": "STRING"
  },
  "metric": {
    "type": "numeric",
    "metric": "totalcount"
  },
  "intervals": {
    "type": "intervals",
    "intervals": []
  },
  "filter": {},
  "granularity": "all",
  "aggregations": [{
      "type": "filtered",
      "aggregator": {
        "type": "count",
        "name": "totalcount"
      },
      "filter": {},
      "name": "totalcount"
    }],
  "context": {},
  "descending": false,
  "threshold": 10
}```

-----
yes, i think i need to o an approx count on ds2.
But since i am performing a kind of a background task with the 2nd ds, i was hoping to get aggregated count value there. something like
```email1   32
email1 has 32 unique dev_hashes in the last hour```
instead of the following as that would again need an aggregate.
email1.   AHeee
email1    AHfff

Sql version
```select count(*), a.host from myDatasource a inner join (
 select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000
    ) b
    on a.emailid=b.emailid
  where   
	"__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   GROUP BY a.host```

People are not happy with the inner join here. so i am trying to come up with 2 other DS with hourly and daily granularity. which can be used as a lookup source. that would contain
email vs distinct dev_hash count
I dont have a query yet for that

I assume the query is
```select count(*), a.host,a.emailid from myDatasource a inner join (
 select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000
    ) b
    on a.emailid=b.emailid
  where   
	"__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   GROUP BY a.host```

not really since the use of the emailid is only for joining as it represents a unique user. after that i need the host count irrespective of the emailid.

how much time does
```select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000```
take?

8s on a dataset of 21M records with 500k unique emailid

one thing is that your native query does not have any time filter.

oh i removed that for the sake of a smaller query/simplicity here. i can post the complete query if you need that

what is the requirement....I see that you are doing a hll in the right query but not using it in the final output. This is wasteful.

you need to get top 50k emailids by the dev_hash count?

actually i am working with native query only. i just converted this into sql and missed that condition. pasting it again.
```select count(*), a.host from myDatasource a inner join (
 select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') dev_hash_count from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000
    ) b
    on a.emailid=b.emailid
  where   
	"__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' AND b.dev_hash_count &gt; 20
   GROUP BY a.host```

I am looking for - “number of times host has been accessed if the unique hash count for any user is greater than a threshold”

try
```select a.host,sum(total) from (select count(*) total, host,emailid from myDatasource group by emailid,host) a inner join (
 select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') dev_hash_count from myDatasource from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000
    ) b
    on a.emailid=b.emailid
  where   
  "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
  AND b.dev_hash_count &gt; 20
   GROUP BY a.host```
I have re-written to aggregtae the left table first before the join

this will help if the aggregation reduces the left table significantly.

in your query the entire datasource will scanned for the left part of the join. Joins in druid execute on the broker and hence you will see impact on the broker.

forgot to add time filter into left query
```select a.host,sum(total) from (select count(*) total, host,emailid from myDatasource 
   where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00'  group by emailid,host) a inner join (
 select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') dev_hash_count from myDatasource from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000
    ) b
    on a.emailid=b.emailid
  where   
  "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
  AND b.dev_hash_count &gt; 20
   GROUP BY a.host```

moved the dev_hash condition into having....this way the number of rows coming out of right query is reduced
```select a.host,sum(total) from (select count(*) total, host,emailid from myDatasource 
   where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00'  group by emailid,host) a inner join (
 select emailid, APPROX_COUNT_DISTINCT_DS_HLL(dev_hash, 12, 'HLL_4') dev_hash_count from myDatasource from myDatasource
 where "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
   group by emailid
   order by 2 desc
   limit 50000 having dev_hash&gt;20
    ) b
    on a.emailid=b.emailid
  where   
  "__time" BETWEEN TIMESTAMP '2023-03-06 07:10:00' AND TIMESTAMP '2023-03-06 11:11:00' 
  
   GROUP BY a.host```

Thanks Vijay. I will try that and let you know.

The entire stack trace:
```Exception in thread "main" java.lang.ExceptionInInitializerError
	at com.google.inject.internal.cglib.reflect.$FastClass$Generator.getProtectionDomain(FastClass.java:73)
	at com.google.inject.internal.cglib.core.$AbstractClassGenerator.create(AbstractClassGenerator.java:206)
	at com.google.inject.internal.cglib.reflect.$FastClass$Generator.create(FastClass.java:65)
	at com.google.inject.internal.BytecodeGen.newFastClassForMember(BytecodeGen.java:252)
	at com.google.inject.internal.BytecodeGen.newFastClassForMember(BytecodeGen.java:203)
	at com.google.inject.internal.ProviderMethod.create(ProviderMethod.java:69)
	at com.google.inject.internal.ProviderMethodsModule.createProviderMethod(ProviderMethodsModule.java:275)
	at com.google.inject.internal.ProviderMethodsModule.getProviderMethods(ProviderMethodsModule.java:144)
	at com.google.inject.internal.ProviderMethodsModule.configure(ProviderMethodsModule.java:123)
	at com.google.inject.spi.Elements$RecordingBinder.install(Elements.java:340)
	at com.google.inject.spi.Elements$RecordingBinder.install(Elements.java:349)
	at com.google.inject.spi.Elements.getElements(Elements.java:110)
	at com.google.inject.internal.InjectorShell$Builder.build(InjectorShell.java:138)
	at com.google.inject.internal.InternalInjectorCreator.build(InternalInjectorCreator.java:104)
	at com.google.inject.Guice.createInjector(Guice.java:99)
	at com.google.inject.Guice.createInjector(Guice.java:73)
	at org.apache.druid.guice.BaseInjectorBuilder.build(BaseInjectorBuilder.java:66)
	at org.apache.druid.cli.Main.main(Main.java:96)
Caused by: java.lang.reflect.InaccessibleObjectException: Unable to make protected final java.lang.Class java.lang.ClassLoader.defineClass(java.lang.String,byte[],int,int,java.security.ProtectionDomain) throws java.lang.ClassFormatError accessible: module java.base does not "opens java.lang" to unnamed module @cf73d01
	at java.base/java.lang.reflect.AccessibleObject.throwInaccessibleObjectException(AccessibleObject.java:387)
	at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:363)
	at java.base/java.lang.reflect.AccessibleObject.checkCanSetAccessible(AccessibleObject.java:311)
	at java.base/java.lang.reflect.Method.checkCanSetAccessible(Method.java:201)
	at java.base/java.lang.reflect.Method.setAccessible(Method.java:195)
	at com.google.inject.internal.cglib.core.$ReflectUtils$1.run(ReflectUtils.java:52)
	at java.base/java.security.AccessController.doPrivileged(AccessController.java:319)
	at com.google.inject.internal.cglib.core.$ReflectUtils.&lt;clinit&gt;(ReflectUtils.java:42)
	... 18 more```

Hmm. If you're using the new `run-java` script it automatically adds a bunch, including that one

here's the list for java 17+ <https://github.com/apache/druid/blob/master/examples/bin/run-java#L33-L42>

That's the script we use when we test with Java 17, so things should work with those flags

There is a `sql` input source in native batch ingestion, which supports ingestion from the `postgres` database. You can use that to ingest the data from the existing postgres database to druid. More info can be found here: <https://druid.apache.org/docs/latest/ingestion/native-batch-input-sources.html#sql-input-source>.

The simpler approach is to dump the data from Postgres into file and then do a batch ingestion to druid

Thank Laksh and Vijay,,  I will check this . on a side note.. we want this process to be continuous streaming , not one time.

In case you want this to be continuous then both of the above approaches won’t work. Either you’ll have to schedule a job which ingests that new data (based on some filter, like time) or ingest the postgres data to something like Kafka and then run a continuous streaming ingestion from there.