Merge 0b0786c into 6636e3a

Markliniubility · web-flow · commit d9e7030aabe9 · 2023-03-01T10:16:42.000Z
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
@@ -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 ConsumerImpl(final Properties properties) {
         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
@@ -62,6 +62,10 @@ public long getOffset(CloudEvent cloudEvent) {
     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 void commit(EventMeshAction action) {
                                         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 void shutdown() {
         closed.set(true);
         consumer.wakeup();
     }
-}
-
-
-
+}
