feat: add request logging (#3518)

* fix: add request logging

While working with KSQL server logs I noticed that the logs output by rest-utils is not of much use:

```
[2019-10-03 09:22:06,518] INFO 10.16.50.25 - - [03/Oct/2019:07:22:05 +0000] "POST /ksql HTTP/1.1" 200 16820  1298 (io.confluent.rest-utils.requests:60)
```

All it tells us is the time of the request, the status code and the size of the payload.  What it does not tell us is what was in the request. This would be useful when debugging!

With this change we get logs like:

```
[2019-10-09 22:34:06,193] INFO Received: KsqlRequest{ksql='CREATE STREAM TEST (ID INT) WITH (kafka_topic='test_topic', value_format='JSON');INSERT INTO TEST (ROWTIME, ROWKEY, ID) VALUES (1234, 'key', 10);', streamsProperties={}, commandSequenceNumber=Optional.empty} (io.confluent.ksql.rest.server.resources.KsqlResource:198)
```

Which tells us not only what command was run, but also with what local property overrides. Useful stuff!

* chore: add log4j help

* Update config/log4j.properties

Co-Authored-By: Victoria Xia <[email protected]>

config/log4j.properties

@@ -30,3 +30,6 @@ log4j.logger.org.apache.zookeeper=ERROR, stdout
 log4j.logger.org.apache.kafka=ERROR, stdout
 log4j.logger.org.I0Itec.zkclient=ERROR, stdout
 log4j.logger.org.reflections=ERROR, stdout
+
+# Uncomment the following line to stop KSQL from logging out each request it receives:
+#log4j.logger.io.confluent.ksql.rest.server.resources.KsqlResource=WARN, stdout

ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java

@@ -67,6 +67,8 @@
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import org.apache.kafka.streams.StreamsConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 // CHECKSTYLE_RULES.OFF: ClassDataAbstractionCoupling
 @Path("/ksql")
@@ -75,6 +77,8 @@
 public class KsqlResource implements KsqlConfigurable {
   // CHECKSTYLE_RULES.ON: ClassDataAbstractionCoupling
 
+  private static final Logger LOG = LoggerFactory.getLogger(KsqlResource.class);
+
   private static final List<ParsedStatement> TERMINATE_CLUSTER =
       new DefaultKsqlParser().parse(TerminateCluster.TERMINATE_CLUSTER_STATEMENT_TEXT);
 
@@ -171,6 +175,8 @@ public Response terminateCluster(
       @Context final ServiceContext serviceContext,
       final ClusterTerminateRequest request
   ) {
+    LOG.info("Received: " + request);
+
     throwIfNotConfigured();
 
     ensureValidPatterns(request.getDeleteTopicList());
@@ -189,6 +195,8 @@ public Response handleKsqlStatements(
       @Context final ServiceContext serviceContext,
       final KsqlRequest request
   ) {
+    LOG.info("Received: " + request);
+
     throwIfNotConfigured();
 
     activenessRegistrar.updateLastRequestTime();

ksql-rest-model/src/main/java/io/confluent/ksql/rest/entity/ClusterTerminateRequest.java

@@ -67,4 +67,11 @@ public boolean equals(final Object o) {
     final ClusterTerminateRequest that = (ClusterTerminateRequest) o;
     return Objects.equals(deleteTopicList, that.deleteTopicList);
   }
+
+  @Override
+  public String toString() {
+    return "ClusterTerminateRequest{"
+        + "deleteTopicList=" + deleteTopicList
+        + '}';
+  }
 }

ksql-rest-model/src/main/java/io/confluent/ksql/rest/entity/KsqlRequest.java

@@ -85,6 +85,15 @@ public int hashCode() {
     return Objects.hash(ksql, streamsProperties, commandSequenceNumber);
   }
 
+  @Override
+  public String toString() {
+    return "KsqlRequest{"
+        + "ksql='" + ksql + '\''
+        + ", streamsProperties=" + streamsProperties
+        + ", commandSequenceNumber=" + commandSequenceNumber
+        + '}';
+  }
+
   /**
    * Converts all Class references values to their canonical String value.
    * </p>