@Override
    public SystemInfoResult getSystemInfo() {

        final List<Map<String, Object>> producerInfo = allSharedProducersMap.values().stream()
                .map(sharedKafkaProducer -> {
                    Map<String, Object> map = new HashMap<>();
                    map.put("docName", sharedKafkaProducer.getConfigName());
                    map.put("docUuid", sharedKafkaProducer.getConfigUuid());
                    map.put("docVersion", sharedKafkaProducer.getConfigVersion());
                    map.put("useCount", sharedKafkaProducer.getUseCount());
                    map.put("isSuperseded", sharedKafkaProducer.isSuperseded());
                    map.put("createdTime", sharedKafkaProducer.getCreatedTime().toString());
                    map.put("lastAccessedTime", sharedKafkaProducer.getLastAccessedTime().toString());

                    sharedKafkaProducer.getKafkaProducer().ifPresent(kafkaProducer -> {
                        final Map<String, Map<String, Object>> metrics = kafkaProducer.metrics()
                                .entrySet()
                                .stream()
                                .map(entry -> {
                                    final String groupName = entry.getKey().group()
                                            + " ("
                                            + entry.getKey().tags().entrySet()
                                            .stream()
                                            .filter(entry2 -> !entry2.getKey().equals("client-id"))
                                            .map(entry2 -> entry2.getKey() + "=" + entry2.getValue())
                                            .collect(Collectors.joining(","))
                                            + ")";

                                    return Tuple.of(
                                            groupName,
                                            entry.getKey().name(),
                                            entry.getValue().metricValue());
                                })
                                .collect(Collectors.groupingBy(Tuple3::_1, Collectors.toMap(
                                        Tuple3::_2,
                                        Tuple3::_3)));
                        map.put("kafkaProducerMetrics", metrics);
                    });
                    return map;
                })
                .toList();

        return SystemInfoResult.builder(this)
                .addDetail("sharedProducers", producerInfo)
                .build();
    }
Class to provide shared {@link KafkaProducer} instances wrapped inside a {@link SharedKafkaProducer}.
We want one {@link KafkaProducer} per {@link KafkaConfigDoc}, however the config in a {@link KafkaConfigDoc}
can change at runtime so on each request for a {@link SharedKafkaProducer} we need to check if the
{@link KafkaConfigDoc} has changed since the time we created it. New calls to
{@link KafkaProducerFactory#getSharedProducer(DocRef)} will always get a {@link SharedKafkaProducer} for the
latest config. {@link KafkaProducer}
stroom-kafka/stroom-kafka-impl/src/main/java/stroom/kafka/impl/KafkaProducerFactoryImpl.java