public String metricsJson(Reconciliation reconciliation, ConfigMap configMap) {
        if (isEnabled)  {
            if (configMap == null) {
                LOGGER.warnCr(reconciliation, "ConfigMap {} does not exist.", configMapName);
                throw new InvalidConfigurationException("ConfigMap " + configMapName + " does not exist");
            } else {
                String data = configMap.getData().get(configMapKey);

                if (data == null) {
                    LOGGER.warnCr(reconciliation, "ConfigMap {} does not contain specified key {}.", configMapName, configMapKey);
                    throw new InvalidConfigurationException("ConfigMap " + configMapName + " does not contain specified key " + configMapKey);
                } else {
                    if (data.isEmpty()) {
                        return "{}";
                    }

                    try {
                        ObjectMapper yamlReader = new ObjectMapper(new YAMLFactory());
                        Object yaml = yamlReader.readValue(data, Object.class);
                        ObjectMapper jsonWriter = new ObjectMapper();

                        return jsonWriter.writeValueAsString(yaml);
                    } catch (JsonProcessingException e) {
                        throw new InvalidConfigurationException("Failed to parse metrics configuration", e);
                    }
                }
            }
        } else {
            return null;
        }
    }
@Test
    public void testProblemWithConfigMap()   {
        MetricsConfig metricsConfig = new JmxPrometheusExporterMetricsBuilder()
                .withNewValueFrom()
                .withConfigMapKeyRef(new ConfigMapKeySelector("my-key", "my-name", false))
                .endValueFrom()
                .build();

        MetricsModel metrics = new MetricsModel(new KafkaConnectSpecBuilder().withMetricsConfig(metricsConfig).build());

        InvalidConfigurationException ex = assertThrows(InvalidConfigurationException.class, () -> metrics.metricsJson(Reconciliation.DUMMY_RECONCILIATION, null));
        assertThat(ex.getMessage(), is("ConfigMap my-name does not exist"));

        ex = assertThrows(InvalidConfigurationException.class, () -> metrics.metricsJson(Reconciliation.DUMMY_RECONCILIATION, new ConfigMapBuilder().withData(Map.of("other-key", "foo: bar")).build()));
        assertThat(ex.getMessage(), is("ConfigMap my-name does not contain specified key my-key"));

        ex = assertThrows(InvalidConfigurationException.class, () -> metrics.metricsJson(Reconciliation.DUMMY_RECONCILIATION, new ConfigMapBuilder().withData(Map.of("my-key", "foo: -")).build()));
        assertThat(ex.getMessage(), startsWith("Failed to parse metrics configuration"));
    }