Apache Druid

8101 seems to be PEON port, where the peon process runs. Please check middle manager log for the specific IP to have more information. What type if ingestion you are doing is it kafka and if yes are you using to read from first available offset, i mean is their data available in kafka.

Hi Bibek, I am trying to ingest Wikipedia sample data (available with Druid installation). The IP address is of Data node

Yes data server will have historical and middle manager services running, check the middle manager log to see if we get any more information on this

to run ingestion jobs, middlemanager service(from dataserver) launches new process which are called PEON process and it runs for the time of job/task and then dies. and each time the peon process starts in a new port so make sure your environment does have below ports open and accessible TCP PORT 8100 to 8199

Seems its not working with sample data. However,  I think ingestion is working via Kafka. Checking....if yes, this fulfills my requirement. Thanks

Hi All, I am trying to setup druid cluster with s3 deep storage, I was success in setting up coordinator, overload, broker, router and zookeeper, but while starting historical I am getting below error 

```druid.segmentCache.locations must be set on historicals```

But in my runtime properties I have this parameter value set 

Can you give an example of what you have set, and where you have set it?

The standalone start-Druid command is working fine 

The error shows to set druid.segmentCache.locations 

But I already have it in conf/druid/cluster/data/historical 

start-druid -s=middleManager,historical -c the cluster config location 

in `conf/druid/cluster/data/historical/runtime.properties` you have set `druid.segmentCache.locations`like
```druid.segmentCache.locations=[{"path":"var/druid/segment-cache","maxSize":"300g"}]```
?

Yes <@U030MBK46BD> updated the same but I have modified the path to a different mount point 

<@U030MBK46BD> should I be concerned of changing the location to a custom mount point ? 

u can specify whatever mount point you want

as long as its mapped to a partition that has the appropriate size and its at least the same if not higher than maxSize in terms of capacity

If it was already mounted to another mount point prior you may want to delete the old druid-segments folder

We use a `queryGranularity`of `minute`during ingestion and query.

image.png

I just tried to use `longLast` for ingestion and it seems to work at least in the data loader (via the sampler):

Opa! nope, that did not actually work: `LongLastAggregatorFactory is not supported during ingestion for rollup`

Yes, that it is quite frustrating. Here is the link to the relevant documentation: <https://druid.apache.org/docs/latest/querying/aggregations.html#:~:text=Do%20not%20use%20first%20and%20last%20aggregators%20for%20the%20double%2C%20float%2C%20and%20long%20types%20in%20an%20ingestion%20spec.%20They%20are%20only%20supported%20for%20queries.%20The%20string%2Dtyped%20aggregators%2C%20stringFirst%20and%20stringLast%2C%20are%20supported%20for%20both%20ingestion%20and%20querying>. I tried looking at the source code but couldn't figure out the reason.

I use stringFirst and stringLast to get the first and last values of an event that has a fixed daily timestamp, effectively getting the latest and earliest values that were emitted that day. This works well, however you can't group by an aggregation so you can't `select count(*), latest(myfield) from table group by 2` because it is an agg of an agg. But I found you can actually group by it by using a scalar instead of `latest` , eg `json_value(parse_json(to_json_string("myfield")),'$.rhs', 'STRING'))`. Ideally you could just use `json_value`, but you can't seem to treat a `Complex&lt;Tuple&lt;T1,T2&gt;&gt;`  as json even though it renders that way (probably an oversight in druid, no more than that).

I should warn doing that without perfect rollup or reasonably sized subqueries is not advised.

Thank you all for your replies :grinning:. <@U04P4A4J3JP> : Fortunately, we don’t have to group by that field, at least to my knowledge. Do you specify a timeColumn during ingestion, or are you using the default?

there is a PR to add this missing support - <https://github.com/apache/druid/pull/10949> - posting here in case anyone in this thread wants to take that PR to the finish line :wink:

<@U031A9D19NU> Thank you, that looks great :slightly_smiling_face: If I continue to be assigned on this project I might just pick that PR up. I have to brush up my java skills though :sweat_smile:

Too many small segments could negatively affect:
• Query performance, lots of segment files to open/scan
• Storage size, loss of economies of scale on column encoding dictionaries
• Coordinator and Broker memory heap, as they hold a segment metadata cache
• Metadata DB and interaction with it, if the number of segments is very large
Are you running compaction to defragment your time chunks?  How big are these segments?  How many segments per time chunk are we talking about?

Use automatic compaction it will will reduce file size based on your granularity

These are post compaction segments. Our proof of concept system post compaction has around 5-10,000 segments per timechunk (30 minutes). Each segment is between 500MB &gt; 1GB.

We're still scaling things and have already reduced the granularity from an hour to 30 minute but I think to stay below the 2,000 segment per timechunk theory, we need to move to 1 or 5 minute granularity.

Hi <@U03HF9GLL1X>,
Ideally each segment should have around 5 million rows as described here: <https://druid.apache.org/docs/latest/operations/segment-optimization.html>. How many rows are you ingesting every hour? Reducing the segment granularity will actually increase the number of segments. Think about it this way, if you have 5 million rows getting ingested every hour and your segment granularity is 1 hour, then you will have 24 segments for the day. If you reduce the segment granularity to 30 minutes then you will have double the number of segments in general. Also please check your kafka partition for data skews. Uneven partitions are very common when there is a data skew in kafka partitioning. I hope this helps.

We have 5 million rows per segment already. Our Kafka topic has around 1000 partitions (which is necessary to scale the ingest rate we need and probably also the root cause of the issue). In terms of rows per hour, we’re talking tens of billions. It’s a big system we’re developing for a large ISP…

Nick, In your case it sounds like you are already filling up the existing time chunks with large numbers of full sized segments.

So ... making granularity finer in this case would help with query pruning ONLY if your queries are filtering on time ranges smaller than 30 minutes ... e.g if your queries are mostly looking at the most recent 5-10 minutes of activity, then pruning would work better if your segment was at the 5-10 minute granularity or smaller.

If your __time value is being truncated to the same level as the segment granularity, and you are using range partitioning, then you may benefit slightly from smaller segments due to the records within the segment logically being sorted by the partitionig columns.   The records are always logically sorted first by the __time value, so for this to happen all of the __time values within the segment should be the same value.

You also want to make sure you don't lose the benefits of paralellism on the Historicals. For a given query, you want each Historical cpu to process N segments.  If there are fewer segments than historical cpus then you are leaving resources unused to process the query.   So, let's say you have 100 Historical cpus on your cluster, then for any given query, processing 500 or more segments should be good.

Hi <@U03HF9GLL1X> having around 1000 kafka partitions will definitely increase the number of segments that are getting generated during ingestion time. How many events are you ingesting per second? As can be seen in this blog post <https://imply.io/blog/exactly-once-streaming-ingestion/> each kafka partiton should be able to handle roughly around 100K events. I have worked with systems that ingest millions of events per second with less than 100 kafka partitions. This presentation shows the design decisions that should be followed for those circumstances <https://www.youtube.com/watch?v=2QDuSL6lNVk&amp;ab_channel=ironSource>.

At the scale you're talking, where each segmentGranular bucket would be fully populated no matter what, the main downside to the smaller granularities is that it's more likely you'll get fragmentation due to late data or task rollover

The way to think about it:

• If late data occurs, you get a new segment created for each time chunk that sees late data. If you typically get late data over, lets say, 90 minutes, then segmentGranularity of "hour" leads to two segments being open. But segmentGranularity of "minute" leads to at least 90 being open
• If tasks roll over crossing a segmentGranularity period, then the old and new tasks may both write some shorty segments. This is more likely to happen if segmentGranularity is short

The rollover point is more minor; the late data one tends to be the bigger issue

May not be an issue for you if you have minimal late data

We do have to contend with late data. We process the data before it hits kafka to ensure similar data ends up on the same partition which helps mitigate the scenario of fragmented data but not completely. This is where we're focusing a lot of our effort at the moment, in trying to ensure during ingest that segments are created as efficiently as possible with the minimum number of segments.

Did we ever track down what was causing the slowdown you saw with lots of segments in a single time chunk?

No - we couldn't get the coordinator to start successfully in the end (it kept running out of memory during startup probably due to the number of segments per time chunk). We trashed the cluster and started again with 30 minute granularity (and I feel we're in the same scenario again in that if I bounce a coordinator, I don't feel it'll come back as the 30 minute granularity is still seeing upwards of 20,000 segments per timechunk).

aargh. Let me see if we can repro something like this on our end. Although 20k is still a lot, so lowering your granularity further may be a good idea.

Gut check— what's the average row size of those 20,000 segments? Like, are they very small or are they "reasonable" size (a few million rows)?

I think it's about half and half. 50% percent are a few million rows or more, and then the rest is made up of late data / fragmentation which causes the rest.

Druid 26 has an UNNEST function that will let you do this. 26 is expected to be released this month.  Here is more about UNNEST - <https://imply.io/blog/exploring-unnest-in-druid/>

Thanks, nice timing on the article :slightly_smiling_face:

<@U031A9D19NU> looking over the article, it doesn't really resolve this issue though. It was more there's no obvious way to convert a COMPLEX value type that is actually holding an array to a MV in the first place.

For now, I've modified the ingestion to promote the array values to a first-child key in the incoming json payload. This means the SQL ingestion automatically identifies them as MV strings. However, if they're left lower in the JSON (very much preferable for dynamic payloads) then when using a JSON function to select the property, it's a COMPLEX.

that should be possible in 26 as far as I know. the changes are very recent. you can also build druid from master branch and try it out.

&gt; no obvious way to convert a COMPLEX value type that is actually holding an array to a MV in the first place
the idea is more that explicit `UNNEST` on an array provides the same behavior as the implicit unnest on an MV. So you're not converting to MV, but you're getting the same effect

Great, thanks <@U0344FW86DD>, sounds great, will give it a go

Let us know how it goes; as the blog says this is a new feature, beta, with a feature flag to enable it; at this point we're looking for feedback :slightly_smiling_face:

Are you still facing this? Could you provide a little more of the log for context?

Hi <@U030K4UM3H7> i missed to add port in AWS SG, now I got the issue resolved 

Hi All has anyone faced this issue before ? 

Screenshot 2023-04-18 at 11.37.57.png

Screenshot 2023-04-18 at 11.37.40.png

yes but that is not the way we would like to do manual way. we would like to automate it setting it via api requests

but let me double check again to set them via UI now

if you are able to do it console, you can look what request is console issueing. then compare it with the curl command that you are running

yeah, how can I check the curl command UI issues?

alright gonna check that. and any idea on why our requests seem in history section?

ups, just tried again but setting it in UI manually also does not work. after I set, it says it successfully set but then I see no retention policy set

it's possible that the retention policy isn't correct. you should check the coordinator logs.

I think <@U031A9D19NU> is correct the format for the rules you have should be:
```[
  { 
    "type":"dropBeforeByPeriod",  
    "period":"P4M",
    "tieredReplicants":{
      "hot":2,
      "_default_tier":2
     },
  },
  { "type":"loadForever" } 
]```
That said, the rule seem a bit odd. It will process the first rule that matches each segment in consideration and processes the rules in order. If you are trying to only hold 4 months of data, a more straightforward set of rules would look like:
```[
  { 
    "type":"loadByPeriod",  
    "period":"P4M",
    "tieredReplicants":{
      "hot":2,
      "_default_tier":2
     },
  },
  { "type":"dropForever" } 
]```
In this way, newly ingested segments with new timestamps will have 2 copies in _hot_ and 2 copies in __default_tier._ When the segments age is more than 4 months, the segments will be dropped.

Hey <@U030K4UM3H7>,

Thank you so much for your reply. My problem is more no matter what retention policy I apply, it just does not set even though it gives me success statu after API request.

That was just an example actually. And I tested it before, it was working I was able to see only segments in last 4 month

but then something happened and now I cant set any of the rules

it keeps going to history section , rather than actually setting it

I just checked the code for this and the only scenario I see where the history would be updated, but the actual rules would not be updated, is if your clock somehow went backwards

(The rules with the latest timestamp are the ones that are active, and if your clock goes backwards, it's possible that the latest timestamp corresponds to an earlier set of rules.)

Is it possible this is the case you're in?

You could check the `druid_rules` table in your metadata store, as well, to see what's in there and verify it makes sense

I recently learned that somehow druid_rules can have entries that have `null` payload. do you see something like that in your `druid_rules` table in the metadata store?

if you called the API to set retention rules with an empty payload, then your cluster can end up in a bad state.

Also tried with ztsd compressCodec as well. Its still not working.

```  "ioConfig": {
      "type": "hadoop",
      "inputSpec": {
        "type": "static",
        "inputFormat": "org.apache.druid.data.input.parquet.DruidParquetInputFormat",
        "paths": "&lt;path&gt;",
        "compressCodec": "zstd"
      }
    },```
Should we install any packages on druid servers to be able to read these data?

I think `<http://org.apache.hadoop.io|org.apache.hadoop.io>.compress.ZStandardCodec` is available from hadoop 2.9 onwards. we're currently using 2.8.2

We haven't been upgrading it since we more-or-less decided to keep it frozen in time, as people gradually move from hadoop onto native ingestion or MSQ

perhaps one of those (native or MSQ) would work for you too

or, you could try building a version of Druid with a later Hadoop.

You should be able access druid console on router port 8888(bydeafult) . e.g localhost:8888
How you are connecting deby database and what error it shows.

<@U03QYUNNFCN>
HTTP ERROR 500 org.apache.druid.java.util.common.IAE: No userMap is available for authenticator with prefix: [MyBasicMetadataAuthenticator]
 URI:/unified-console.html STATUS:500 MESSAGE:org.apache.druid.java.util.common.IAE: No userMap is available for authenticator with prefix: [MyBasicMetadataAuthenticator] SERVLET:org.eclipse.jetty.servlet.DefaultServlet-591a4d25 CAUSED BY:org.apache.druid.java.util.common.IAE: No userMap is available for authenticator with prefix: [MyBasicMetadataAuthenticator]

Can you pls share how you are connecting derby database .

also looks like some issues with basic auth setup  , Please use below sample configs
<https://gist.github.com/davidagee/c0c839cd23f047b838e8a3ea73320346>

druid.metadata.storage.type=derby
druid.metadata.storage.connector.connectURI=jdbc:<derby://localhost:1527/var/druid/metadata.db;create=true>
druid.metadata.storage.connector.host=localhost
druid.metadata.storage.connector.port=1527

Just to be clear: your metadata store configuration (connection to derby) and how you connect and authenticate to Druid (connecting to Druid to query it via JDBC or web console) are two totally different things that are not related to each other. If your Druid is generally working (i.e. you do not see a big scary warning when you open the web console) then you are connecting to Derby just fine

port 8888 is the port where the router runs on. The router process starts by default (in the default configs).

Can you access your router? it is where the web console is served from

I just trying to setup druid cluster in my local
I ran start-single-server-small script
then   created few tables in the server
from a java application connected  trough  the jdbc avatica   and able to fetch data

till here everything is fine
i'm able to access 8888 port and load new data and query in ui  every thing is fine

I added basic authentication  as per the doc
<https://druid.apache.org/docs/latest/development/extensions-core/druid-basic-security.html#coordinator-security-api>

```Connection connection = DriverManager.getConnection("jdbc:avatica:remote:url=<http://localhost:8082/druid/v2/sql/avatica/>","admin","password2");```


getting this error
IAE: No userMap is available for authenticator with prefix: [MyBasicMetadataAuthenticator]

I think you need to use these APIs to configure your user map: <https://druid.apache.org/docs/latest/development/extensions-core/druid-basic-security.html#usercredential-management>

Hi <@U053FHHSRJT> as part latest druid setup can you try to use start-druid command 

```start-druid -m 5000m ```

basic security API examples user management
<https://apachedruidworkspace.slack.com/archives/C0303FDCZEZ/p1680079104087319?thread_ts=1680008828.277329&amp;cid=C0303FDCZEZ>

Imply Druid has the View Manager extension:  <https://docs.imply.io/latest/druid/operations/views/view-manager/>

OSS Druid has a form of Materialized Views: <https://druid.apache.org/docs/latest/development/extensions-contrib/materialized-view.html#materialized-view-maintenance>

thanks <@U04C4593VDY>. I checked this, but my use case requires the view on union of two tables. is there any workaround for that?

Not that I know of, although you would be able to do it with an extension that adds a new `NamedDruidSchema` with the views you want. Gonna need to break out the IDE :slightly_smiling_face:

At some point we'll introduce ability to define SQL views, then you'd be able to use those without writing an extension

Thanks <@U0344FW86DD>, I will check this. My actual use case is to overcome some of the limitations in druid sql when using 'union all'. By creating a view on native query and providing the view as datasource to users would solve that.

Hi Oliver,

Do you have late arrival data?  If so, then you are ingesting for multiple time chunks at any given time ... when one time chunk persists, they all have to persist.

Also, the thresholds here apply to the sum of the data in the row buffers, not just the largest one.

Could the above explanation be what's happening in your case?

Sorry not sure to get your point no. What means the `(estimated) bytes[691,997,144]` ? Is it the estimated size of what it is going to be persist on disk ?

are you using sketches by any chance? They tend to over-estimate the memory

&gt; What means the `(estimated) bytes[691,997,144]`
According to the code it is the row buffer size:

```   // NB: The rows are still in memory until they're done persisting, but we only count rows in active indexes.
    rowsCurrentlyInMemory.addAndGet(-numPersistedRows);
    bytesCurrentlyInMemory.addAndGet(-bytesPersisted);

    <http://log.info|log.info>("Persisted rows[%,d] and (estimated) bytes[%,d]", numPersistedRows, bytesPersisted);```
I'm not sure why it is considered an estimate ... I see (estimated) used in multiple places ...

ah, the estimated memory usage is for in-memory data prior to persist

once persisted, it's compressed and would typically be much smaller

it's an estimate since it's all objects in the Java heap, so we estimate the size since the JVM does not provide a way to see what it really is.

This seems like a bug. can you create a github issue for this?

It's likely the real error is coming from a data server, possibly a Historical. Could you try checking the Historical logs for more info?

If you don't see anything, you might need to enable debug logging on the Historical

That'd help us fix it, or if you have steps to repro that'd help too :slightly_smiling_face:

indeed the error is from historical and i think its coming from here

```java.util.concurrent.ExecutionException: java.lang.NullPointerException
	at org.apache.druid.client.JsonParserIterator.convertException(JsonParserIterator.java:268) ~[druid-server-25.0.0.jar:25.0.0]```
no more additional logs in historical i can see only NullPointerException i can see
but i am not able to reproduce it in wiki datasource or the same s3 file datasource created in local.

the LATEST column is applied on a row where there are 2 distinct values `null and dummy`
also with EARLIEST  aggregator or LIMIT issue is not seen

when i add limit this passes but beyond 100000 (1 lakh) it again fails with same error

<@U04G9R2QA0K> - you can set `debug` to `true` in the query context. That could generate additional logging

i am getting same error msgs when trying with
"context": {
        "debug":*true*
    }

Hi Ross - you can use a `not` filter to help with this.
```"transformSpec": {
        "filter": {
          "type": "not",
          "field": {
            "type": "and",
            "fields": [
              {
                "type": "interval",
                "dimension": "__time",
                "intervals": [
                  "2016-06-27T00:00:00/2016-06-27T00:01:00"
                ]
              }​​​​​​,
              {
                "type": "selector",
                "dimension": "channel",
                "value": "#sv.wikipedia"
              }
            ]
          }
        }```
In the `wikipedia` example, this would remove any data with interval `2016-06-27T00:00:00/2016-06-27T00:01:00` and channel value of `#sv.wikipedia`

thanks <@U030MBK46BD> but wouldn't that remove the whole row from what I understand? I still want the row but a particular dimension value

So you are looking to substitute a dimension value?

<@U030MBK46BD> Yeah essentially substitute it with empty if it contains a certain value

Maybe then a `replace` expression in the transform <https://druid.apache.org/docs/latest/misc/math-expr.html#string-functions>

Not tested:
```"transformSpec": {
  "transforms": [
    { "type": "expression", "name": "columName", "expression": "replace(columnName, badValue, goodValue)" }
  ],```

If you are doing batch ingestion and you switch to SQL based ingestion you can express these transformations in SQL

Yes -- you can use "SQL Based Ingestion" ... please check out this doc page and let us know if you have specific questions ... <https://druid.apache.org/docs/latest/multi-stage-query/index.html>

In it's simplest form you are doing an "insert into table2 select * from table1", with partitioning and optional cluster clauses added to the end of the statement.

But the "select" part of the statement can really be any select statement that returns data.

You can also do this with a native spec using a Druid Input source in the ioconfig:  <https://druid.apache.org/docs/latest/ingestion/native-batch-input-sources.html#druid-input-source>

What do you mean exactly by "support for querying to supply data for Native Batch"? If you are referring to querying Druid as the source of data for a native batch job then you need the Druid input source <https://druid.apache.org/docs/latest/ingestion/native-batch-input-sources.html#druid-input-source> (follow the inputSource link from the page you linked). You can also have the console craft that spec for you by selecting "Reindex from Druid"

I've gotten this a few times:

&gt; [*INFO*] *------------------------------------------------------------------------*
&gt; [*INFO*] *BUILD FAILURE*
&gt; [*INFO*] *------------------------------------------------------------------------*
&gt; [*INFO*] Total time: 1.271 s
&gt; [*INFO*] Finished at: 2023-04-07T11:58:37-07:00
&gt; [*INFO*] *------------------------------------------------------------------------*
&gt; [*ERROR*] Failed to execute goal pl.project13.maven:git-commit-id-plugin:4.9.10:revision *(default)* on project druid: *Could not complete Mojo execution...*: pl.project13.core.NativeGitProvider$NativeCommandException: Git command exited with invalid status [1]: directory: `/Users/johnkowtko/BuildDruid/druid`, command: `git symbolic-ref --short HEAD`, stdout: ``, stderr: `xcrun: error: invalid active developer path (/Library/Developer/CommandLineTools), missing xcrun at: /Library/Developer/CommandLineTools/usr/bin/xcrun` -&gt; *[Help 1]*
&gt; [*ERROR*] 
&gt; [*ERROR*] To see the full stack trace of the errors, re-run Maven with the *-e* switch.
&gt; [*ERROR*] Re-run Maven using the *-X* switch to enable full debug logging.
&gt; [*ERROR*] 
&gt; [*ERROR*] For more information about the errors and possible solutions, please read the following articles:
&gt; [*ERROR*] *[Help 1]* <http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException>
I basically just started over with the clean and build process.  here is my instruction set that Iwrote down during a working session with a colleague ... I'm not a Java developer or Maven user (my colleague is) so not sure how much overlap there is here ... but it ended up working for me.

```mvn clean
mvn compile -DskipTests
mvn package -DskipTests  
mvn install -DskipTests  
cd distribution
mvn install -DskipTests -Pdist```

Hi Koustav,

When the the real-time data is first read from the streaming buffer (e.g. Kafka/Kinesis) it is stored in row format in a memory buffer.  Each row (or set of rows) that is pulled from the stream is parsed and placed in this buffer immediately, so if a query needs to read from this time chunk it can read it immediately directly from this memory buffer, although from another processing thread than the ingesting thread.

The "persist" is an intermediate form of storage of the real-time data ... in this stage the content of the row buffer is periodically converted to columnar format and stored in a small file on local disk on the ingesting node.  I don't know the exact format of this file but you can consider it as a "mini segment".  These persist operations can happen very frequently compared to the number of times an actual segment is built.  Dozens or even hundreds of persisted column store files can be persisted to disk.  And again, any query that needs to look at this time chunk will (via separate thread) read through all of these locally persisted files to scan for query results.

The final phase is to build a segment.  This happens when you either exceed a time limit (intermediate handoff period or task duration) or exceed a size limit (# rows/bytes per segment).  In this phase another thread in the ingestion tasks gathers up the persisted files, merges them into a *real* segment and pushes it to deep storage.

So, to answer you question in short, the "real-time" data is stored in both row and columnar format, at different times.

Thank You for your response.

Two more things I would like to know.

• Is it documented somewhere ?
• Is there any index on this data on memory buffer ?

I don't know if it's in the official documentation, you would have to search around on the Druid docs <https://druid.apache.org/docs/latest/design/index.html|HERE> or search through the Github repo <https://github.com/apache/druid|HERE> ...

Otherwise I suggest you look at the <http://learn.imply.io|learn.imply.io> Basics and Ingestion courses (they're free) ... they may take a few hours each to go through but they are very informative and offer online lab exercises.

I also suggest you poke around the Imply YouTube channel <https://www.youtube.com/@Implydata|HERE> ... there are lots of very informative videos there as well.  Fyi <@U030K4UM3H7> is working on a more detailed presentation on streaming ingestion specifically ... I don't know if he has any material to share yet but I am cc'ing him here just in case.

Thanks.  John

It depends on your workload and deployment mechanism. If Druid is deployed directly on physical nodes or VMs, it can make sense to run them together in a single node and split the CPU and Memory resources among them. Multiple nodes provide H/A for both MMs and Historicals with less nodes overall.
On kubernetes deployments, the same can be true. MMs and Historicals are deployed on separate containers. It is usually up to affinity and anti-affinity rules to select nodes on which to deploy the containers. So you can choose to also have them share underlying nodes or not.

The recommendation to share servers like that is for a couple reasons:

1) to keep things simple :slightly_smiling_face: with fewer different kinds of servers to worry about

2) to encourage CPU-sharing: if Historical or MM is not using the full CPU, then the other one can grab some. It's common for both of them to be bursty so this can help get better overall CPU utilization

However, if you are using k8s, or doing auto-scaling, or have a fancy multi-tier setup, it can make more sense to separate them

*In-person Meetup in Boston*

Hi all! For anyone in the Boston area there is an upcoming in-person Apache Druid meetup on Monday, May 1st at the WeWork on 1 Lincoln Street in Boston, MA.

The event is free to attend and food/drinks will be provided! You can learn more about the speakers/topics and <https://www.meetup.com/data-engineering-boston/events/292735125/>

Going to be tough to say what's going on without logs! I would suggest first trying to fix the logging problem. I wonder if you've customized something? By default, in recent versions, logs go "log" (or "logs", I don't remember the spelling) in the install directory 

Ok. we are using helm chart for the deployment. Any idea on how to enable it there

Ah, I am not familiar with the helm chart, sorry

Unknown exception / Unexpected response status [503] description [Service Unavailable] from request url[<http://10.101.35.229:8081/druid/indexer/v1/supervisor?system>] / <http://org.apache.druid.java.util.common.RE|org.apache.druid.java.util.common.RE> / on host 10.101.35.229

We have a new release for the druid operator
<https://github.com/datainfrahq/druid-operator/releases/tag/v1.1.0>
Slack Conversation

Our other clusters are still running 0.17.0 version… but new one is going to be the latest version.. so not sure if new version has better setup suggestions.

It's good to have 3 ZKs, for majority-rule

Some colocation would also help make best use of resources here.

I would do 3 "master server" (Coordinator, Overlord, ZK colocated on same server): <https://druid.apache.org/docs/latest/design/processes.html#master-server>

Since these are so big (64 CPU, 500GB RAM is quite large for a master) you could even co-locate the "query server" stuff (Broker, Router) on those same 3 physical servers: <https://druid.apache.org/docs/latest/design/processes.html#query-server>

That leaves you with 22 remaining servers

For these you have two basic options:

1) Colocate MM and Historical, "data server" style: <https://druid.apache.org/docs/latest/design/processes.html#data-server>

2) Separate MM and Historical

Advantage of (1) is potentially better resource utilization, as both kinds of processes can burst to use the full resources of those 22 servers.

Advantage of (2) is less resource conflict between the two kinds of processes. This is the typical approach in containerized or virtualized environments.

For a bare metal environment IMO (1) is a good place to start, as you don't have to think so much about how much resources to allocate to Historical vs MM

Yeah.. i am trying to figure out if this new cluster will have more query or injestion traffic… and then go off that.. but i figured, regardless 10 historical nodes should be a safe future proof bet and since those boxes will have much bigger SSD drives

Than you for your suggestions again, Gian!

do you have some more details on what you mean by "won't properly group"?

Getting `0` as result of aggregate function of sketch when changing group by from (str, str) -&gt; (str)

Could you please share the query (before and after) to illustrate what you mean by "when changing group by from (str, str) -&gt; (str)"

seconding the above :point_up:  it would be helpful to see the query text (&amp; results) for the one that works and the one that doesn't

```with t1(merge_column, s1) as
    (
      SELECT merge_column,
             DS_THETA(c1)
        FROM table1
    )


      SELECT 
        gp_col1,
        sum(ts2),
        sum(ts1),
        sum(ts3)
      FROM (
               SELECT gp_col1,
                      gp_multiplier,
                      THETA_SKETCH_ESTIMATE((DS_THETA(s2))) as ts2,
                      THETA_SKETCH_ESTIMATE((DS_THETA(s1))) as ts1,
                      gp_multiplier * THETA_SKETCH_ESTIMATE(THETA_SKETCH_INTERSECT(DS_THETA(s2), DS_THETA(s1))) as ts3
                 FROM (
                        SELECT merge_column,
                               gp_col1,
                               gp_multiplier,
                               DS_THETA(c2) as s2
                          FROM table2
                         WHERE gp_multiplier != 0 
                         GROUP BY gp_col1,
                      gp_multiplier
                      ) t2
                 JOIN t1
                   ON t1.merge_column = t2.merge_column
                GROUP BY gp_col1,
                      gp_multiplier
            )
      GROUP BY gp_col1```

The above query works fine in v22 and we get get non zero values for sum(ts2), sum(ts1) and sum(ts3)
In version 25, only getting non zero value for sum(ts2), others sums are returning 0. Though in version 25, ts2, ts1 and ts3 in subquery are returning correct values.

one more question… could you attach the results of `EXPLAIN PLAN FOR &lt;query&gt;` on the two versions?

hmm, could you double check your query too? When I try to validate it, `t2` doesn't validate:

```SELECT merge_column,
                               gp_col1,
                               gp_multiplier,
                               DS_THETA(c2) as s2
                          FROM table2
                         WHERE gp_multiplier != 0 
                         GROUP BY gp_col1,
                      gp_multiplier```
Because `merge_column` is neither in `GROUP BY` nor is it an aggregate

Actually we took static value `xyz` as merge_column

Explained query for part working in v25.
```{
  "queryType": "groupBy",
  "dataSource": {
    "type": "join",
    "left": {
      "type": "query",
      "query": {
        "queryType": "groupBy",
        "dataSource": {
          "type": "table",
          "name": "table2"
        },
        "intervals": {
          "type": "intervals",
          "intervals": [
            "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
          ]
        },
        "filter": {
          "type": "and",
          "fields": [
            {
              "type": "not",
              "field": {
                "type": "bound",
                "dimension": "gp_multiplier",
                "lower": "0",
                "upper": "0",
                "ordering": {
                  "type": "numeric"
                }
              }
            },
            {
              "type": "not",
              "field": {
                "type": "selector",
                "dimension": "gp_col1",
                "value": null
              }
            }
          ]
        },
        "granularity": {
          "type": "all"
        },
        "dimensions": [
          {
            "type": "default",
            "dimension": "gp_col1",
            "outputName": "d0",
            "outputType": "STRING"
          },
          {
            "type": "default",
            "dimension": "gp_multiplier",
            "outputName": "d1",
            "outputType": "STRING"
          }
        ],
        "aggregations": [
          {
            "type": "thetaSketch",
            "name": "a0",
            "fieldName": "c2",
            "size": 16384,
            "shouldFinalize": false
          }
        ],
        "postAggregations": [
          {
            "type": "expression",
            "name": "p0",
            "expression": "0"
          }
        ],
        "limitSpec": {
          "type": "NoopLimitSpec"
        },
        "context": {
          "queryId": "06fae828-88e7-4963-a7e4-8da9061e949e",
          "sqlOuterLimit": 1001,
          "sqlQueryId": "06fae828-88e7-4963-a7e4-8da9061e949e",
          "useNativeQueryExplain": true
        }
      }
    },
    "right": {
      "type": "query",
      "query": {
        "queryType": "timeseries",
        "dataSource": {
          "type": "table",
          "name": "table1"
        },
        "intervals": {
          "type": "intervals",
          "intervals": [
            "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
          ]
        },
        "granularity": {
          "type": "all"
        },
        "aggregations": [
          {
            "type": "thetaSketch",
            "name": "a0",
            "fieldName": "c1",
            "size": 16384,
            "shouldFinalize": false
          }
        ],
        "postAggregations": [
          {
            "type": "expression",
            "name": "p0",
            "expression": "0"
          }
        ],
        "context": {
          "queryId": "06fae828-88e7-4963-a7e4-8da9061e949e",
          "sqlOuterLimit": 1001,
          "sqlQueryId": "06fae828-88e7-4963-a7e4-8da9061e949e",
          "useNativeQueryExplain": true
        }
      }
    },
    "rightPrefix": "j0.",
    "condition": "(\"p0\" == \"j0.p0\")",
    "joinType": "INNER"
  },
  "intervals": {
    "type": "intervals",
    "intervals": [
      "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
    ]
  },
  "granularity": {
    "type": "all"
  },
  "dimensions": [
    {
      "type": "default",
      "dimension": "d0",
      "outputName": "_d0",
      "outputType": "STRING"
    },
    {
      "type": "default",
      "dimension": "d1",
      "outputName": "_d1",
      "outputType": "STRING"
    }
  ],
  "aggregations": [
    {
      "type": "thetaSketch",
      "name": "_a0",
      "fieldName": "a0",
      "size": 16384,
      "shouldFinalize": false
    },
    {
      "type": "thetaSketch",
      "name": "_a1",
      "fieldName": "j0.a0",
      "size": 16384,
      "shouldFinalize": false
    }
  ],
  "postAggregations": [
    {
      "type": "thetaSketchEstimate",
      "name": "s1",
      "field": {
        "type": "fieldAccess",
        "name": "s0",
        "fieldName": "_a0"
      }
    },
    {
      "type": "thetaSketchEstimate",
      "name": "s3",
      "field": {
        "type": "fieldAccess",
        "name": "s2",
        "fieldName": "_a1"
      }
    },
    {
      "type": "thetaSketchEstimate",
      "name": "s7",
      "field": {
        "type": "thetaSketchSetOp",
        "name": "s6",
        "func": "INTERSECT",
        "size": 16384,
        "fields": [
          {
            "type": "fieldAccess",
            "name": "s4",
            "fieldName": "_a0"
          },
          {
            "type": "fieldAccess",
            "name": "s5",
            "fieldName": "_a1"
          }
        ]
      }
    },
    {
      "type": "expression",
      "name": "s8",
      "expression": "(CAST(\"_d1\", 'DOUBLE') * \"s7\")"
    }
  ],
  "limitSpec": {
    "type": "default",
    "columns": [],
    "limit": 1001
  },
  "context": {
    "queryId": "06fae828-88e7-4963-a7e4-8da9061e949e",
    "sqlOuterLimit": 1001,
    "sqlQueryId": "06fae828-88e7-4963-a7e4-8da9061e949e",
    "useNativeQueryExplain": true
  }
}```

Explained query for Query working in v22 and giving issues in v25.
```{
  "queryType": "groupBy",
  "dataSource": {
    "type": "query",
    "query": {
      "queryType": "groupBy",
      "dataSource": {
        "type": "join",
        "left": {
          "type": "query",
          "query": {
            "queryType": "groupBy",
            "dataSource": {
              "type": "table",
              "name": "table2"
            },
            "intervals": {
              "type": "intervals",
              "intervals": [
                "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
              ]
            },
            "filter": {
              "type": "and",
              "fields": [
                {
                  "type": "not",
                  "field": {
                    "type": "bound",
                    "dimension": "gp_multiplier",
                    "lower": "0",
                    "upper": "0",
                    "ordering": {
                      "type": "numeric"
                    }
                  }
                },
                {
                  "type": "not",
                  "field": {
                    "type": "selector",
                    "dimension": "gp_col1",
                    "value": null
                  }
                }
              ]
            },
            "granularity": {
              "type": "all"
            },
            "dimensions": [
              {
                "type": "default",
                "dimension": "gp_col1",
                "outputName": "d0",
                "outputType": "STRING"
              },
              {
                "type": "default",
                "dimension": "gp_multiplier",
                "outputName": "d1",
                "outputType": "STRING"
              }
            ],
            "aggregations": [
              {
                "type": "thetaSketch",
                "name": "a0",
                "fieldName": "c2",
                "size": 16384,
                "shouldFinalize": false
              }
            ],
            "postAggregations": [
              {
                "type": "expression",
                "name": "p0",
                "expression": "0"
              }
            ],
            "limitSpec": {
              "type": "NoopLimitSpec"
            },
            "context": {
              "queryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
              "sqlOuterLimit": 1001,
              "sqlQueryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
              "useNativeQueryExplain": true
            }
          }
        },
        "right": {
          "type": "query",
          "query": {
            "queryType": "timeseries",
            "dataSource": {
              "type": "table",
              "name": "table1"
            },
            "intervals": {
              "type": "intervals",
              "intervals": [
                "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
              ]
            },
            "granularity": {
              "type": "all"
            },
            "aggregations": [
              {
                "type": "thetaSketch",
                "name": "a0",
                "fieldName": "c1",
                "size": 16384,
                "shouldFinalize": false
              }
            ],
            "postAggregations": [
              {
                "type": "expression",
                "name": "p0",
                "expression": "0"
              }
            ],
            "context": {
              "queryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
              "sqlOuterLimit": 1001,
              "sqlQueryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
              "useNativeQueryExplain": true
            }
          }
        },
        "rightPrefix": "j0.",
        "condition": "(\"p0\" == \"j0.p0\")",
        "joinType": "INNER"
      },
      "intervals": {
        "type": "intervals",
        "intervals": [
          "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
        ]
      },
      "granularity": {
        "type": "all"
      },
      "dimensions": [
        {
          "type": "default",
          "dimension": "d0",
          "outputName": "_d0",
          "outputType": "STRING"
        },
        {
          "type": "default",
          "dimension": "d1",
          "outputName": "_d1",
          "outputType": "STRING"
        }
      ],
      "aggregations": [
        {
          "type": "thetaSketch",
          "name": "_a0:a",
          "fieldName": "a0",
          "size": 16384,
          "shouldFinalize": false
        },
        {
          "type": "thetaSketch",
          "name": "_a1:a",
          "fieldName": "j0.a0",
          "size": 16384,
          "shouldFinalize": false
        }
      ],
      "postAggregations": [
        {
          "type": "thetaSketchEstimate",
          "name": "p1",
          "field": {
            "type": "fieldAccess",
            "name": "p0",
            "fieldName": "_a0:a"
          }
        },
        {
          "type": "thetaSketchEstimate",
          "name": "p3",
          "field": {
            "type": "fieldAccess",
            "name": "p2",
            "fieldName": "_a1:a"
          }
        },
        {
          "type": "thetaSketchEstimate",
          "name": "p7",
          "field": {
            "type": "thetaSketchSetOp",
            "name": "p6",
            "func": "INTERSECT",
            "size": 16384,
            "fields": [
              {
                "type": "fieldAccess",
                "name": "p4",
                "fieldName": "_a0:a"
              },
              {
                "type": "fieldAccess",
                "name": "p5",
                "fieldName": "_a1:a"
              }
            ]
          }
        },
        {
          "type": "expression",
          "name": "p8",
          "expression": "(CAST(\"_d1\", 'DOUBLE') * \"p7\")"
        }
      ],
      "limitSpec": {
        "type": "NoopLimitSpec"
      },
      "context": {
        "queryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
        "sqlOuterLimit": 1001,
        "sqlQueryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
        "useNativeQueryExplain": true
      }
    }
  },
  "intervals": {
    "type": "intervals",
    "intervals": [
      "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
    ]
  },
  "granularity": {
    "type": "all"
  },
  "dimensions": [
    {
      "type": "default",
      "dimension": "_d0",
      "outputName": "d0",
      "outputType": "STRING"
    }
  ],
  "aggregations": [
    {
      "type": "doubleSum",
      "name": "a0",
      "fieldName": "p1"
    },
    {
      "type": "doubleSum",
      "name": "a1",
      "fieldName": "p3"
    },
    {
      "type": "doubleSum",
      "name": "a2",
      "fieldName": "p8"
    }
  ],
  "limitSpec": {
    "type": "default",
    "columns": [],
    "limit": 1001
  },
  "context": {
    "queryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
    "sqlOuterLimit": 1001,
    "sqlQueryId": "1b05327f-6904-4e6b-8876-225929d81ab9",
    "useNativeQueryExplain": true
  }
}```

is the native query (from `EXPLAIN PLAN FOR`) the same in 22 and 25? I'm wondering if the issue is the plan changed, or the plan stayed the same but the behavior of the native query changed

Explain Query plan on version 22
```
DruidOuterQueryRel(query=[{"queryType":"topN","dataSource":
  {"type":"table","name":"__subquery__"},
  "virtualColumns":[],
  "dimension":{"type":"default","dimension":"gp_col1","outputName":"d0","outputType":"STRING"},
  "metric":{"type":"dimension","previousStop":null,"ordering":{"type":"lexicographic"}},
  "threshold":101,"intervals":{"type":"intervals",
  "intervals":["-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"]},
  "filter":null,
  "granularity":{"type":"all"},
  "aggregations":[{"type":"doubleSum","name":"a0","fieldName":"ts2","expression":null},
    {"type":"doubleSum","name":"a1","fieldName":"ts1","expression":null},
    {"type":"doubleSum","name":"a2","fieldName":"ts3","expression":null}],
  "postAggregations":[],"context":{"sqlOuterLimit":101,"sqlQueryId":"ba9dcb86-9f4f-4c01-ab04-569ffd1cec08"},
"descending":false}],
 signature=[{d0:STRING, a0:DOUBLE, a1:DOUBLE, a2:DOUBLE}]) 
DruidJoinQueryRel(condition=[=($0, $4)], joinType=[inner], 
  query=[{"queryType":"groupBy",
    "dataSource":{
    "type":"table","name":"__join__"},
    "intervals":{
    "type":"intervals","intervals":["-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"]},
    "virtualColumns":[],"filter":null,"granularity":{"type":"all"},
    "dimensions":
    [{"type":"default","dimension":"gp_col1","outputName":"d0","outputType":"STRING"},
      {"type":"default","dimension":"gp_multiplier","outputName":"d1","outputType":"STRING"}],
    "aggregations":[
      {"type":"thetaSketch","name":"a0","fieldName":"s2","size":16384,"shouldFinalize":true,"isInputThetaSketch":false,"errorBoundsStdDev":null},
  {"type":"thetaSketch","name":"a1","fieldName":"EXPR$1","size":16384,
  "shouldFinalize":true,"isInputThetaSketch":false,"errorBoundsStdDev":null}],
    "postAggregations":[{"type":"thetaSketchEstimate","name":"p1",
      "field":{
      "type":"fieldAccess","name":"p0","fieldName":"a0"},"errorBoundsStdDev":null},
      {"type":"thetaSketchEstimate","name":"p3",
        "field":{"type":"fieldAccess","name":"p2","fieldName":"a1"},"errorBoundsStdDev":null},
        {"type":"thetaSketchEstimate","name":"p7",
          "field":{"type":"thetaSketchSetOp","name":"p6","func":"INTERSECT","size":16384,
            "fields":[{"type":"fieldAccess","name":"p4","fieldName":"a0"},
              {"type":"fieldAccess","name":"p5","fieldName":"a1"}]},"errorBoundsStdDev":null},
            {"type":"expression","name":"p8","expression":"(CAST(\"d1\", 'DOUBLE') * p7)","ordering":null}
          ],"having":null,"limitSpec":{"type":"NoopLimitSpec"},
    "context":{"sqlOuterLimit":101,"sqlQueryId":"ba9dcb86-9f4f-4c01-ab04-569ffd1cec08"},"descending":false}], 
  signature=[{d0:STRING, p1:DOUBLE, p3:DOUBLE, p8:DOUBLE}]) 
DruidQueryRel(query=[{"queryType":"groupBy","dataSource":{"type":"table","name":"table2"},
  "intervals":{"type":"intervals","intervals":["2023-03-01T00:00:00.000Z/2023-03-01T00:00:00.001Z"]},
  "virtualColumns":[],"filter":{
    "type":"and","fields":[{
      "type":"not","field":{
        "type":"bound","dimension":"gp_multiplier","lower":"0","upper":"0","lowerStrict":false,"upperStrict":false,"extractionFn":null,"ordering":{
        "type":"numeric"}}},{
          "type":"not","field":{
          "type":"selector","dimension":"gp_col1","value":null,"extractionFn":null}}]},
    "granularity":{"type":"all"},
    "dimensions":[{"type":"default","dimension":"gp_col1","outputName":"d0","outputType":"STRING"},
      {"type":"default","dimension":"gp_multiplier","outputName":"d1","outputType":"STRING"}],
    "aggregations":[{"type":"thetaSketch","name":"a0","fieldName":"c2","size":16384,"shouldFinalize":true,"isInputThetaSketch":false,"errorBoundsStdDev":null}],
    "postAggregations":[{
    "type":"expression","name":"p0","expression":"0","ordering":null},{
    "type":"fieldAccess","name":"p1","fieldName":"a0"}],"having":null,
    "limitSpec":{"type":"NoopLimitSpec"},"context":{
    "sqlOuterLimit":101,"sqlQueryId":"ba9dcb86-9f4f-4c01-ab04-569ffd1cec08"},"descending":false}],
     signature=[{p0:LONG, d0:STRING, d1:STRING, p1:COMPLEX}]) 
DruidQueryRel(query=[{"queryType":"timeseries","dataSource":{
"type":"table","name":"table1"},
"intervals":{
"type":"intervals",
"intervals":["2023-03-15T00:00:00.000Z/2023-03-16T00:00:00.000Z"]},
"descending":false,"virtualColumns":[],"filter":null,"granularity":{"type":"all"},
"aggregations":[{
"type":"thetaSketch","name":"a0","fieldName":"c1","size":16384,"shouldFinalize":true,"isInputThetaSketch":false,"errorBoundsStdDev":null}],
"postAggregations":[{"type":"expression","name":"p0","expression":"0","ordering":null},
  {"type":"fieldAccess","name":"p1","fieldName":"a0"}],"limit":2147483647,"context":{
"sqlOuterLimit":101,"sqlQueryId":"ba9dcb86-9f4f-4c01-ab04-569ffd1cec08"}}], signature=[{p0:LONG, p1:COMPLEX}])```

<@U030W50CNAC> <@U0344FW86DD> We got different query plan in v22 and v25 for same query. We verified the query plan from v25 works in v22 but doesn't work in v25.
We are feeling issue comes in v25 when we join and multiplex / demultiplex rows of the sketch values.
Any pointers what could be the problem and how to resolve this ?

oh… we did see very similar issue after upgrading to Druid 25.
The smoking gun seemed to be:
1. The query plan changed.
2. Because of that, some of our historical daemons started to handle a lot more data and thus starts timing out.
Join regression seemed to come from v24 and newer.

<@U03149ALHN3> were you able to find any solution for the Join issue ?

Unfortunately, because it’s mission critical to stay on Druid 25, we asked the developers and data engineers to rewrite their queries either at Druid level or Spark level (creating new denormalized columns instead of performing JOIN at Druid level).

<@U03FN50E7ND> Sorry, I did spend some time trying to reproduce this but haven't yet figured it out. If you have a way to repro this that works on a fresh install, that would be extremely helpful! Otherwise I'll keep poking at it trying to come up with a repro

<@U03149ALHN3> if you have any more info about the one you saw, I'd love to learn about it so we can add a planner test case. Hate to see plans get worse when versions change, &amp; the best way to prevent that is having more planner test cases

Hey! I have some of that info in my vid on approximation: <https://youtu.be/fSWwJs1gCvQ?list=PLDZysOZKycN7MZvNxQk_6RbwSJqjSrsNR>

(Props to <@U0344FW86DD> for a lot of the actual content in that deck)

Thanks for the video, it was helpful! You said that there are published error tables showing the accuracy of theta sketches, can you please tell me where can I find those published tables?

You can usually find these in the Data Sketches docs e.g., <https://datasketches.apache.org/docs/Theta/ThetaErrorTable.html>

You can add additional <https://druid.apache.org/docs/latest/development/extensions-core/kafka-supervisor-reference.html#kafkasupervisorioconfig|consumer properties to the ingestion spec> that are used to establish the connection. Here's an example:

consumerProperties are any <https://kafka.apache.org/documentation/#consumerconfigs|kafka consumer config> setting you want to use, like:
```{
"bootstrap.servers": "&lt;kafka bootstrap servers&gt;",
"security.protocol": "SASL_SSL",  
"sasl.mechanism": "AWS_MSK_IAM",
"sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required awsRoleArn='arn:aws:iam::xxxxxxxxx:role/msk-imply-dev' awsRoleSessionName='xxxxxxxx' awsStsRegion='us-east-1';",
"sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
}```


So far my understanding is we can deploy the Druid cluster using the Imply manager, so i have a K8 template to deploy the Imply manager. which is not working for me. Will  appreciate any help here.

hi, this is an Apache Druid space; questions about Imply's distribution would be better directed to your Imply rep / support contact

It works:
(SELECT * FROM “wikipedia” WHERE “__time” &lt; TIMESTAMP ‘2016-06-27 00:00:58.599’ LIMIT 3)
UNION ALL
SELECT * FROM “wikipedia” WHERE “__time” &gt;= TIMESTAMP ‘2016-06-27 00:00:58.599’