{"record":{"id":"9f58408360144f93","repo":"t8y2/dbx","slug":"kafka-broker-does-not-support-op-optype","errorCode":null,"errorMessage":"Kafka broker does not support \" + op.opType() + \" config operations through the legacy alterConfigs API","messagePattern":"Kafka broker does not support \" \\+ op\\.opType\\(\\) \\+ \" config operations through the legacy alterConfigs API","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java","lineNumber":957,"sourceCode":"        return Collections.singletonMap(\"ok\", true);\n    }\n\n    static Map<String, String> legacyTopicConfig(Config current, List<AlterConfigOp> ops) {\n        Map<String, String> values = new LinkedHashMap<>();\n        for (ConfigEntry entry : current.entries()) {\n            boolean topicOverride = entry.source() == ConfigEntry.ConfigSource.DYNAMIC_TOPIC_CONFIG\n                || entry.source() == ConfigEntry.ConfigSource.UNKNOWN;\n            if (topicOverride && !entry.isReadOnly() && !entry.isSensitive() && entry.value() != null) {\n                values.put(entry.name(), entry.value());\n            }\n        }\n\n        for (AlterConfigOp op : ops) {\n            String key = op.configEntry().name();\n            switch (op.opType()) {\n                case SET -> values.put(key, op.configEntry().value());\n                case DELETE -> values.remove(key);\n                case APPEND, SUBTRACT -> throw new IllegalArgumentException(\n                    \"Kafka broker does not support \" + op.opType() + \" config operations through the legacy alterConfigs API\"\n                );\n            }\n        }\n        return values;\n    }\n\n    // -----------------------------------------------------------------------\n    // Consumer groups\n    // -----------------------------------------------------------------------\n\n    private static Object listConsumerGroups(JsonObject params) throws Exception {\n        AdminClient admin = requireAdmin();\n        int timeout = requestTimeout(params);\n        String filterTopic = stringOrEmpty(params, \"topic\");\n\n        Collection<ConsumerGroupListing> groups = admin.listConsumerGroups(\n                new ListConsumerGroupsOptions().timeoutMs(timeout))","sourceCodeStart":939,"sourceCodeEnd":975,"githubUrl":"https://github.com/t8y2/dbx/blob/c0390bff16418b651f4728520d99adf8ce48829a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java#L939-L975","documentation":"When altering topic/broker configs, the agent may fall back to the legacy AdminClient.alterConfigs API, which only supports SET and DELETE semantics — it replaces the whole config rather than incrementally appending/subtracting values. If a caller submits an AlterConfigOp with opType APPEND or SUBTRACT through this path, the agent throws IllegalArgumentException because legacy brokers cannot express those operations.","triggerScenarios":"Calling the agent's config-alter operation (incrementalAlterConfigs) against a broker that predates incrementalAlterConfigs (Kafka < 2.3) or where the incremental API is unavailable, with ops containing APPEND or SUBTRACT op types.","commonSituations":"Running against an old Kafka broker or a managed service/mirror that does not expose the incremental alter API; code written for modern Kafka (using append/subtract to grow list-type configs like allowed.retention.ms lists) reused unchanged against a legacy cluster.","solutions":["Upgrade the Kafka brokers to >= 2.3 so incrementalAlterConfigs supports APPEND/SUBTRACT natively.","Rewrite the operation as SET/DELETE: read the current config value, compute the resulting list locally, and submit a single SET op with the full new value.","Gate the code path: detect broker capability and only send APPEND/SUBTRACT when the incremental API is available; otherwise error before sending ops.","Remove APPEND/SUBTRACT ops from the request if the target value can be expressed as a plain SET."],"exampleFix":"// before\nops.add(new AlterConfigOp(new ConfigEntry(\"metric.reporters\", \"r2\"), AlterConfigOp.OpType.APPEND)); // legacy broker\n// after\nString current = describeConfig(\"metric.reporters\");\nops.add(new AlterConfigOp(new ConfigEntry(\"metric.reporters\", current + \",r2\"), AlterConfigOp.OpType.SET));","handlingStrategy":"fallback","validationCode":null,"typeGuard":null,"tryCatchPattern":null,"preventionTips":[],"tags":["kafka","config","compatibility","unsupported-operation"],"backgroundTag":"unsupported-config-operation","analyzedSha":"c0390bff16418b651f4728520d99adf8ce48829a","analyzedAt":"2026-09-05T23:05:10.900Z","contentChangedAt":"2026-09-05T23:05:10.900Z","schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}