[type:feat] add Logging-Kafka Plugin

shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-http/k8s/script/e2e-http-sync.sh

@@ -37,6 +37,8 @@ for sync in ${SYNC_ARRAY[@]}; do
 
   kubectl apply -f "${PRGDIR}"/shenyu-rocketmq.yml
 
+  kubectl apply -f "${PRGDIR}"/shenyu-kafka.yml
+
   sleep 30s
   echo "[Start ${sync} synchronous] create shenyu-admin-${sync}.yml shenyu-bootstrap-${sync}.yml shenyu-examples-springcloud.yml"
   # shellcheck disable=SC2199
@@ -75,6 +77,8 @@ for sync in ${SYNC_ARRAY[@]}; do
   kubectl delete -f "${SHENYU_TESTCASE_DIR}"/k8s/sync/shenyu-bootstrap-"${sync}".yml
   kubectl delete -f "${PRGDIR}"/shenyu-examples-http.yml
   kubectl delete -f "${PRGDIR}"/shenyu-rocketmq.yml
+  kubectl delete -f "${PRGDIR}"/shenyu-kafka.yml
+
   # shellcheck disable=SC2199
   # shellcheck disable=SC2076
   if [[ "${MIDDLEWARE_SYNC_ARRAY[@]}" =~ "${sync}" ]]; then

shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-http/k8s/shenyu-kafka.yml

@@ -0,0 +1,98 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: zookeeper
+  namespace: default
+  labels:
+    app: zookeeper
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: zookeeper
+  template:
+    metadata:
+      labels:
+        app: zookeeper
+    spec:
+      containers:
+        - name: zookeeper
+          image: wurstmeister/zookeeper:3.4.6
+          ports:
+            - containerPort: 2181
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: zookeeper
+  namespace: default
+  labels:
+    app: zookeeper
+spec:
+  ports:
+    - port: 2181
+      name: client
+  selector:
+    app: zookeeper
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: kafka
+  namespace: default
+  labels:
+    app: kafka
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: kafka
+  template:
+    metadata:
+      labels:
+        app: kafka
+    spec:
+      containers:
+        - name: kafka
+          image: wurstmeister/kafka:2.12-2.1.1
+          env:
+            - name: KAFKA_ADVERTISED_LISTENERS
+              value: PLAINTEXT://kafka:9092
+            - name: KAFKA_ZOOKEEPER_CONNECT
+              value: zookeeper:2181
+          ports:
+            - containerPort: 9092
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: kafka
+  namespace: default
+  labels:
+    app: kafka
+spec:
+  ports:
+    - port: 9092
+      name: client
+      protocol: TCP
+      targetPort: 9092
+      nodePort: 31877
+
+  selector:
+    app: kafka

shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-http/pom.xml

@@ -32,5 +32,10 @@
             <artifactId>rocketmq-client</artifactId>
             <version>4.9.3</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>2.4.1</version>
+        </dependency>
     </dependencies>
 </project>

shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-http/src/test/java/org/apache/shenyu/e2e/testcase/http/DividePluginCases.java

@@ -20,6 +20,10 @@
 import com.google.common.collect.Lists;
 import io.restassured.http.Method;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
@@ -35,8 +39,13 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Properties;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.apache.shenyu.e2e.engine.scenario.function.HttpCheckers.exists;
 import static org.apache.shenyu.e2e.template.ResourceDataTemplate.newConditions;
@@ -75,6 +84,67 @@ private ShenYuScenarioSpec testDivideHello() {
                 .build();
     }
 
+    private  ShenYuScenarioSpec testKafkaHello(){
+        return ShenYuScenarioSpec.builder()
+                .name("testKafkaHello")
+                .beforeEachSpec(
+                        ShenYuBeforeEachSpec.builder()
+                                .addSelectorAndRule(
+                                        newSelectorBuilder("selector",Plugin.LOGGING_KAFKA)
+                                                .name("2")
+                                                .matchMode(MatchMode.OR)
+                                                .conditionList(newConditions(Condition.ParamType.URI, Condition.Operator.STARTS_WITH, "/http"))
+                                                .build(),
+                                        newRuleBuilder("rule")
+                                                .name("2")
+                                                .matchMode(MatchMode.OR)
+                                                .conditionList(newConditions(Condition.ParamType.URI, Condition.Operator.STARTS_WITH, "/http"))
+                                                .build()
+                                )
+                                .checker(exists(TEST))
+                                .build()
+                )
+                .caseSpec(
+                        ShenYuCaseSpec.builder()
+                                .add(request -> {
+                                    AtomicBoolean isLog = new AtomicBoolean(false);
+                                    try {
+                                        Thread.sleep(1000 * 30);
+                                        request.request(Method.GET, "/http/order/findById?id=23");
+                                        Properties props = new Properties();
+                                        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+                                                StringDeserializer.class.getName());
+                                        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+                                                StringDeserializer.class.getName());
+                                        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+                                                "127.0.0.1:31877");
+                                        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+                                        consumer.subscribe(Arrays.asList(TOPIC));
+                                        AtomicReference<Boolean> keepCosuming = new AtomicReference<>(true);
+                                        Instant start = Instant.now();
+                                        while(keepCosuming.get()){
+                                            if (Duration.between(start, Instant.now()).toMillis() > 60000) {
+                                                keepCosuming.set(false);
+                                            }
+                                            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
+                                            records.forEach(record ->{
+                                                String message = record.value();
+                                                if (message.contains("/http/order/findById?id=23")) {
+                                                    isLog.set(true);
+                                                    keepCosuming.set(false);
+                                                }
+                                            });
+                                        }
+                                        Assertions.assertTrue(isLog.get());
+                                    } catch (InterruptedException e) {
+                                        LOG.info("isLog.get():{}", isLog.get());
+                                        LOG.error("error", e);
+                                        throw new RuntimeException(e);
+                                    }
+                                }).build()
+                ).build();
+    }
+
     private ShenYuScenarioSpec testRocketMQHello() {
         return ShenYuScenarioSpec.builder()
                 .name("testRocketMQHello")

shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-http/src/test/java/org/apache/shenyu/e2e/testcase/http/DividePluginTest.java

@@ -110,6 +110,16 @@ void setup(final AdminClient adminClient, final GatewayClient gatewayClient) thr
         formData.add("config", "{\"topic\":\"shenyu-access-logging\", \"namesrvAddr\": \"rocketmq-dialevoneid:9876\",\"producerGroup\":\"shenyu-plugin-logging-rocketmq\"}");
         adminClient.changePluginStatus("29", formData);
         WaitDataSync.waitGatewayPluginUse(gatewayClient, "org.apache.shenyu.plugin.logging.rocketmq");
+        LOG.info("start loggingKafka plugin");
+        formData.add("id","33");
+        formData.add("name","loggingKafka");
+        formData.add("enabled","true");
+        formData.add("role","Logging");
+        formData.add("sort","180");
+        formData.add("config","{\"topic\":\"shenyu-access-logging\",\"namesrvAddr\":\"kafka:9092\",\"sampleRate\":\"1\",\"maxResponseBody\":524288,\"maxRequestBody\":524288,\"compressAlg\":\"none\"}");
+        adminClient.changePluginStatus("33",formData);
+        WaitDataSync.waitGatewayPluginUse(gatewayClient,"org.apache.shenyu.plugin.logging.kafka");
+
     }
 
     @ShenYuScenario(provider = DividePluginCases.class)