Page History

public class MyInterceptor<K, V> implements ProducerInterceptor<K, V>, Monitorable {

    private Sensor sensor;
    private PluginMetrics metrics;

    public void withPluginMetrics(PluginMetrics metrics) {
        this.metrics = metrics;
        sensor = metrics.sensor("onSend");
        MetricName rate = metrics.metricName("rate", "Average number of calls per second.", Collections.emptyMap());
        MetricName total = metrics.metricName("total", "Total number of calls.", Collections.emptyMap());
        sensor.add(rate, new Rate());
        sensor.add(total, new CumulativeCount());
    }

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        sensor.record();
        return record;
    }

    @Override
    public void close() {
        try {
            if (metrics != null) metrics.close();
        } catch (IOException e) { // Even though ProducerInterceptor extends AutoCloseable which has "void close() throws Exception;", ProducerInterceptor has its own close() method without "throws"!
            throw new RuntimeException(e);
        }
    }
}