{"record":{"id":"1dd0b72df100a5c6","repo":"apache/kafka","slug":"when-the-security-protocol-configuration-enables-s","errorCode":null,"errorMessage":"When the security.protocol configuration enables SASL, mechanism must be non-null and non-empty string.","messagePattern":"When the security\\.protocol configuration enables SASL, mechanism must be non-null and non-empty string\\.","errorType":"validation","errorClass":"ConfigException","httpStatus":null,"severity":"error","filePath":"clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java","lineNumber":322,"sourceCode":"                RETRY_BACKOFF_MAX_MS_CONFIG, retryBackoffMaxMs, retryBackoffMaxMs);\n        }\n\n        long connectionSetupTimeoutMs = config.getLong(SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG);\n        long connectionSetupTimeoutMaxMs = config.getLong(SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG);\n        if (connectionSetupTimeoutMs > connectionSetupTimeoutMaxMs) {\n            log.warn(\"Configuration '{}' with value '{}' is greater than configuration '{}' with value '{}'. \" +\n                    \"A static connection setup timeout with value '{}' will be applied.\",\n                SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG, connectionSetupTimeoutMs,\n                SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG, connectionSetupTimeoutMaxMs, connectionSetupTimeoutMaxMs);\n        }\n    }\n\n    public static void postValidateSaslMechanismConfig(AbstractConfig config) {\n        SecurityProtocol securityProtocol = SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));\n        String clientSaslMechanism = config.getString(SaslConfigs.SASL_MECHANISM);\n        if (securityProtocol == SecurityProtocol.SASL_PLAINTEXT || securityProtocol == SecurityProtocol.SASL_SSL) {\n            if (clientSaslMechanism == null || clientSaslMechanism.isEmpty()) {\n                throw new ConfigException(SaslConfigs.SASL_MECHANISM, null, \"When the \" + CommonClientConfigs.SECURITY_PROTOCOL_CONFIG +\n                        \" configuration enables SASL, mechanism must be non-null and non-empty string.\");\n            }\n        }\n    }\n\n    public static List<MetricsReporter> metricsReporters(AbstractConfig config) {\n        return metricsReporters(Collections.emptyMap(), config);\n    }\n\n    public static List<MetricsReporter> metricsReporters(String clientId, AbstractConfig config) {\n        return metricsReporters(Collections.singletonMap(CommonClientConfigs.CLIENT_ID_CONFIG, clientId), config);\n    }\n\n    public static List<MetricsReporter> metricsReporters(Map<String, Object> clientIdOverride, AbstractConfig config) {\n        return config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,\n                MetricsReporter.class, clientIdOverride);\n    }\n","sourceCodeStart":304,"sourceCodeEnd":340,"githubUrl":"https://github.com/apache/kafka/blob/996fb4585aa1bcc8980b0e1b8d6b168b986cd979/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java#L304-L340","documentation":"ConfigException thrown by CommonClientConfigs.postValidateSaslMechanismConfig when the security.protocol is SASL_PLAINTEXT or SASL_SSL but sasl.mechanism is null or empty. The post-validation runs after the config is parsed, ensuring a SASL-enabled protocol always has a concrete mechanism to negotiate. This is a fail-fast guard against an incomplete SASL setup.","triggerScenarios":"At line 322, after reading security.protocol and sasl.mechanism, if the protocol is one of the two SASL variants and clientSaslMechanism is null/empty, throw ConfigException naming SASL_MECHANISM. Triggered by setting security.protocol=SASL_SSL/SASL_PLAINTEXT without sasl.mechanism.","commonSituations":"Switching from PLAINTEXT/SSL to SASL_SSL and forgetting sasl.mechanism; jaas config present but mechanism missing; templated config that leaves mechanism blank; mechanism supplied only in a client-specific override that did not apply.","solutions":["Set sasl.mechanism to a supported value, e.g. 'SCRAM-SHA-512', 'PLAIN', 'GSSAPI', or 'OAUTHBEARER'.","Ensure the matching jaas config and login handler are configured for that mechanism.","If you did not intend SASL, set security.protocol back to PLAINTEXT or SSL.","Verify the override chain (global vs producer/consumer/admin) actually sets the mechanism."],"exampleFix":"# before\nsecurity.protocol=SASL_SSL\n# sasl.mechanism missing\n# after\nsecurity.protocol=SASL_SSL\nsasl.mechanism=SCRAM-SHA-512\nsasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \\\n  username=\"user\" password=\"secret\";","handlingStrategy":"validation","validationCode":"SecurityProtocol sp = SecurityProtocol.forName(config.getString(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));\nString mech = config.getString(SaslConfigs.SASL_MECHANISM);\nif ((sp == SecurityProtocol.SASL_PLAINTEXT || sp == SecurityProtocol.SASL_SSL) && (mech == null || mech.isEmpty()))\n  throw new ConfigException(SaslConfigs.SASL_MECHANISM, null, \"SASL protocol requires a non-empty mechanism\");","typeGuard":null,"tryCatchPattern":"try { CommonClientConfigs.postValidateSaslMechanismConfig(config); } catch (ConfigException e) { if (e.message.contains('mechanism must be non-null')) { /* set sasl.mechanism */ } else throw e; }","preventionTips":["Whenever security.protocol is SASL_*, also set sasl.mechanism.","Keep security.protocol and sasl.mechanism in the same config block to avoid partial overrides.","Include a JAAS config matching the chosen mechanism."],"tags":["kafka-client","config","security","sasl"],"backgroundTag":null,"analyzedSha":"996fb4585aa1bcc8980b0e1b8d6b168b986cd979","analyzedAt":"2026-08-11T22:03:28.655Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}