Repository: apache/eventmesh
Issue: [Bug] v1.7.0 KafkaConsumerRunner.java Use break to break out of the while

### 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

Other

### What happened

Use break to break out of the while Cause the consumer thread to stop


### How to reproduce

.

### Debug logs

_No response_

### Are you willing to submit PR?

- [ ] 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

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 26b86e525..300289254 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
@@ -58,8 +58,8 @@ public class KafkaConsumerRunner implements Runnable {
 
     @Override
     public void run() {
-        while (!closed.get()) {
-            try {
+        try {
+            while (!closed.get()) {
                 ConsumerRecords<String, CloudEvent> records = consumer.poll(Duration.ofMillis(10000));
                 // Handle new records
                 records.forEach(rec -> {
@@ -73,7 +73,7 @@ public class KafkaConsumerRunner implements Runnable {
                                     case CommitMessage:
                                         // update offset
                                         logger.info("message commit, topic: {}, current offset:{}", topicName,
-                                            rec.offset());
+                                                rec.offset());
                                         break;
                                     case ReconsumeLater:
                                         // don't update offset
@@ -81,7 +81,7 @@ public class KafkaConsumerRunner implements Runnable {
                                     case ManualAck:
                                         // update offset
                                         logger
-                                            .info("message ack, topic: {}, current offset:{}", topicName, rec.offset());
+                                                .info("message ack, topic: {}, current offset:{}", topicName, rec.offset());
                                         break;
                                     default:
                                 }
@@ -95,15 +95,14 @@ public class KafkaConsumerRunner implements Runnable {
                         logger.info("Error parsing cloudevents: {}", e.getMessage());
                     }
                 });
-            } catch (WakeupException e) {
-                // Ignore exception if closing
-                if (!closed.get()) {
-                    throw e;
-                }
-            } finally {
-                consumer.close();
-                break;
             }
+        } catch (WakeupException e) {
+            // Ignore exception if closing
+            if (!closed.get()) {
+                throw e;
+            }
+        } finally {
+            consumer.close();
         }
     }