From 346375cc9dcb427db8d84293f6ab2a55322bba90 Mon Sep 17 00:00:00 2001 From: Diogo Pereira Date: Wed, 6 May 2026 16:38:46 +0100 Subject: [PATCH 1/3] fix(topology_lag): Kafka Topology Lag breaking when no offsets are commited for topic/partition --- .../kafka/monitor/KafkaOffsetLagUtil.java | 12 +++++++- .../apache/storm/utils/TopologySpoutLag.java | 30 +++++++++++++++---- 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java index fa06ffa3e76..7fc85f0ff6e 100644 --- a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java +++ b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java @@ -174,7 +174,7 @@ public static List getOffsetLags(NewKafkaSpoutOffsetQuery consumer.assign(topicPartitionList); for (TopicPartition topicPartition : topicPartitionList) { Map offsetAndMetadata = consumer.committed(Collections.singleton(topicPartition)); - long committedOffset = offsetAndMetadata != null ? offsetAndMetadata.get(topicPartition).offset() : -1; + long committedOffset = resolveCommittedOffset(offsetAndMetadata, topicPartition); consumer.seekToEnd(toArrayList(topicPartition)); result.add(new KafkaOffsetLagResult(topicPartition.topic(), topicPartition.partition(), committedOffset, consumer.position(topicPartition))); @@ -187,6 +187,16 @@ public static List getOffsetLags(NewKafkaSpoutOffsetQuery return result; } + /** + * Read the committed offset for a partition out of the map returned by + * {@link org.apache.kafka.clients.consumer.KafkaConsumer#committed(java.util.Set)}. + * Returns {@code -1} when no offset is committed (the map's value for the partition is null). + */ + private static long resolveCommittedOffset(Map committedOffsets, TopicPartition topicPartition) { + OffsetAndMetadata partitionOffset = committedOffsets.get(topicPartition); + return partitionOffset != null ? partitionOffset.offset() : -1; + } + private static Collection toArrayList(final TopicPartition tp) { return new ArrayList(1) { { diff --git a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java index 4859552eb0f..76d9af008b8 100644 --- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java +++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java @@ -170,11 +170,9 @@ private static Map getLagResultForKafka(String spoutId, SpoutSpe try { String resultFromMonitor = new ShellCommandRunnerImpl().execCommand(commands.toArray(new String[0])); - try { - result = (Map) JSONValue.parseWithException(resultFromMonitor); - } catch (ParseException e) { - LOGGER.debug("JSON parsing failed, assuming message as error message: {}", resultFromMonitor); - // json parsing fail -> error received + result = parseMonitorOutput(resultFromMonitor); + if (result == null) { + // parsing failed or did not yield a Map -> treat output as error message errorMsg = resultFromMonitor; } } finally { @@ -201,4 +199,26 @@ private static Map getLagResultForKafka(String spoutId, SpoutSpe private static Map getLagResultForNewKafkaSpout(String spoutId, SpoutSpec spoutSpec) throws IOException { return getLagResultForKafka(spoutId, spoutSpec); } + + /** + * Parse the stdout from {@code storm-kafka-monitor}. Returns the parsed JSON map on success, + * or {@code null} when the output is not parseable as a JSON object — which happens when the + * monitor printed a plaintext error string (json-smart parses unquoted text leniently as a + * String rather than throwing). The caller treats {@code null} as "monitor failed; surface + * the raw stdout as the error message". + */ + @SuppressWarnings("unchecked") + private static Map parseMonitorOutput(String resultFromMonitor) { + try { + Object parsed = JSONValue.parseWithException(resultFromMonitor); + if (parsed instanceof Map) { + return (Map) parsed; + } + LOGGER.debug("JSON parsing did not yield a Map, assuming message as error message: {}", resultFromMonitor); + return null; + } catch (ParseException e) { + LOGGER.debug("JSON parsing failed, assuming message as error message: {}", resultFromMonitor); + return null; + } + } } From 87827f0c6fb6ff2e8ad2f856a860f0f0164a3596 Mon Sep 17 00:00:00 2001 From: Diogo Pereira Date: Wed, 6 May 2026 22:02:37 +0100 Subject: [PATCH 2/3] add integration test --- external/storm-kafka-monitor/pom.xml | 12 ++ .../kafka/monitor/KafkaOffsetLagUtil.java | 45 +++---- .../kafka/monitor/KafkaOffsetLagUtilTest.java | 112 ++++++++++++++++++ .../apache/storm/utils/TopologySpoutLag.java | 35 ++---- 4 files changed, 148 insertions(+), 56 deletions(-) create mode 100644 external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java diff --git a/external/storm-kafka-monitor/pom.xml b/external/storm-kafka-monitor/pom.xml index 16046f25270..ca31562e543 100644 --- a/external/storm-kafka-monitor/pom.xml +++ b/external/storm-kafka-monitor/pom.xml @@ -65,6 +65,18 @@ jakarta.xml.bind-api + + org.testcontainers + kafka + ${testcontainers.version} + test + + + org.testcontainers + junit-jupiter + ${testcontainers.version} + test + diff --git a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java index 7fc85f0ff6e..6d48c3616c0 100644 --- a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java +++ b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java @@ -19,9 +19,8 @@ package org.apache.storm.kafka.monitor; import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Properties; @@ -69,10 +68,10 @@ public static void main(String[] args) { printUsageAndExit(options, OPTION_GROUP_ID_LONG + " and " + OPTION_BOOTSTRAP_BROKERS_LONG + " are required"); } NewKafkaSpoutOffsetQuery newKafkaSpoutOffsetQuery = - new NewKafkaSpoutOffsetQuery(commandLine.getOptionValue(OPTION_TOPIC_LONG), - commandLine.getOptionValue(OPTION_BOOTSTRAP_BROKERS_LONG), - commandLine.getOptionValue(OPTION_GROUP_ID_LONG), securityProtocol, saslMechanism, - commandLine.getOptionValue(OPTION_CONSUMER_CONFIG_LONG)); + new NewKafkaSpoutOffsetQuery(commandLine.getOptionValue(OPTION_TOPIC_LONG), + commandLine.getOptionValue(OPTION_BOOTSTRAP_BROKERS_LONG), + commandLine.getOptionValue(OPTION_GROUP_ID_LONG), securityProtocol, saslMechanism, + commandLine.getOptionValue(OPTION_CONSUMER_CONFIG_LONG)); List results = getOffsetLags(newKafkaSpoutOffsetQuery); Map> keyedResult = keyByTopicAndPartition(results); @@ -84,7 +83,7 @@ public static void main(String[] args) { } private static Map> keyByTopicAndPartition( - List results) { + List results) { Map> resultKeyedByTopic = new HashMap<>(); for (KafkaOffsetLagResult result : results) { @@ -96,7 +95,7 @@ private static Map> keyByTopicAndP } topicResultKeyedByPartition.put(result.getPartition(), - new KafkaPartitionOffsetLag(result.getConsumerCommittedOffset(), result.getLogHeadOffset())); + new KafkaPartitionOffsetLag(result.getConsumerCommittedOffset(), result.getLogHeadOffset())); } return resultKeyedByTopic; @@ -114,7 +113,7 @@ private static Options buildOptions() { options.addOption(OPTION_TOPIC_SHORT, OPTION_TOPIC_LONG, true, "REQUIRED Topics (comma separated list) for fetching log head and spout committed " - + "offset"); + + "offset"); options.addOption(OPTION_BOOTSTRAP_BROKERS_SHORT, OPTION_BOOTSTRAP_BROKERS_LONG, true, "Comma separated list of bootstrap broker hosts for new " @@ -137,6 +136,7 @@ private static Options buildOptions() { /** * Get offset lags. + * * @param newKafkaSpoutOffsetQuery represents the information needed to query kafka for log head and spout offsets * @return log head offset, spout offset and lag for each partition */ @@ -172,12 +172,13 @@ public static List getOffsetLags(NewKafkaSpoutOffsetQuery } } consumer.assign(topicPartitionList); + Map committedOffsets = consumer.committed(new HashSet<>(topicPartitionList)); + consumer.seekToEnd(topicPartitionList); for (TopicPartition topicPartition : topicPartitionList) { - Map offsetAndMetadata = consumer.committed(Collections.singleton(topicPartition)); - long committedOffset = resolveCommittedOffset(offsetAndMetadata, topicPartition); - consumer.seekToEnd(toArrayList(topicPartition)); + OffsetAndMetadata partitionOffset = committedOffsets.get(topicPartition); + long committedOffset = partitionOffset != null ? partitionOffset.offset() : -1; result.add(new KafkaOffsetLagResult(topicPartition.topic(), topicPartition.partition(), committedOffset, - consumer.position(topicPartition))); + consumer.position(topicPartition))); } } finally { if (consumer != null) { @@ -187,22 +188,4 @@ public static List getOffsetLags(NewKafkaSpoutOffsetQuery return result; } - /** - * Read the committed offset for a partition out of the map returned by - * {@link org.apache.kafka.clients.consumer.KafkaConsumer#committed(java.util.Set)}. - * Returns {@code -1} when no offset is committed (the map's value for the partition is null). - */ - private static long resolveCommittedOffset(Map committedOffsets, TopicPartition topicPartition) { - OffsetAndMetadata partitionOffset = committedOffsets.get(topicPartition); - return partitionOffset != null ? partitionOffset.offset() : -1; - } - - private static Collection toArrayList(final TopicPartition tp) { - return new ArrayList(1) { - { - add(tp); - } - }; - } - } diff --git a/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java b/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java new file mode 100644 index 00000000000..509ec808e47 --- /dev/null +++ b/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.kafka.monitor; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.StringDeserializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.kafka.KafkaContainer; +import org.testcontainers.utility.DockerImageName; + +@Testcontainers +class KafkaOffsetLagUtilTest { + + private static final String TOPIC = "lag-test-topic"; + private static final String GROUP_ID = "lag-test-group"; + private static final int PARTITIONS = 2; + private static final long COMMITTED_OFFSET_PARTITION_0 = 7L; + + @Container + private static final KafkaContainer KAFKA = new KafkaContainer(DockerImageName.parse("apache/kafka:4.0.0")); + + @BeforeAll + static void seedKafka() throws Exception { + Properties adminProps = new Properties(); + adminProps.put("bootstrap.servers", KAFKA.getBootstrapServers()); + try (Admin admin = Admin.create(adminProps)) { + admin.createTopics(Collections.singletonList(new NewTopic(TOPIC, PARTITIONS, (short) 1))).all().get(); + } + produceOneRecordToEachPartition(); + commitOffsetForPartitionZeroOnly(); + } + + @Test + void getOffsetLagsReportsCommittedOffsetForCommittedPartitionsAndMinusOneForUncommittedPartitions() throws Exception { + NewKafkaSpoutOffsetQuery query = new NewKafkaSpoutOffsetQuery( + TOPIC, KAFKA.getBootstrapServers(), GROUP_ID, null, null, null); + + List results = KafkaOffsetLagUtil.getOffsetLags(query); + + assertEquals(PARTITIONS, results.size(), "Expected one result per partition"); + Map committedByPartition = new HashMap<>(); + for (KafkaOffsetLagResult r : results) { + committedByPartition.put(r.getPartition(), r.getConsumerCommittedOffset()); + } + assertNotNull(committedByPartition.get(0)); + assertEquals(COMMITTED_OFFSET_PARTITION_0, committedByPartition.get(0), + "Partition with a committed offset should report it"); + // Regression assertion: pre-fix this NPE'd inside the monitor and surfaced as ClassCastException upstream. + assertEquals(-1L, committedByPartition.get(1), + "Partition with no committed offset should report -1, not throw"); + } + + private static void produceOneRecordToEachPartition() { + Properties producerProps = new Properties(); + producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers()); + producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + try (KafkaProducer producer = new KafkaProducer<>(producerProps)) { + for (int p = 0; p < PARTITIONS; p++) { + producer.send(new ProducerRecord<>(TOPIC, p, "k", "v")); + } + producer.flush(); + } + } + + private static void commitOffsetForPartitionZeroOnly() { + Properties consumerProps = new Properties(); + consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA.getBootstrapServers()); + consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID); + consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); + try (KafkaConsumer consumer = new KafkaConsumer<>(consumerProps)) { + TopicPartition p0 = new TopicPartition(TOPIC, 0); + consumer.commitSync(Collections.singletonMap(p0, new OffsetAndMetadata(COMMITTED_OFFSET_PARTITION_0))); + } + } +} diff --git a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java index 76d9af008b8..65ab2323dc9 100644 --- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java +++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java @@ -170,9 +170,16 @@ private static Map getLagResultForKafka(String spoutId, SpoutSpe try { String resultFromMonitor = new ShellCommandRunnerImpl().execCommand(commands.toArray(new String[0])); - result = parseMonitorOutput(resultFromMonitor); - if (result == null) { - // parsing failed or did not yield a Map -> treat output as error message + try { + Object parsed = JSONValue.parseWithException(resultFromMonitor); + if (parsed instanceof Map) { + result = (Map) parsed; + } else { + // json-smart parses unquoted plain text leniently as a String, so we can land here + // when the monitor printed an error message instead of JSON; surface it as the error. + errorMsg = resultFromMonitor; + } + } catch (ParseException e) { errorMsg = resultFromMonitor; } } finally { @@ -199,26 +206,4 @@ private static Map getLagResultForKafka(String spoutId, SpoutSpe private static Map getLagResultForNewKafkaSpout(String spoutId, SpoutSpec spoutSpec) throws IOException { return getLagResultForKafka(spoutId, spoutSpec); } - - /** - * Parse the stdout from {@code storm-kafka-monitor}. Returns the parsed JSON map on success, - * or {@code null} when the output is not parseable as a JSON object — which happens when the - * monitor printed a plaintext error string (json-smart parses unquoted text leniently as a - * String rather than throwing). The caller treats {@code null} as "monitor failed; surface - * the raw stdout as the error message". - */ - @SuppressWarnings("unchecked") - private static Map parseMonitorOutput(String resultFromMonitor) { - try { - Object parsed = JSONValue.parseWithException(resultFromMonitor); - if (parsed instanceof Map) { - return (Map) parsed; - } - LOGGER.debug("JSON parsing did not yield a Map, assuming message as error message: {}", resultFromMonitor); - return null; - } catch (ParseException e) { - LOGGER.debug("JSON parsing failed, assuming message as error message: {}", resultFromMonitor); - return null; - } - } } From 49835fe9533711a3ec5a9d3036cb67f9d422717a Mon Sep 17 00:00:00 2001 From: Diogo Pereira Date: Thu, 7 May 2026 22:12:09 +0100 Subject: [PATCH 3/3] address PR comments --- .../storm/kafka/monitor/KafkaOffsetLagUtil.java | 15 +++++++-------- .../kafka/monitor/KafkaOffsetLagUtilTest.java | 5 ++++- .../org/apache/storm/utils/TopologySpoutLag.java | 2 ++ 3 files changed, 13 insertions(+), 9 deletions(-) diff --git a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java index 6d48c3616c0..d5918f78443 100644 --- a/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java +++ b/external/storm-kafka-monitor/src/main/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtil.java @@ -68,10 +68,10 @@ public static void main(String[] args) { printUsageAndExit(options, OPTION_GROUP_ID_LONG + " and " + OPTION_BOOTSTRAP_BROKERS_LONG + " are required"); } NewKafkaSpoutOffsetQuery newKafkaSpoutOffsetQuery = - new NewKafkaSpoutOffsetQuery(commandLine.getOptionValue(OPTION_TOPIC_LONG), - commandLine.getOptionValue(OPTION_BOOTSTRAP_BROKERS_LONG), - commandLine.getOptionValue(OPTION_GROUP_ID_LONG), securityProtocol, saslMechanism, - commandLine.getOptionValue(OPTION_CONSUMER_CONFIG_LONG)); + new NewKafkaSpoutOffsetQuery(commandLine.getOptionValue(OPTION_TOPIC_LONG), + commandLine.getOptionValue(OPTION_BOOTSTRAP_BROKERS_LONG), + commandLine.getOptionValue(OPTION_GROUP_ID_LONG), securityProtocol, saslMechanism, + commandLine.getOptionValue(OPTION_CONSUMER_CONFIG_LONG)); List results = getOffsetLags(newKafkaSpoutOffsetQuery); Map> keyedResult = keyByTopicAndPartition(results); @@ -83,7 +83,7 @@ public static void main(String[] args) { } private static Map> keyByTopicAndPartition( - List results) { + List results) { Map> resultKeyedByTopic = new HashMap<>(); for (KafkaOffsetLagResult result : results) { @@ -95,7 +95,7 @@ private static Map> keyByTopicAndP } topicResultKeyedByPartition.put(result.getPartition(), - new KafkaPartitionOffsetLag(result.getConsumerCommittedOffset(), result.getLogHeadOffset())); + new KafkaPartitionOffsetLag(result.getConsumerCommittedOffset(), result.getLogHeadOffset())); } return resultKeyedByTopic; @@ -113,7 +113,7 @@ private static Options buildOptions() { options.addOption(OPTION_TOPIC_SHORT, OPTION_TOPIC_LONG, true, "REQUIRED Topics (comma separated list) for fetching log head and spout committed " - + "offset"); + + "offset"); options.addOption(OPTION_BOOTSTRAP_BROKERS_SHORT, OPTION_BOOTSTRAP_BROKERS_LONG, true, "Comma separated list of bootstrap broker hosts for new " @@ -136,7 +136,6 @@ private static Options buildOptions() { /** * Get offset lags. - * * @param newKafkaSpoutOffsetQuery represents the information needed to query kafka for log head and spout offsets * @return log head offset, spout offset and lag for each partition */ diff --git a/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java b/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java index 509ec808e47..c70a2db4f09 100644 --- a/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java +++ b/external/storm-kafka-monitor/src/test/java/org/apache/storm/kafka/monitor/KafkaOffsetLagUtilTest.java @@ -42,7 +42,10 @@ import org.testcontainers.kafka.KafkaContainer; import org.testcontainers.utility.DockerImageName; -@Testcontainers +/** + * Integration test — requires a Docker daemon. Skipped automatically when Docker is unavailable. + */ +@Testcontainers(disabledWithoutDocker = true) class KafkaOffsetLagUtilTest { private static final String TOPIC = "lag-test-topic"; diff --git a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java index 65ab2323dc9..2bf1f2bd8d9 100644 --- a/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java +++ b/storm-core/src/jvm/org/apache/storm/utils/TopologySpoutLag.java @@ -177,9 +177,11 @@ private static Map getLagResultForKafka(String spoutId, SpoutSpe } else { // json-smart parses unquoted plain text leniently as a String, so we can land here // when the monitor printed an error message instead of JSON; surface it as the error. + LOGGER.debug("Monitor returned non-JSON output, treating as error: {}", resultFromMonitor); errorMsg = resultFromMonitor; } } catch (ParseException e) { + LOGGER.debug("JSON parsing failed, assuming message as error message: {}", resultFromMonitor); errorMsg = resultFromMonitor; } } finally {