Issue178 (#179)

* fixes #178 remove open tracing as it is replaced by open telemetry

* add precommit config

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: stevehu

.pre-commit-config.yaml

@@ -0,0 +1,14 @@
+repos:
+-   repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.5.0
+    hooks:
+    -   id: trailing-whitespace
+    -   id: end-of-file-fixer
+    -   id: check-yaml
+    -   id: check-added-large-files
+-   repo: https://github.com/networknt/pre-commit-hook-keyword
+    rev: f17c4de14fc24420f6768c19cad06ba03af06d86
+    hooks:
+    -   id: keywordscan
+        args: ["--keywords=c3VubGlmZQ==,Y2liYw==,c3VuIGxpZmU="]
+        types: ["text"]

README.md

@@ -12,11 +12,11 @@ A fast and light-weight reverse proxy with embedded gateway to wrap third party
 
 ## Why Reverse Proxy
 
-All the services developed on top of light-4j frameworks support [client side service discovery](http://microservices.io/patterns/client-side-discovery.html), 
+All the services developed on top of light-4j frameworks support [client side service discovery](http://microservices.io/patterns/client-side-discovery.html),
 load balance and cluster natively. So there is no need to put a reverse proxy instance in front of our
 services like other API frameworks that support only [server side service discovery](http://microservices.io/patterns/server-side-discovery.html).
 
-Also, light services embed a distributed gateway to address all the cross-cutting concerns in the 
+Also, light services embed a distributed gateway to address all the cross-cutting concerns in the
 request/response chain and work with the ecosystem that consists:
 
 * [light-oauth2](https://doc.networknt.com/service/oauth/) for security
@@ -31,14 +31,14 @@ request/response chain and work with the ecosystem that consists:
 Currently, we only support Java language; however, we are planning to support Nodejs and Go in the future
 if there are enough customer demands. For some of our customers, they have some existing RESTful APIs that
 built on top of other Java frameworks or other languages. We've been asked frequently on how to interact
-with these services to/from light services and how to enable security, metrics, logging, discovery, 
-validation, sanitization etc. on the existing services. 
+with these services to/from light services and how to enable security, metrics, logging, discovery,
+validation, sanitization etc. on the existing services.
 
-Our answer is to deploy a reverse proxy built on top of light-4j framework that wraps the existing service. 
+Our answer is to deploy a reverse proxy built on top of light-4j framework that wraps the existing service.
 
 The reverse proxy has the following features:
 
-* High throughput, low latency and small footprint. 
+* High throughput, low latency and small footprint.
 * Integrate light-oauth2 to protect un-secured services
 * Built-in load balancer
 * Can be started with Docker or standalone
@@ -58,7 +58,7 @@ Click [here](https://doc.networknt.com/tutorial/kafka-sidecar/local-dev/#reactiv
 
 If the evnironment is windows OS, we can start confluent kafka docker-compose for testing:
 
-From command line, create a docker network first which will indicate kafka and sidecar running in same network. 
+From command line, create a docker network first which will indicate kafka and sidecar running in same network.
 
 ```
 cd kafka-sidecar
@@ -159,7 +159,7 @@ value schema:
 There are two options for starting kafka sidecar and backend api:
 
 - by standalone APIs
-  
+
 - by docker-compose
 ----
 #### By docker-compose
@@ -255,12 +255,12 @@ The Reactive Consumer will send those message to DLQ if deadLetterEnabled.
 
 ### Using the sidecar endpoint to producer error message to DLQ:
 
-- endpoint: 
+- endpoint:
 
   /consumers/deadLetter/active
 
 - method
-  
+
    POST
 
 Sample request payload:
@@ -305,19 +305,19 @@ Sample request payload:
 
 ### Active Consumer Workflow:
 
-Kafka sidecar provide the end-to-end workflow for actively consumer the records from kafka topic(s).  
+Kafka sidecar provide the end-to-end workflow for actively consumer the records from kafka topic(s).
 
 When user try  following workflow to consumer records:
 
 - enable ActiveConsumerStartupHook on service.yml (or values.yml)
-  
+
    This will create kafka consumer manager for active consumer on kafka sidecar server startup
 
 
 -  create consumer group
-   
+
     endpoint: /consumers/{group}
- 
+
     method: POST
 
 
@@ -342,16 +342,15 @@ When user try  following workflow to consumer records:
   method: POST
 
 
-- add the records process detail to audit 
+- add the records process detail to audit
 
   endpoint: /consumers/active/audit
 
   method: POST
 
-### To learn how to use this proxy, pleases refer to 
+### To learn how to use this proxy, pleases refer to
 
 * [Getting Started](https://doc.networknt.com/getting-started/light-proxy/) to learn core concepts
 * [Tutorial](https://doc.networknt.com/tutorial/proxy/) with step by step guide for RESTful proxy
 * [Configuration](https://doc.networknt.com/service/proxy/configuration/) for different configurations based on your situations
 * [Artifact](https://doc.networknt.com/service/proxy/artifact/) to guide customer to choose the right artifact to deploy light-proxy.
-

doc/ksql.md

@@ -205,14 +205,14 @@ Sample request body:
             "key": "2",
             "value": {
                 "userId": "2222",
-                 "firstName": "test2"                
+                 "firstName": "test2"
             }
         },
         {
             "key": "3",
             "value": {
                 "userId": "3333",
-                 "firstName": "test3"                       
+                 "firstName": "test3"
             }
         }
     ]
@@ -222,8 +222,8 @@ Sample request body:
 - Create KTable based on the topic created above:
 
 ```json
-CREATE TABLE USERS 
-   (ID STRING PRIMARY KEY, USERID STRING, FIRSTNAME STRING, LASTNAME STRING, COUNTRY STRING) 
+CREATE TABLE USERS
+   (ID STRING PRIMARY KEY, USERID STRING, FIRSTNAME STRING, LASTNAME STRING, COUNTRY STRING)
     WITH (KAFKA_TOPIC='test', KEY_FORMAT='KAFKA', VALUE_FORMAT='JSON_SR');
 ```
 
@@ -240,7 +240,7 @@ CREATE TABLE USERS
 ```text
 curl --location --request POST 'http://localhost:8442/ksqldb/active' \
 --header 'Content-Type: application/json' \
---data-raw ' 
+--data-raw '
 {
     "offset": "earliest",
     "deserializationError": false,
@@ -261,4 +261,4 @@ Response:
         "ID": "1"
     }
 ]
-```
+```

docker-compose-demo.yml

@@ -28,4 +28,4 @@ services:
 networks:
   localnet:
     # driver: bridge
-    external: true
+    external: true

docker-compose.yml

@@ -198,4 +198,4 @@ services:
 networks:
   localnet:
     # driver: bridge
-    external: true
+    external: true

src/main/java/com/networknt/mesh/kafka/AuditProducerShutdownHook.java

@@ -19,5 +19,5 @@ public void onShutdown() {
         }
         logger.info("AuditProducerShutdownHook ends");
     }
-    
+
 }

src/main/java/com/networknt/mesh/kafka/AuditProducerStartupHook.java

@@ -20,5 +20,5 @@ public void onStartup() {
         auditProducer = lightProducer.getProducer();
         logger.info("AuditProducerStartupHook ends");
     }
-    
+
 }

src/main/java/com/networknt/mesh/kafka/handler/ConsumersGroupPostHandler.java

@@ -29,7 +29,7 @@ public ConsumersGroupPostHandler () {
         if(logger.isDebugEnabled()) logger.debug("ConsumersGroupPostHandler constructed!");
     }
 
-    
+
     @Override
     public void handleRequest(HttpServerExchange exchange) throws Exception {
         String group = exchange.getPathParameters().get("group").getFirst();

src/main/java/com/networknt/mesh/kafka/handler/DeadlettersQueueActiveGetHandler.java

@@ -37,7 +37,7 @@ public DeadlettersQueueActiveGetHandler() {
         if(logger.isDebugEnabled()) logger.debug("ReplayDeadLetterTopicGetHandler constructed!");
     }
 
-    
+
     @Override
     public void handleRequest(HttpServerExchange exchange) throws Exception {
         String groupId = exchange.getQueryParameters().get("group")==null? config.getGroupId() : exchange.getQueryParameters().get("group").getFirst();

src/main/java/com/networknt/mesh/kafka/handler/ProducersTopicPostHandler.java

@@ -1,54 +1,27 @@
 package com.networknt.mesh.kafka.handler;
 
-import com.fasterxml.jackson.databind.node.NullNode;
-import com.google.protobuf.ByteString;
 import com.networknt.body.BodyHandler;
 import com.networknt.config.Config;
 import com.networknt.config.JsonMapper;
-import com.networknt.exception.FrameworkException;
 import com.networknt.handler.LightHttpHandler;
-import com.networknt.httpstring.AttachmentConstants;
-import com.networknt.httpstring.HttpStringConstants;
 import com.networknt.kafka.common.KafkaProducerConfig;
 import com.networknt.kafka.entity.*;
 import com.networknt.kafka.producer.*;
 import com.networknt.mesh.kafka.ProducerStartupHook;
 import com.networknt.mesh.kafka.WriteAuditLog;
-import com.networknt.server.Server;
 import com.networknt.server.ServerConfig;
 import com.networknt.service.SingletonServiceFactory;
-import com.networknt.status.Status;
 import com.networknt.utility.Constants;
 import com.networknt.kafka.entity.EmbeddedFormat;
-import com.networknt.utility.Util;
-import io.confluent.kafka.schemaregistry.avro.AvroSchemaProvider;
-import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
-import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
-import io.confluent.kafka.schemaregistry.client.rest.RestService;
-import io.confluent.kafka.schemaregistry.json.JsonSchemaProvider;
-import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchemaProvider;
-import io.confluent.kafka.serializers.subject.TopicNameStrategy;
-import io.opentracing.Tracer;
-import io.opentracing.propagation.Format;
-import io.opentracing.tag.Tags;
 import io.undertow.server.HttpServerExchange;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.errors.AuthenticationException;
-import org.apache.kafka.common.errors.AuthorizationException;
-import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.charset.StandardCharsets;
-import java.time.Instant;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
-
-import static java.util.Collections.singletonList;
 
 /**
  * The producer endpoint that can receive request from the backend service messages and push them
@@ -139,6 +112,7 @@ public Headers populateHeaders(HttpServerExchange exchange, KafkaProducerConfig
         if(token != null) {
             headers.add(Constants.AUTHORIZATION_STRING, token.getBytes(StandardCharsets.UTF_8));
         }
+        /*
         if(config.isInjectOpenTracing()) {
             // maybe we should move this to the ProduceRecord in the future like the correlationId and traceabilityId.
             Tracer tracer = exchange.getAttachment(AttachmentConstants.EXCHANGE_TRACER);
@@ -148,7 +122,7 @@ public Headers populateHeaders(HttpServerExchange exchange, KafkaProducerConfig
                 tracer.inject(tracer.activeSpan().context(), Format.Builtin.TEXT_MAP, new KafkaHeadersCarrier(headers));
             }
         }
-
+        */
         // remove the correlationId from the HTTP header and moved it to the ProduceRecord as it is per record attribute.
         // remove the traceabilityId from the HTTP header and moved it to the ProduceRecord as it is per record attribute.
 

src/main/java/com/networknt/mesh/kafka/handler/TopicReplayPostHandler.java

@@ -11,17 +11,12 @@
 import com.networknt.kafka.common.KafkaProducerConfig;
 import com.networknt.kafka.common.KafkaStreamsConfig;
 import com.networknt.kafka.entity.*;
-import com.networknt.kafka.producer.KafkaHeadersCarrier;
 import com.networknt.kafka.producer.NativeLightProducer;
 import com.networknt.kafka.producer.SidecarProducer;
-import com.networknt.server.Server;
 import com.networknt.server.ServerConfig;
 import com.networknt.service.SingletonServiceFactory;
 import com.networknt.utility.Constants;
 import com.networknt.utility.StringUtils;
-import io.opentracing.Tracer;
-import io.opentracing.propagation.Format;
-import io.opentracing.tag.Tags;
 import io.undertow.server.HttpServerExchange;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -175,6 +170,7 @@ public Headers populateHeaders(HttpServerExchange exchange, KafkaProducerConfig
         if(token != null) {
             headers.add(Constants.AUTHORIZATION_STRING, token.getBytes(StandardCharsets.UTF_8));
         }
+        /*
         if(config.isInjectOpenTracing()) {
             // maybe we should move this to the ProduceRecord in the future like the correlationId and traceabilityId.
             Tracer tracer = exchange.getAttachment(AttachmentConstants.EXCHANGE_TRACER);
@@ -184,7 +180,7 @@ public Headers populateHeaders(HttpServerExchange exchange, KafkaProducerConfig
                 tracer.inject(tracer.activeSpan().context(), Format.Builtin.TEXT_MAP, new KafkaHeadersCarrier(headers));
             }
         }
-
+        */
         // remove the correlationId from the HTTP header and moved it to the ProduceRecord as it is per record attribute.
         // remove the traceabilityId from the HTTP header and moved it to the ProduceRecord as it is per record attribute.
 

src/main/java/com/networknt/mesh/kafka/util/ConvertToList.java

@@ -18,4 +18,4 @@ public static List<Map<String, Object>> string2List(ObjectMapper objectMapper,St
             throw new RuntimeException(e);
         }
     }
-}
+}

src/main/java/com/networknt/mesh/kafka/util/CreateConsumerGroup.java

@@ -49,15 +49,15 @@ public String createConsumer() {
     }
 
     /**
-     * 
+     *
      * @return instance Id Utility
      */
     public String getInstanceId(){
         return StringUtils.isEmpty(instanceId) ? null : instanceId;
     }
 
     /**
-     * 
+     *
      * @param instanceId A string instance Id
      */
     public void setInstanceId(String instanceId){

src/main/java/com/networknt/mesh/kafka/util/KafkaConsumerManagerFactory.java

@@ -10,4 +10,4 @@ private KafkaConsumerManagerFactory(){}
     public static KafkaConsumerManager createKafkaConsumerManager(KafkaConsumerConfig config) {
         return new KafkaConsumerManager(config);
     }
-}
+}

src/main/java/com/networknt/mesh/kafka/util/StreamsFactory.java

@@ -32,4 +32,4 @@ public static SubscribeTopic createSubscribeTopic(String group) {
     }
 
     public static KafkaStreams createKafkaStreams(Topology topology, Properties props) { return new KafkaStreams(topology, props); }
-}
+}

src/main/resources/config/kafka-consumer.yml

@@ -89,4 +89,3 @@ iteratorBackoffMs: ${kafka-consumer.iteratorBackoffMs:50}
 #keep alive time out occurs and sidecar consumer does not move forward. Hence we are adding this property so that we can explicitly close the connection
 #when we receive the response and not wait for FinAck.
 backendConnectionReset: ${kafka-consumer.backendConnectionReset:false}
-

src/test/java/com/networknt/mesh/kafka/handler/ConsumersGroupInstancesInstanceAssignmentsGetHandlerTest.java

@@ -63,11 +63,11 @@ public void testConsumersGroupInstancesInstanceAssignmentsGetHandlerTest() throw
             ClientConnection connection = (ClientConnection) connectionToken.getRawConnection();
 
             ClientRequest request = new ClientRequest().setPath(requestUri).setMethod(Methods.GET);
-            
-            //customized header parameters 
+
+            //customized header parameters
             request.getRequestHeaders().put(new HttpString("host"), "localhost");
             connection.sendRequest(request, client.createClientCallback(reference, latch));
-            
+
             latch.await();
         } catch (Exception e) {
             logger.error("Exception: ", e);
@@ -89,4 +89,3 @@ public void testConsumersGroupInstancesInstanceAssignmentsGetHandlerTest() throw
         Assert.assertNull(status);
     }
 }
-