apache/kafka · error · IllegalArgumentException

Sensor {name} already exists

Error message

Sensor {name} already exists

What it means

Thrown by PluginMetricsImpl.addSensor when a sensor with the given name was already added through this same PluginMetrics handle (the 'sensors' set contains the name). Each handle maintains its own namespace; within it, sensor names must be unique. The error guards against silently overwriting or double-registering a sensor which would confuse metric reporting and cleanup.

Source

Thrown at clients/src/main/java/org/apache/kafka/common/metrics/internals/PluginMetricsImpl.java:85

        metricNames.add(metricName);
    }

    @Override
    public void removeMetric(MetricName metricName) {
        if (closing) throw new IllegalStateException("This PluginMetrics instance is closed");
        if (metricNames.contains(metricName)) {
            metrics.removeMetric(metricName);
            metricNames.remove(metricName);
        } else {
            throw new IllegalArgumentException("Unknown metric " + metricName);
        }
    }

    @Override
    public Sensor addSensor(String name) {
        if (closing) throw new IllegalStateException("This PluginMetrics instance is closed");
        if (sensors.contains(name)) {
            throw new IllegalArgumentException("Sensor " + name + " already exists");
        }
        Sensor sensor = metrics.sensor(name);
        sensors.add(name);
        return sensor;
    }

    @Override
    public void removeSensor(String name) {
        if (closing) throw new IllegalStateException("This PluginMetrics instance is closed");
        if (sensors.contains(name)) {
            metrics.removeSensor(name);
            sensors.remove(name);
        } else {
            throw new IllegalArgumentException("Unknown sensor " + name);
        }
    }

    @Override

View on GitHub (pinned to c31c9215e1)

Solutions

1. Make sensor names unique per instance — include task id, partition, or a counter in the name (e.g. "records-" + taskId).
2. Cache the created Sensor in a field and reuse it; check sensor != null before calling addSensor.
3. Call removeSensor(name) before re-adding, or rely on PluginMetrics.close() to clear the namespace when the plugin instance is retired.

Example fix

// before
sensor = pluginMetrics.addSensor("records"); // throws on 2nd task
// after
sensor = pluginMetrics.addSensor("records-" + taskId);

// Maintain a Set<String> of sensor names you created.
private final Set<String> registeredSensors = ConcurrentHashMap.newKeySet();
if (name == null || name.isEmpty()) throw new IllegalArgumentException("sensor name required");
if (!registeredSensors.add(name)) {
    return registeredSensorRefs.get(name); // already exists, reuse
}
Sensor s = pluginMetrics.addSensor(name);
registeredSensorRefs.put(name, s);
return s;

When it happens

Trigger: Calling pluginMetrics.addSensor("foo") twice on the same handle without an intervening removeSensor("foo"). Common when a plugin is initialized multiple times (e.g. multiple Connect tasks in one worker sharing a plugin instance) or when sensor names are templated from a non-unique source.

Common situations: Connect converters/transformations instantiated per-task but sharing a PluginMetrics; plugins using a fixed sensor name like "records" across multiple parallel instances; re-init after a transient failure that does not run stop().

Related errors

• Unknown metric {metricName}
• Unknown sensor {name}
• Telemetry is not enabled. Set config `enable.metrics.push` t
• Error creating mbean attribute for metricName :{metricName}
• Error unregistering mbean

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/b681caa81d017172.json. Report an issue: GitHub.