Update log4j used by java liberty sample app to 2.17 (#82)

* Update log4j used by java liberty sample app to 2.17

* Update gradle.build

* Specify the required Gradle version in prerequisites

* Update KafkaServlet.java

Author: tinaselenge

kafka-java-console-sample/docs/Local.md

@@ -20,7 +20,7 @@ We will not discuss establishing a connection from your laptop to Event Streams
 2. **If you don't already have them, install the following prerequisites:**
 
 	* [git](https://git-scm.com/)
-	* [Gradle 7 or higher](https://gradle.org/)
+	* [Gradle 7 of higher](https://gradle.org/)
 	* Java 8 or higher
 
 

kafka-java-liberty-sample/build.gradle

@@ -37,8 +37,9 @@ repositories {
 dependencies {
   implementation 'org.apache.geronimo.specs:geronimo-servlet_3.0_spec:1.0'
   implementation 'org.apache.kafka:kafka-clients:2.7.+'
-  implementation 'log4j:log4j:1.2.17'
-  implementation 'org.slf4j:slf4j-log4j12:1.7.25'
+  implementation group: 'org.apache.logging.log4j', name: 'log4j-core', version: '2.17.+'
+  implementation group: 'org.apache.logging.log4j', name: 'log4j-api', version: '2.17.+'
+  implementation 'org.apache.logging.log4j:log4j-slf4j-impl:2.17.+'
   implementation 'com.fasterxml.jackson.core:jackson-annotations:2.9.+'
   implementation 'com.fasterxml.jackson.core:jackson-core:2.9.+'
   implementation 'com.fasterxml.jackson.core:jackson-databind:2.9.+'

kafka-java-liberty-sample/log4j/log4j.properties

@@ -0,0 +1,7 @@
+rootLogger.level=info
+rootLogger.appenderRef.console.ref=STDOUT
+
+appender.console.type=Console
+appender.console.name=STDOUT
+appender.console.layout.type=PatternLayout
+appender.console.layout.pattern=[%d] %p %m (%c)%n

kafka-java-liberty-sample/log4j/log4j2.properties

@@ -21,21 +21,21 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.log4j.Level;
-import org.apache.log4j.Logger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.Iterator;
 
 public class Environment {
 
-    private static final Logger logger = Logger.getLogger(Environment.class);
+    private static final Logger logger = LoggerFactory.getLogger(Environment.class);
 
     public static EventStreamsCredentials getEventStreamsCredentials() {
         String vcapServices = System.getenv("VCAP_SERVICES");
-        logger.log(Level.INFO, "VCAP_SERVICES: \n" + vcapServices);
+        logger.info("VCAP_SERVICES: \n" + vcapServices);
         if (vcapServices == null) {
-            logger.log(Level.ERROR, "VCAP_SERVICES environment variable is null.");
+            logger.error("VCAP_SERVICES environment variable is null.");
             throw new IllegalStateException("VCAP_SERVICES environment variable is null.");
         }
         return transformVcapServices(vcapServices);
@@ -55,7 +55,7 @@ private static EventStreamsCredentials transformVcapServices(String vcapServices
                     String potentialKey = it.next();
                     String messageHubJsonKey = "messagehub";
                     if (potentialKey.startsWith(messageHubJsonKey)) {
-                        logger.log(Level.WARN, "Using the '" + potentialKey + "' key from VCAP_SERVICES.");
+                        logger.warn("Using the '" + potentialKey + "' key from VCAP_SERVICES.");
                         instanceCredentials = instanceCredentials.get(potentialKey)
                                 .get(0)
                                 .get("credentials");
@@ -65,7 +65,7 @@ private static EventStreamsCredentials transformVcapServices(String vcapServices
             }
             return mapper.readValue(instanceCredentials.toString(), EventStreamsCredentials.class);
         } catch (IOException e) {
-            logger.log(Level.ERROR, "VCAP_SERVICES environment variable parses failed.");
+            logger.error("VCAP_SERVICES environment variable parses failed.");
             throw new IllegalStateException("VCAP_SERVICES environment variable parses failed.", e);
         }
     }

kafka-java-liberty-sample/src/main/java/com/eventstreams/samples/env/Environment.java

@@ -28,8 +28,8 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
-import org.apache.log4j.Level;
-import org.apache.log4j.Logger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.servlet.annotation.WebServlet;
 import javax.servlet.http.HttpServlet;
@@ -52,7 +52,7 @@
 @WebServlet("/KafkaServlet")
 public class KafkaServlet extends HttpServlet {
     private static final long serialVersionUID = 1L;
-    private static final Logger logger = Logger.getLogger(KafkaServlet.class);
+    private static final Logger logger = LoggerFactory.getLogger(KafkaServlet.class);
 
     private final String serverConfigDir = System.getProperty("server.config.dir");
     private final String resourceDir = serverConfigDir + File.separator
@@ -73,9 +73,9 @@ public class KafkaServlet extends HttpServlet {
      * Intialising the KafkaServlet
      */
     public void init() {
-        logger.log(Level.WARN, "Initialising Kafka Servlet");
-        logger.log(Level.WARN, "Server Config directory: " + serverConfigDir);
-        logger.log(Level.WARN, "Resource directory: " + resourceDir);
+        logger.warn("Initialising Kafka Servlet");
+        logger.warn("Server Config directory: " + serverConfigDir);
+        logger.warn("Resource directory: " + resourceDir);
 
         // Retrieve credentials from environment
         EventStreamsCredentials credentials = Environment.getEventStreamsCredentials();
@@ -93,7 +93,7 @@ public void init() {
         restApi.post("/admin/topics", "{ \"name\": \"" + topic + "\" }", new int[]{422});
 
         String topics = restApi.get("/admin/topics", false);
-        logger.log(Level.WARN, "Topics: " + topics);
+        logger.warn("Topics: " + topics);
 
         // Initialize Kafka Producer
         kafkaProducer = new KafkaProducer<>(getClientConfiguration(bootstrapServers, credentials.getApiKey(), true));
@@ -172,12 +172,12 @@ private Properties getClientConfiguration(String broker, String apikey, boolean
         }
 
         try {
-            logger.log(Level.WARN, "Reading properties file from: " + fileName);
+            logger.warn("Reading properties file from: " + fileName);
             propsStream = new FileInputStream(fileName);
             props.load(propsStream);
             propsStream.close();
         } catch (IOException e) {
-            logger.log(Level.ERROR, e);
+            logger.error(e.getMessage(),e);
             return props;
         }
 
@@ -188,7 +188,7 @@ private Properties getClientConfiguration(String broker, String apikey, boolean
         saslJaasConfig = saslJaasConfig.replace("APIKEY", apikey);
         props.setProperty("sasl.jaas.config", saslJaasConfig);
 
-        logger.log(Level.WARN, "Using properties: " + props);
+        logger.warn("Using properties: " + props);
 
         return props;
     }
@@ -199,7 +199,7 @@ private Properties getClientConfiguration(String broker, String apikey, boolean
      * @param topic
      */
     private void produce(String topic) {
-        logger.log(Level.WARN, "Producer is starting.");
+        logger.warn("Producer is starting.");
 
         String key = "key";
         // Push a message into the list to be sent.
@@ -217,7 +217,7 @@ private void produce(String topic) {
             RecordMetadata m = kafkaProducer.send(record).get();
             producedMessages++;
 
-            logger.log(Level.WARN, "Message produced, offset: " + m.offset());
+            logger.warn("Message produced, offset: " + m.offset());
 
             Thread.sleep(1000);
         } catch (final Exception e) {
@@ -226,7 +226,7 @@ private void produce(String topic) {
             System.exit(-1);
         }
         messageProduced = true;
-        logger.log(Level.WARN, "Producer is shutting down.");
+        logger.warn("Producer is shutting down.");
     }
 
     /**
@@ -250,19 +250,19 @@ public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                 @Override
                 public void onPartitionsAssigned(Collection<org.apache.kafka.common.TopicPartition> partitions) {
                     try {
-                        logger.log(Level.WARN, "Partitions " + partitions + " assigned, consumer seeking to end.");
+                        logger.warn("Partitions " + partitions + " assigned, consumer seeking to end.");
 
                         for (TopicPartition partition : partitions) {
                             long position = kafkaConsumer.position(partition);
-                            logger.log(Level.WARN, "current Position: " + position);
+                            logger.warn( "current Position: " + position);
 
-                            logger.log(Level.WARN, "Seeking to end...");
+                            logger.warn( "Seeking to end...");
                             kafkaConsumer.seekToEnd(Arrays.asList(partition));
-                            logger.log(Level.WARN,
+                            logger.warn(
                                     "Seek from the current position: " + kafkaConsumer.position(partition));
                             kafkaConsumer.seek(partition, position);
                         }
-                        logger.log(Level.WARN, "Producer can now begin producing messages.");
+                        logger.warn("Producer can now begin producing messages.");
                     } catch (final Exception e) {
                         logger.error("Error when assigning partitions", e);
                     }
@@ -273,7 +273,7 @@ public void onPartitionsAssigned(Collection<org.apache.kafka.common.TopicPartiti
 
         @Override
         public void run() {
-            logger.log(Level.WARN, "Consumer is starting.");
+            logger.warn("Consumer is starting.");
 
             while (!closing) {
                 try {
@@ -291,15 +291,15 @@ public void run() {
 
                     Thread.sleep(1000);
                 } catch (final InterruptedException e) {
-                    logger.log(Level.ERROR, "Producer/Consumer loop has been unexpectedly interrupted");
+                    logger.error("Producer/Consumer loop has been unexpectedly interrupted");
                     shutdown();
                 } catch (final Exception e) {
-                    logger.log(Level.ERROR, "Consumer has failed with exception: " + e);
+                    logger.error("Consumer has failed with exception: " + e);
                     shutdown();
                 }
             }
 
-            logger.log(Level.WARN, "Consumer is shutting down.");
+            logger.warn("Consumer is shutting down.");
             kafkaConsumer.close();
             consumedMessages.clear();
         }