Scan query gracefully closing connections which results in partial data read from datasource

[bharadwajrembar]

Affected Version

Druid 0.19.1

Description

The scenario arises when we fire a Scan query to read all rows for a given day (typically around 1mill records, takes around 10-15 min to complete) from the datasource.

We observe that if there are N records for that given day, sometimes it will read <N records and throws no errors/timeouts resulting in a mismatch of data. We believe Druid is closing the connection gracefully. We have set a very high query timeout on all of our components as well (around 60min).

Cluster size

6 brokers - m5.2xlarge
15 historicals - i3.2xlarge
2 routers - t3.small

Configurations

Broker

druid.broker.balancer.type=connectionCount
druid.broker.http.maxQueuedBytes=10000000
druid.broker.http.numConnections=50
druid.broker.http.readTimeout=PT40M
druid.broker.http.unusedConnectionTimeout=PT35M
druid.broker.retryPolicy.numTries=2
druid.broker.segment.watchedTiers=["ingestion_tier"]
druid.broker.select.tier=custom
druid.broker.select.tier.custom.priorities=[1]
druid.plaintextPort=11025
druid.processing.buffer.sizeBytes=750000000
druid.processing.numMergeBuffers=6
druid.processing.numThreads=1
druid.query.groupBy.maxOnDiskStorage=10000000000
druid.server.http.defaultQueryTimeout=3600000
druid.server.http.maxIdleTime=PT30M
druid.server.http.numThreads=60
druid.service=druid/broker-ingestion
druid.processing.tmpDir=/mnt/data/processing

Historical

druid.plaintextPort=11026
druid.processing.buffer.sizeBytes=500000000
druid.processing.numMergeBuffers=4
druid.processing.numThreads=15
druid.query.groupBy.maxOnDiskStorage=10000000000
druid.segmentCache.numBootstrapThreads=30
druid.segmentCache.numLoadingThreads=4
druid.server.http.defaultQueryTimeout=3600000
druid.server.http.maxIdleTime=PT30M
druid.server.http.numThreads=250
druid.server.maxSize=1850000000000
druid.server.priority=1
druid.server.tier=ingestion_tier
druid.service=historical-ingestion
druid.processing.tmpDir=/mnt/data/processing
druid.segmentCache.locations=[{"path":"/mnt/data/segment-cache","maxSize":1850000000000 }]
druid.server.maxSize=1850000000000

Router

druid.plaintextPort=11024
druid.router.coordinatorServiceName=druid/coordinator
druid.router.defaultBrokerServiceName=druid/broker-ingestion
druid.router.http.numConnections=100
druid.router.http.numMaxThreads=200
druid.router.http.readTimeout=PT55M
druid.router.managementProxy.enabled=true
druid.router.tierToBrokerMap={"ingestion_tier":"druid/broker-ingestion"}
druid.server.http.numThreads=200
druid.service=druid/router-ingestion

Posted below in the comments more context about how we stress-tested Druid by running parallel scan query over a time interval

[n4j]

Bit of a context, I work with @bharadwajrembar and we are experiencing this issue in Production.

We pull all the rows falling for a given time range (typically a day) via ScanQuery with array as response format.

What we have noticed is the connection closes before all the rows for a given day are fetched and since this is "graceful" no exception is thrown while parsing the response and we experience inconsistent data pull.

We have noticed the above behaviour for a datasource which has about 200 columns and data in tune of 3 million rows for a given day (we have 365 days worth of data), for our other less heavy datasources it's not an issue

To validate our hypothesis I wrote below two scripts, one of them counts the number of rows scanned from the datasource and then compares it against the total number of rows present obtained via count(1)

# Driver script
#!/bin/bash

for (( i=1; i<=$N; i++ ))
do
   #echo $i
    nohup sh -c "time ./sql.sh" > log_$i.log &
done

sleep 3
tail -f log_*.log | grep Consumed

# Executes query agains datasource and counts the number of rows returned and compares it against number of rows actually present
#!/bin/bash

LINES=$(curl --location --request POST 'druid-router.io/druid/v2/sql' \
--header 'Content-Type: application/json' \
--data-raw '
{
    "query": "select  col1, col2, col3, col4, col200   from    \"large-datasource\" where  __time in (?)",
    "resultFormat": "arrayLines",
    "header": "false",
    "parameters":
    [
        {
            "type": "TIMESTAMP",
            "value": "2021-05-01"
        }
    ]
}' | wc -l)

EXPECTED_LINES=$(curl --location --request POST 'druid-router.io/druid/v2/sql' \
--header 'Content-Type: application/json' \
--data-raw '
{
    "query": "select  count(1) from    \"large-datasource\" where __time in (?)",
    "resultFormat": "arrayLines",
    "header": "false",
    "parameters":
    [
        {
            "type": "TIMESTAMP",
            "value": "2021-05-01"
        }
    ]
}' | tail -n 2)

echo "Consumed [$LINES] rows  out of [$EXPECTED_LINES] rows for time range 2021-05-01"

Below is the output of the above script

nohup: redirecting stderr to stdout
nohup: redirecting stderr to stdout
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01

========


Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
**Consumed [3435173] rows  out of [[3696999]] rows for time range 2021-05-01**
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
**Consumed [3435173] rows  out of [[3696999]] rows for time range 2021-05-01**
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
**Consumed [3435173] rows  out of [[3696999]] rows for time range 2021-05-01**
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01
Consumed [3697000] rows  out of [[3696999]] rows for time range 2021-05-01

As you can see from the above we receive less number of rows then present via ScanQuery and since Druid terminates the connection gracefully, we have no way of knowing if we were able to read all the rows.

We have checked our logs and tweaked all timeout parameters but the root cause of this issue remains elusive.