{"record":{"id":"035483a83f9adda7","repo":"apache/druid","slug":"fetchthreads-d-being-lowered-to-d","errorCode":null,"errorMessage":"fetchThreads [%d] being lowered to [%d]","messagePattern":"fetchThreads \\[(.+?)\\] being lowered to \\[(.+?)\\]","errorType":"console","errorClass":null,"httpStatus":null,"severity":"info","filePath":"extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java","lineNumber":216,"sourceCode":"      fetchThreads = runtimeInfo.getAvailableProcessors() * 2;\n    }\n\n    // Each fetchThread can return upto 10MB at a time\n    // (https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html), cap fetchThreads so that\n    // we don't exceed more than the least of 100MB or 5% of heap at a time. Don't fail if fetchThreads specified\n    // is greater than this as to not cause failure for older configurations, but log warning in this case, and lower\n    // fetchThreads implicitly.\n    final long memoryToUse = Math.min(\n        KinesisIndexTaskIOConfig.MAX_RECORD_FETCH_MEMORY,\n        (long) (runtimeInfo.getMaxHeapSizeBytes() * KinesisIndexTaskIOConfig.RECORD_FETCH_MEMORY_MAX_HEAP_FRACTION)\n    );\n    int maxFetchThreads = Math.max(\n        1,\n        (int) (memoryToUse / GET_RECORDS_MAX_BYTES_PER_CALL)\n    );\n    if (fetchThreads > maxFetchThreads) {\n      if (configuredFetchThreads != null) {\n        log.warn(\"fetchThreads [%d] being lowered to [%d]\", configuredFetchThreads, maxFetchThreads);\n      }\n      fetchThreads = maxFetchThreads;\n    }\n\n    Preconditions.checkArgument(\n        fetchThreads > 0,\n        \"Must have at least one background fetch thread for the record supplier\"\n    );\n\n    return fetchThreads;\n  }\n}\n","sourceCodeStart":198,"sourceCodeEnd":229,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java#L198-L229","documentation":"KinesisIndexTask.computeFetchThreads() caps the number of fetch threads by available memory: maxFetchThreads = memoryToUse / GET_RECORDS_MAX_BYTES_PER_CALL. When the configured value exceeds this cap, it logs 'fetchThreads [%d] being lowered to [%d]' and silently reduces fetchThreads to the memory-safe maximum.","triggerScenarios":"Configuring a large fetchThreads in the Kinesis tuning config while the task's memory budget (divided by GET_RECORDS_MAX_BYTES_PER_CALL, max bytes per GetRecords call) allows fewer threads.","commonSituations":"Setting fetchThreads=high on small task containers (low -Xmx or task memory); memory reductions after right-sizing; copied configs from larger clusters.","solutions":["Lower fetchThreads in the tuning config to a value within the memory-derived cap.","Increase the task's memory allocation (worker capacity / task runtime properties) if more threads are genuinely needed.","Rely on the automatic cap if acceptable — behavior stays correct, just fewer threads.","Recalculate: fetchThreads must be <= taskMemoryBytes / GET_RECORDS_MAX_BYTES_PER_CALL (and > 0)."],"exampleFix":"// before\n\"tuningConfig\": { \"type\": \"kinesis\", \"fetchThreads\": 20 } // task has 2GB\n// after\n\"tuningConfig\": { \"type\": \"kinesis\", \"fetchThreads\": 4 } // <= memory / GET_RECORDS_MAX_BYTES_PER_CALL","handlingStrategy":"validation","validationCode":"int max = (int) (taskMemoryBytes / GET_RECORDS_MAX_BYTES_PER_CALL);\nif (fetchThreads > max) { fetchThreads = max; } // pre-clamp before submitting spec","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Size fetchThreads against the task container memory","Prefer defaults over large explicit fetchThreads","Right-size memory before increasing parallelism"],"tags":["kinesis","memory","tuning-config","fetch-threads"],"backgroundTag":"invalid-config-value","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-17T15:17:12.973Z"}