From f37499aa3c175073e2faa85daf69ae2b5cfd0098 Mon Sep 17 00:00:00 2001 From: Konstantinos Dalianis <dean.dalianis@cern.ch> Date: Mon, 29 Apr 2024 11:21:30 +0200 Subject: [PATCH 1/2] Removed redundant exception. Added better logs for Streaming Applications --- .../nile/common/StreamingApplication.java | 16 +++--- .../common/clients/KafkaStreamsClient.java | 53 +++++++++++++------ .../common/exceptions/StreamingException.java | 17 ------ .../nile/common/streams/AbstractStream.java | 5 +- 4 files changed, 45 insertions(+), 46 deletions(-) delete mode 100644 src/main/java/ch/cern/nile/common/exceptions/StreamingException.java diff --git a/src/main/java/ch/cern/nile/common/StreamingApplication.java b/src/main/java/ch/cern/nile/common/StreamingApplication.java index cc719be..98d67b0 100644 --- a/src/main/java/ch/cern/nile/common/StreamingApplication.java +++ b/src/main/java/ch/cern/nile/common/StreamingApplication.java @@ -1,12 +1,12 @@ package ch.cern.nile.common; - import java.io.IOException; import java.lang.reflect.InvocationTargetException; import java.nio.file.Files; import java.nio.file.Paths; import java.util.Properties; +import org.apache.kafka.streams.errors.StreamsException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -16,7 +16,6 @@ import ch.cern.nile.common.clients.KafkaStreamsClient; import ch.cern.nile.common.configuration.PropertiesCheck; import ch.cern.nile.common.configuration.StreamType; import ch.cern.nile.common.configuration.properties.CommonProperties; -import ch.cern.nile.common.exceptions.StreamingException; import ch.cern.nile.common.streams.Streaming; /** @@ -27,6 +26,7 @@ import ch.cern.nile.common.streams.Streaming; public final class StreamingApplication { private static final Logger LOGGER = LoggerFactory.getLogger(StreamingApplication.class); + private static final int MIN_ARGS_LENGTH = 1; private StreamingApplication() { @@ -39,7 +39,7 @@ public final class StreamingApplication { * * @param args command-line arguments, expecting the path to the properties file as the first argument * @throws IllegalArgumentException if the properties file path is not provided - * @throws StreamingException if there are issues with loading the properties file, + * @throws StreamsException if there are issues with loading the properties file, * validating properties, or starting the streaming process */ @SuppressFBWarnings(value = "PATH_TRAVERSAL_IN", justification = "This method is only used internally") @@ -55,11 +55,10 @@ public final class StreamingApplication { } catch (IOException e) { final String message = "Error while loading the properties file"; LOGGER.error(message, e); - throw new StreamingException(message, e); + throw new StreamsException(message, e); } - final StreamType sType = - StreamType.valueOf(configs.getProperty(CommonProperties.STREAM_TYPE.getValue(), null)); + final StreamType sType = StreamType.valueOf(configs.getProperty(CommonProperties.STREAM_TYPE.getValue(), null)); PropertiesCheck.validateProperties(configs, sType); @@ -67,8 +66,7 @@ public final class StreamingApplication { client.configure(configs); try { - final Class<?> clazz = - Class.forName(configs.getProperty(CommonProperties.STREAM_CLASS.getValue())); + final Class<?> clazz = Class.forName(configs.getProperty(CommonProperties.STREAM_CLASS.getValue())); final Streaming streaming; streaming = (Streaming) clazz.getDeclaredConstructor().newInstance(); streaming.configure(configs); @@ -77,7 +75,7 @@ public final class StreamingApplication { | InvocationTargetException | NoSuchMethodException e) { final String message = "Error while starting the stream"; LOGGER.error(message, e); - throw new StreamingException(message, e); + throw new StreamsException(message, e); } } diff --git a/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java b/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java index 5ec6f05..ab87135 100644 --- a/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java +++ b/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java @@ -12,6 +12,9 @@ import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.errors.DefaultProductionExceptionHandler; import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler; +import org.apache.kafka.streams.errors.StreamsException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import ch.cern.nile.common.configuration.Configure; import ch.cern.nile.common.configuration.properties.ClientProperties; @@ -23,6 +26,12 @@ import ch.cern.nile.common.json.JsonSerde; */ public class KafkaStreamsClient implements Configure { + private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamsClient.class); + + private static final String SECURITY_PROTOCOL = "SASL_SSL"; + private static final String SASL_MECHANISM = "GSSAPI"; + private static final String SASL_KERBEROS_SERVICE_NAME = "kafka"; + private static final String TEST_CLUSTER_NAME = "test"; private Properties properties; @@ -37,29 +46,33 @@ public class KafkaStreamsClient implements Configure { */ @Override public void configure(final Properties props) { + if (LOGGER.isDebugEnabled()) { + LOGGER.debug("Configuring KafkaStreamsClient with properties: {}", props); + } + final String clientId = props.getProperty(ClientProperties.CLIENT_ID.getValue()); - this.properties = new Properties(); - this.properties.put(StreamsConfig.APPLICATION_ID_CONFIG, clientId); - this.properties.put(StreamsConfig.CLIENT_ID_CONFIG, clientId); + properties = new Properties(); + properties.put(StreamsConfig.APPLICATION_ID_CONFIG, clientId); + properties.put(StreamsConfig.CLIENT_ID_CONFIG, clientId); final String kafkaCluster = props.getProperty(ClientProperties.KAFKA_CLUSTER.getValue()); if (TEST_CLUSTER_NAME.equals(kafkaCluster)) { - this.properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, props.getProperty("bootstrap.servers")); + properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, props.getProperty("bootstrap.servers")); } else { - this.properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.reverseDnsLookup(kafkaCluster)); - this.properties.put(StreamsConfig.SECURITY_PROTOCOL_CONFIG, "SASL_SSL"); - this.properties.put(SaslConfigs.SASL_MECHANISM, "GSSAPI"); - this.properties.put(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, "kafka"); - this.properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, + properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, this.reverseDnsLookup(kafkaCluster)); + properties.put(StreamsConfig.SECURITY_PROTOCOL_CONFIG, SECURITY_PROTOCOL); + properties.put(SaslConfigs.SASL_MECHANISM, SASL_MECHANISM); + properties.put(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, SASL_KERBEROS_SERVICE_NAME); + properties.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, props.getProperty(ClientProperties.TRUSTSTORE_LOCATION.getValue())); } - this.properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); - this.properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); - this.properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, + properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); + properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); + properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndContinueExceptionHandler.class.getName()); - this.properties.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, + properties.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG, DefaultProductionExceptionHandler.class.getName()); } @@ -70,7 +83,13 @@ public class KafkaStreamsClient implements Configure { * @return a configured KafkaStreams instance */ public KafkaStreams create(final Topology topology) { - return new KafkaStreams(topology, properties); + try { + return new KafkaStreams(topology, properties); + } catch (StreamsException e) { + final String message = "Failed to create KafkaStreams instance with properties: " + properties; + LOGGER.error(message, e); + throw new StreamsException(message, e); + } } /** @@ -78,14 +97,14 @@ public class KafkaStreamsClient implements Configure { * * @param kafkaCluster the domain of the Kafka cluster * @return a comma-separated list of hostnames with port 9093 - * @throws RuntimeException if the hostname resolution fails */ private String reverseDnsLookup(final String kafkaCluster) { try { return performDnsLookup(kafkaCluster); } catch (UnknownHostException e) { - throw new ReverseDnsLookupException( - "Failed to perform reverse DNS lookup for the Kafka cluster: " + kafkaCluster, e); + final String message = "Failed to perform reverse DNS lookup for the Kafka cluster: " + kafkaCluster; + LOGGER.error(message, e); + throw new ReverseDnsLookupException(message, e); } } diff --git a/src/main/java/ch/cern/nile/common/exceptions/StreamingException.java b/src/main/java/ch/cern/nile/common/exceptions/StreamingException.java deleted file mode 100644 index de716e6..0000000 --- a/src/main/java/ch/cern/nile/common/exceptions/StreamingException.java +++ /dev/null @@ -1,17 +0,0 @@ -package ch.cern.nile.common.exceptions; - -import java.io.Serial; - -public class StreamingException extends RuntimeException { - - @Serial - private static final long serialVersionUID = 1L; - - public StreamingException(final Throwable cause) { - super(cause); - } - - public StreamingException(final String message, final Throwable cause) { - super(message, cause); - } -} diff --git a/src/main/java/ch/cern/nile/common/streams/AbstractStream.java b/src/main/java/ch/cern/nile/common/streams/AbstractStream.java index 6a77da1..0fd0abd 100644 --- a/src/main/java/ch/cern/nile/common/streams/AbstractStream.java +++ b/src/main/java/ch/cern/nile/common/streams/AbstractStream.java @@ -6,6 +6,7 @@ import java.util.concurrent.CountDownLatch; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.errors.StreamsException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,7 +19,6 @@ import lombok.Setter; import ch.cern.nile.common.clients.KafkaStreamsClient; import ch.cern.nile.common.configuration.properties.ClientProperties; import ch.cern.nile.common.configuration.properties.DecodingProperties; -import ch.cern.nile.common.exceptions.StreamingException; import ch.cern.nile.common.probes.Health; /** @@ -70,7 +70,6 @@ public abstract class AbstractStream implements Streaming { * to the JVM and terminates the JVM upon completion of the stream. * * @param kafkaStreamsClient the client used to create and manage the Kafka Streams instance - * @throws StreamingException if an error occurs during streaming */ @Override @SuppressWarnings("PMD.DoNotTerminateVM") @@ -124,7 +123,7 @@ public abstract class AbstractStream implements Streaming { latch.await(); } catch (InterruptedException e) { LOGGER.error("Error while waiting for latch", e); - throw new StreamingException("Error while waiting for latch", e); + throw new StreamsException("Error while waiting for latch", e); } } -- GitLab From 44fe54380e246611ffb4df1e1296e6fb19c6f032 Mon Sep 17 00:00:00 2001 From: Konstantinos Dalianis <dean.dalianis@cern.ch> Date: Mon, 29 Apr 2024 11:40:02 +0200 Subject: [PATCH 2/2] Removed redundant exception. Added better logs for Streaming Applications --- .../ch/cern/nile/common/clients/KafkaStreamsClient.java | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java b/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java index ab87135..a07914e 100644 --- a/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java +++ b/src/main/java/ch/cern/nile/common/clients/KafkaStreamsClient.java @@ -46,10 +46,6 @@ public class KafkaStreamsClient implements Configure { */ @Override public void configure(final Properties props) { - if (LOGGER.isDebugEnabled()) { - LOGGER.debug("Configuring KafkaStreamsClient with properties: {}", props); - } - final String clientId = props.getProperty(ClientProperties.CLIENT_ID.getValue()); properties = new Properties(); properties.put(StreamsConfig.APPLICATION_ID_CONFIG, clientId); @@ -86,7 +82,8 @@ public class KafkaStreamsClient implements Configure { try { return new KafkaStreams(topology, properties); } catch (StreamsException e) { - final String message = "Failed to create KafkaStreams instance with properties: " + properties; + final String message = + "Failed to create KafkaStreams instance with properties: " + properties; LOGGER.error(message, e); throw new StreamsException(message, e); } -- GitLab