https://linen.dev logo
Join Slack
Powered by
Hi, we are seeing lag on Kafka ingestion during p...
# general
k
Hi, we are seeing lag on Kafka ingestion during peak hours,so we are trying to implement 
numPersistThreads
as it will help during peak times but when i tried to update supervisor spec with this and submit it. After that when i tried to check the payload i am not seeing this parameter . any one faced similar issue
j
What version are you running? Here is the spec from a fairly recent version:
Copy code
"intermediateHandoffPeriod": "P2147483647D",
      "logParseExceptions": true,
      "maxParseExceptions": 2147483647,
      "maxSavedParseExceptions": 0,
      "numPersistThreads": 1,
      "skipSequenceNumberAvailabilityCheck": false,
      "repartitionTransitionDuration": "PT120S"
    }
and in the ingestion task log:
Copy code
... Running with task: {
...
    "segmentWriteOutMediumFactory" : null,
    "intermediateHandoffPeriod" : "P2147483647D",
    "logParseExceptions" : true,
    "maxParseExceptions" : 2147483647,
    "maxSavedParseExceptions" : 0,
    "numPersistThreads" : 1,
    "skipSequenceNumberAvailabilityCheck" : false
  },
k
29.0.1
i am keeping this parameter under tuning config like below
Copy code
"tuningConfig": {
      "type": "kafka",
      "appendableIndexSpec": {
        "type": "onheap",
        "preserveExistingMetrics": false
      },
      "maxRowsInMemory": 200000,
      "maxBytesInMemory": 0,
      "skipBytesInMemoryOverheadCheck": false,
      "maxRowsPerSegment": 5000000,
      "maxTotalRows": null,
      "numPersistThreads": 4,
      "intermediatePersistPeriod": "PT5M",
      "maxPendingPersists": 0,
      "indexSpec": {
        "bitmap": {
          "type": "roaring"
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
j
Okay I just created a new Supervisor and set the numPersistThreads to 2 ... here is what shows up in my ingestion task log file:
Copy code
2024-09-09T15:44:04,730 INFO [main] com.sun.jersey.server.impl.application.WebApplicationImpl - Initiating Jersey application, version 'Jersey: 1.19.4 05/24/2017 03:20 PM'
2024-09-09T15:44:04,730 INFO [task-runner-0-priority-0] org.apache.druid.curator.discovery.CuratorDruidNodeAnnouncer - Announced self [{"druidNode":{"service":"druid/middlemanager","host":"localhost","bindOnHost":false,"plaintextPort":8101,"port":-1,"tlsPort":-1,"enablePlaintextPort":true,"enableTlsPort":false},"nodeType":"peon","services":{"dataNodeService":{"type":"dataNodeService","tier":"_default_tier","maxSize":0,"type":"indexer-executor","serverType":"indexer-executor","priority":0},"lookupNodeService":{"type":"lookupNodeService","lookupTier":"__default"}},"startTime":"2024-09-09T15:44:04.621Z"}].
2024-09-09T15:44:04,741 INFO [task-runner-0-priority-0] org.apache.druid.segment.realtime.appenderator.StreamAppenderator - Number of persist threads [2]
2024-09-09T15:44:04,814 INFO [task-runner-0-priority-0] org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner - Initialized sequences: SequenceMetadata{sequenceId=0, sequenceName='index_kafka_quickstart-events_265296a781a96d3_0', assignments=[KafkaTopicPartition{partition=1, topic='null', multiTopicPartition=false}, KafkaTopicPartition{partition=0,
k
ok .. will we see additional thread on MM or it will be created during persist time and later it will be terminated
i tried to check task log i am not seeing 
org.apache.druid.segment.realtime.appenderator.StreamAppenderator - Number of persist threads
this below is my supervisor spec
Copy code
{
  "type": "kafka",
  "spec": {
    "dataSchema": {
      "dataSource": "kafka_test",
      "timestampSpec": {
        "column": "event_time_ms",
        "format": "auto",
        "missingValue": null
      },
      "dimensionsSpec": {
        "dimensions": [
         {
           ......
           }
         
        ],
        "dimensionExclusions": [
         ......
        ],
        "includeAllDimensions": false,
        "useSchemaDiscovery": false
      },
      "metricsSpec": [
       .....
       
      ],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "HOUR",
        "queryGranularity": "FIVE_MINUTE",
        "rollup": true,
        "intervals": []
      },
      "transformSpec": {
        "filter": null,
        "transforms": []
      }
    },
    "ioConfig": {
      "topic": "TOPIC",
      "topicPattern": null,
      "inputFormat": {
        "type": "avro_ocf",
        "flattenSpec": {
          "useFieldDiscovery": true,
          "fields": []
        }
      },
      "replicas": 1,
      "taskCount": 16,
      "taskDuration": "PT10800S",
      "consumerProperties": {
        "bootstrap.servers": "<MSK>",
        "max.partition.fetch.bytes": 1048576,
        "security.protocol": "SASL_SSL",
        "sasl.mechanism": "AWS_MSK_IAM",
        "sasl.jaas.config": "software.amazon.msk.auth.iam.IAMLoginModule required;",
        "sasl.client.callback.handler.class": "software.amazon.msk.auth.iam.IAMClientCallbackHandler"
      },
      "autoScalerConfig": null,
      "pollTimeout": 100,
      "startDelay": "PT5S",
      "period": "PT30S",
      "useEarliestOffset": false,
      "completionTimeout": "PT3600S",
      "lateMessageRejectionPeriod": "PT1800S",
      "earlyMessageRejectionPeriod": null,
      "lateMessageRejectionStartDateTime": null,
      "configOverrides": null,
      "idleConfig": null,
      "stopTaskCount": 16,
      "stream": "topic",
      "useEarliestSequenceNumber": false
    },
    "tuningConfig": {
      "type": "kafka",
      "appendableIndexSpec": {
        "type": "onheap",
        "preserveExistingMetrics": false
      },
      "maxRowsInMemory": 200000,
      "maxBytesInMemory": 0,
      "skipBytesInMemoryOverheadCheck": false,
      "maxRowsPerSegment": 5000000,
      "maxTotalRows": null,
      "numPersistThreads": 4,
      "intermediatePersistPeriod": "PT5M",
      "maxPendingPersists": 0,
      "indexSpec": {
        "bitmap": {
          "type": "roaring"
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "indexSpecForIntermediatePersists": {
        "bitmap": {
          "type": "roaring"
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "reportParseExceptions": true,
      "handoffConditionTimeout": 600000,
      "resetOffsetAutomatically": true,
      "segmentWriteOutMediumFactory": null,
      "workerThreads": null,
      "chatRetries": 8,
      "httpTimeout": "PT10S",
      "shutdownTimeout": "PT80S",
      "offsetFetchPeriod": "PT30S",
      "intermediateHandoffPeriod": "P2147483647D",
      "logParseExceptions": false,
      "maxParseExceptions": 0,
      "maxSavedParseExceptions": 0,
      "skipSequenceNumberAvailabilityCheck": false,
      "repartitionTransitionDuration": "PT120S"
    }
  },
  "context": null,
  "suspended": false
}
not sure for any issue with the spec . Can you please share your spec for reference
j
I am using a current Imply version as of 2024 August though ... let me see if I can pull up a version that is close to Druid 29 in release date ... I'll try 2024.01 ...
k
ok .. thanks
j
okay the version from 2024.01 does not have the parameter ... I can put it in the supervisor spec but it doesn't use it or carry it over to the ingestion task. This looks like the PR that implemented it: https://github.com/apache/druid/pull/13982/files
k
ok .. thanks @John Kowtko for testing it . which version i need to download and can i just upgrade specific MM where i need this parameter and have rest of the cluster is in current version ?
j
I am not sure. if you are on Druid 29, can you check Druid 30 to see if it is in there?
k
from the PR looks like it is available in Druid 30
2 Views
Open in Slack
PreviousNext
Powered by