{"record":{"id":"d853f7c6a2be4881","repo":"apache/pulsar","slug":"callbackthreads-must-be-1","errorCode":null,"errorMessage":"callbackThreads must be >= 1","messagePattern":"callbackThreads must be >= 1","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ConnectionPolicy.java","lineNumber":70,"sourceCode":"                             Duration keepAliveInterval,\n                             Duration connectionMaxIdleTime,\n                             int ioThreads,\n                             int callbackThreads,\n                             String proxyServiceUrl,\n                             ProxyProtocol proxyProtocol,\n                             BackoffPolicy connectionBackoff) {\n        Objects.requireNonNull(connectionTimeout, \"connectionTimeout must not be null\");\n        Objects.requireNonNull(keepAliveInterval, \"keepAliveInterval must not be null\");\n        Objects.requireNonNull(connectionMaxIdleTime, \"connectionMaxIdleTime must not be null\");\n        Objects.requireNonNull(connectionBackoff, \"connectionBackoff must not be null\");\n        if (connectionsPerBroker < 1) {\n            throw new IllegalArgumentException(\"connectionsPerBroker must be >= 1\");\n        }\n        if (ioThreads < 1) {\n            throw new IllegalArgumentException(\"ioThreads must be >= 1\");\n        }\n        if (callbackThreads < 1) {\n            throw new IllegalArgumentException(\"callbackThreads must be >= 1\");\n        }\n        this.connectionTimeout = connectionTimeout;\n        this.connectionsPerBroker = connectionsPerBroker;\n        this.enableTcpNoDelay = enableTcpNoDelay;\n        this.keepAliveInterval = keepAliveInterval;\n        this.connectionMaxIdleTime = connectionMaxIdleTime;\n        this.ioThreads = ioThreads;\n        this.callbackThreads = callbackThreads;\n        this.proxyServiceUrl = proxyServiceUrl;\n        this.proxyProtocol = proxyProtocol;\n        this.connectionBackoff = connectionBackoff;\n    }\n\n    /**\n     * @return the maximum duration to wait for a TCP connection to a broker\n     */\n    public Duration connectionTimeout() {\n        return connectionTimeout;","sourceCodeStart":52,"sourceCodeEnd":88,"githubUrl":"https://github.com/apache/pulsar/blob/820761864ed8e2a7d2e52dd9763ad2ae117c1395/pulsar-client-api-v5/src/main/java/org/apache/pulsar/client/api/v5/config/ConnectionPolicy.java#L52-L88","documentation":"ConnectionPolicy requires callbackThreads >= 1 because callbackThreads sizes the executor that runs the client's user-facing callbacks and completion handlers; with zero or negative threads no callback could ever execute. The constructor throws IllegalArgumentException for smaller values.","triggerScenarios":"Calling ConnectionPolicy.builder().callbackThreads(0) or negative; deriving the count from config or a formula that can produce 0, similar to ioThreads sizing errors.","commonSituations":"Copy-pasting a tuning block with callbackThreads(0) intending 'use default'; dividing CPU count in containers with low CPU limits; sentinel-based config parsing mapping 'unset' to 0.","solutions":["Set callbackThreads to a positive number appropriate for your workload (e.g. number of cores).","Clamp computed values with Math.max(1, value) before passing them.","Omit the call to accept the library default."],"exampleFix":"// before\nConnectionPolicy cp = ConnectionPolicy.builder()\n    .callbackThreads(0) // IllegalArgumentException\n    .build();\n\n// after\nConnectionPolicy cp = ConnectionPolicy.builder()\n    .callbackThreads(Runtime.getRuntime().availableProcessors())\n    .build();","handlingStrategy":"validation","validationCode":"int threads = Math.max(1, cfg.callbackThreads());\nConnectionPolicy cp = ConnectionPolicy.builder().callbackThreads(threads).build();","typeGuard":"static boolean isValidCallbackThreads(int v) { return v >= 1; }","tryCatchPattern":"try {\n    cp = ConnectionPolicy.builder().callbackThreads(threads).build();\n} catch (IllegalArgumentException e) {\n    log.warn(\"Invalid callbackThreads, using default\", e);\n    cp = ConnectionPolicy.builder().build();\n}","preventionTips":["Omit callbackThreads to accept the default instead of passing 0.","Clamp any formula-derived thread counts to at least 1.","Validate ioThreads and callbackThreads together since both share the >= 1 rule."],"tags":["java","configuration","threading","callback","illegal-argument"],"backgroundTag":"invalid-configuration-value","analyzedSha":"820761864ed8e2a7d2e52dd9763ad2ae117c1395","analyzedAt":"2026-09-06T00:14:20.138Z","contentChangedAt":"2026-09-06T00:14:20.138Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}