Repository: apache/eventmesh
Issue: [Bug] KafkaConsumerRunner Thread Closes Unexceptedly  

### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/eventmesh/issues?q=is%3Aissue) and found no similar issues.


### Environment

Mac

### EventMesh version

master

### What happened

See below for the `how to reproduce` section

### How to reproduce

0. Start your Kafka
1. Run the `StartUp` of EventMesh
2. Run the `SpringBootDemoApplication` in the HTTP demo
3. Run the `AsyncPublishInsatance` message in HTTP to publish a message

The KafkaConsumerRunner will fail to poll the message and close

### Debug logs

No immediate exception logs

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

Changed files:
eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/KafkaConsumerRunner.java
eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/ConsumerImpl.java

diff --git a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/ConsumerImpl.java b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/ConsumerImpl.java
index eb370c459..6906474c5 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/ConsumerImpl.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/ConsumerImpl.java
@@ -53,12 +53,14 @@ public class ConsumerImpl {
     private Set<String> topicsSet;
 
     public ConsumerImpl(final Properties properties) {
+        // Setting the ClassLoader to null is necessary for Kafka consumer configuration
+        final ClassLoader original = Thread.currentThread().getContextClassLoader();
+        Thread.currentThread().setContextClassLoader(null);
+        
         Properties props = new Properties();
-
-        // Other config props
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CloudEventDeserializer.class);
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CloudEventDeserializer.class);
         props.put(ConsumerConfig.GROUP_ID_CONFIG, properties.getProperty(ConsumerConfig.GROUP_ID_CONFIG));
         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
 
@@ -67,6 +69,8 @@ public class ConsumerImpl {
         kafkaConsumerRunner = new KafkaConsumerRunner(this.kafkaConsumer);
         executorService = Executors.newFixedThreadPool(10);
         topicsSet = new HashSet<>();
+
+        Thread.currentThread().setContextClassLoader(original);
     }
 
     public Properties attributes() {
diff --git a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/KafkaConsumerRunner.java b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/KafkaConsumerRunner.java
index bed94c650..381d1f206 100644
--- a/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/KafkaConsumerRunner.java
+++ b/eventmesh-connector-plugin/eventmesh-connector-kafka/src/main/java/org/apache/eventmesh/connector/kafka/consumer/KafkaConsumerRunner.java
@@ -62,6 +62,10 @@ public class KafkaConsumerRunner implements Runnable {
     public void run() {
         try {
             while (!closed.get()) {
+                if (consumer.subscription().isEmpty()) {
+                    // consumer cannot poll if it is subscribe to nothing
+                    continue;
+                }
                 ConsumerRecords<String, CloudEvent> records = consumer.poll(Duration.ofMillis(10000));
                 // Handle new records
                 records.forEach(rec -> {
@@ -82,8 +86,7 @@ public class KafkaConsumerRunner implements Runnable {
                                         break;
                                     case ManualAck:
                                         // update offset
-                                        log
-                                            .info("message ack, topic: {}, current offset:{}", topicName, rec.offset());
+                                        log.info("message ack, topic: {}, current offset:{}", topicName, rec.offset());
                                         break;
                                     default:
                                 }
@@ -113,7 +116,4 @@ public class KafkaConsumerRunner implements Runnable {
         closed.set(true);
         consumer.wakeup();
     }
-}
-
-
-
+}
\\ No newline at end of file