diff --git a/pom.xml b/pom.xml
index cf7c3fc..274b596 100644
--- a/pom.xml
+++ b/pom.xml
@@ -196,6 +196,13 @@
 		      <version>1.6.15</version>
 		      <scope>compile</scope>
 		</dependency>
+		<!-- https://mvnrepository.com/artifact/org.codehaus.janino/janino -->
+		<dependency>
+            <groupId>org.codehaus.janino</groupId>
+            <artifactId>janino</artifactId>
+            <version>3.1.7</version>
+        </dependency>
+
 		 
 	</dependencies>
 
diff --git a/src/main/java/it/finanze/sanita/fse2/ms/gtw/dispatcher/config/kafka/oauth2/CustomAuthenticateCallbackHandler.java b/src/main/java/it/finanze/sanita/fse2/ms/gtw/dispatcher/config/kafka/oauth2/CustomAuthenticateCallbackHandler.java
index 839d20d..eba83ce 100644
--- a/src/main/java/it/finanze/sanita/fse2/ms/gtw/dispatcher/config/kafka/oauth2/CustomAuthenticateCallbackHandler.java
+++ b/src/main/java/it/finanze/sanita/fse2/ms/gtw/dispatcher/config/kafka/oauth2/CustomAuthenticateCallbackHandler.java
@@ -3,11 +3,9 @@
 
 package it.finanze.sanita.fse2.ms.gtw.dispatcher.config.kafka.oauth2;
 
-import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.util.Arrays;
@@ -34,7 +32,6 @@
 import com.microsoft.aad.msal4j.IClientCredential;
 
 import it.finanze.sanita.fse2.ms.gtw.dispatcher.exceptions.BusinessException;
-import it.finanze.sanita.fse2.ms.gtw.dispatcher.utility.FileUtility;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
diff --git a/src/main/resources/logback-spring.xml b/src/main/resources/logback-spring.xml
index 610e4ab..4533c27 100644
--- a/src/main/resources/logback-spring.xml
+++ b/src/main/resources/logback-spring.xml
@@ -49,45 +49,69 @@
 		</encoder>
 	</appender>
 
-	<logger name="kafka-logger" level="${log.elastic.level}"
-		additivity="false">
-		<appender name="kafka-appender"
-			class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.KafkaAppender">
-			<encoder
-				class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
-				<pattern>%msg%n</pattern>
-			</encoder>
 
-			<topic>${kafka.log.base-topic}</topic>
-			<keyingStrategy
-				class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.HostNameKeyingStrategy" />
-			<deliveryStrategy
-				class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.BlockingDeliveryStrategy">
-				<timeout>0</timeout>
-			</deliveryStrategy>
-
-			<producerConfig>bootstrap.servers=${kafka.bootstrap-servers}</producerConfig>
-			<producerConfig>metadata.fetch.timeout.ms=99999999999</producerConfig>
-			<producerConfig>client.id=${kafka.producer.client-id}</producerConfig>
-			<producerConfig>security.protocol=${kafka.properties.security.protocol}</producerConfig>
-			<producerConfig>sasl.mechanism=${kafka.properties.sasl.mechanism}</producerConfig>
-			<producerConfig>sasl.jaas.config=${kafka.properties.sasl.jaas.config}</producerConfig>
-			<producerConfig>sasl.login.callback.handler.class=it.finanze.sanita.fse2.ms.gtw.dispatcher.config.kafka.oauth2.CustomAuthenticateCallbackHandler</producerConfig>
-			<producerConfig>kafka.oauth.tenantId=${kafka.oauth.tenantId}</producerConfig>
-			<producerConfig>kafka.oauth.appId=${kafka.oauth.appId}</producerConfig>
-			<producerConfig>kafka.oauth.pfxPathName=${kafka.oauth.pfxPathName}</producerConfig>
-			<producerConfig>kafka.oauth.pwd=${kafka.oauth.pwd}</producerConfig>
-			<springProfile name="default">
-				<producerConfig>ssl.truststore.location=${kafka.properties.ssl.truststore.location}</producerConfig>
-				<producerConfig>ssl.truststore.password=${kafka.properties.ssl.truststore.password}</producerConfig>
-			</springProfile>
-		</appender>
-	</logger>
+	<if condition='!property("kafka.properties.sasl.mechanism").equals("OAUTHBEARER")'>
+		<then>
+			<logger name="kafka-logger" level="${log.elastic.level}" additivity="false">
+				<appender name="kafka-appender" class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.KafkaAppender">
+					<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+						<pattern>%msg%n</pattern>
+					</encoder>
+	
+					<topic>${kafka.log.base-topic}</topic>
+					<keyingStrategy class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.HostNameKeyingStrategy" />
+					<deliveryStrategy class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.BlockingDeliveryStrategy">
+						<timeout>0</timeout>
+					</deliveryStrategy>
+	
+					<producerConfig>bootstrap.servers=${kafka.bootstrap-servers}</producerConfig>
+					<producerConfig>metadata.fetch.timeout.ms=99999999999</producerConfig>
+					<producerConfig>client.id=${kafka.producer.client-id}</producerConfig>
+					<producerConfig>security.protocol=${kafka.properties.security.protocol}</producerConfig>
+					<producerConfig>sasl.mechanism=${kafka.properties.sasl.mechanism}</producerConfig>
+					<producerConfig>sasl.jaas.config=${kafka.properties.sasl.jaas.config}</producerConfig>
+					<springProfile name="default">
+						<producerConfig>ssl.truststore.location=${kafka.properties.ssl.truststore.location}</producerConfig>
+						<producerConfig>ssl.truststore.password=${kafka.properties.ssl.truststore.password}</producerConfig>
+					</springProfile>
+				</appender>
+			</logger>
+		</then>
+	</if>
 
+	<if condition='property("kafka.properties.sasl.mechanism").equals("OAUTHBEARER")'>
+		<then>
+			<logger name="kafka-logger" level="${log.elastic.level}" additivity="false">
+				<appender name="kafka-appender" class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.KafkaAppender">
+					<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+						<pattern>%msg%n</pattern>
+					</encoder>
+	
+					<topic>${kafka.log.base-topic}</topic>
+					<keyingStrategy class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.HostNameKeyingStrategy" />
+					<deliveryStrategy class="it.finanze.sanita.fse2.ms.gtw.dispatcher.logging.BlockingDeliveryStrategy">
+						<timeout>0</timeout>
+					</deliveryStrategy>
+	
+					<producerConfig>bootstrap.servers=${kafka.bootstrap-servers}</producerConfig>
+					<producerConfig>metadata.fetch.timeout.ms=99999999999</producerConfig>
+					<producerConfig>client.id=${kafka.producer.client-id}</producerConfig>
+					<producerConfig>security.protocol=${kafka.properties.security.protocol}</producerConfig>
+					<producerConfig>sasl.mechanism=${kafka.properties.sasl.mechanism}</producerConfig>
+					<producerConfig>sasl.jaas.config=${kafka.properties.sasl.jaas.config}</producerConfig>
+					<producerConfig>sasl.login.callback.handler.class=it.finanze.sanita.fse2.ms.gtw.dispatcher.config.kafka.oauth2.CustomAuthenticateCallbackHandler</producerConfig>
+					<producerConfig>kafka.oauth.tenantId=${kafka.oauth.tenantId}</producerConfig>
+					<producerConfig>kafka.oauth.appId=${kafka.oauth.appId}</producerConfig>
+					<producerConfig>kafka.oauth.pfxPathName=${kafka.oauth.pfxPathName}</producerConfig>
+					<producerConfig>kafka.oauth.pwd=${kafka.oauth.pwd}</producerConfig>
+				</appender>
+			</logger>
+		</then>
+	</if>
 
 	<!-- LOG everything at INFO level -->
 	<root level="${APP_LOGGING_ROOT_LEVEL:-INFO}">
 		<appender-ref ref="${log.output.format}" />
 	</root>
 
-</configuration>
+</configuration>
\ No newline at end of file