This is an automated email from the ASF dual-hosted git repository.
chenguangsheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 80674cd20 [ISSUE #3375] Add eventmesh-storage-kafka module
new 3173688bf Merge pull request #3376 from xwm1992/add-storage-kafka
80674cd20 is described below
commit 80674cd20950d8b7589a6a3c462729e92caceced
Author: xwm1992 <[email protected]>
AuthorDate: Thu Mar 9 14:41:22 2023 +0800
[ISSUE #3375] Add eventmesh-storage-kafka module
---
.../eventmesh-storage-kafka/build.gradle | 47 +++++++
.../eventmesh-storage-kafka/gradle.properties | 20 +++
.../eventmesh/storage/kafka/common/Constants.java | 23 ++++
.../storage/kafka/common/EventMeshConstants.java | 27 ++++
.../storage/kafka/config/ClientConfiguration.java | 98 ++++++++++++++
.../storage/kafka/consumer/ConsumerImpl.java | 148 +++++++++++++++++++++
.../storage/kafka/consumer/KafkaConsumerImpl.java | 97 ++++++++++++++
.../kafka/consumer/KafkaConsumerRunner.java | 119 +++++++++++++++++
.../storage/kafka/producer/KafkaProducerImpl.java | 104 +++++++++++++++
.../storage/kafka/producer/ProducerImpl.java | 133 ++++++++++++++++++
.../storage/StorageResourceServiceKafkaImpl.java | 33 +++++
.../org.apache.eventmesh.api.consumer.Consumer | 16 +++
.../org.apache.eventmesh.api.producer.Producer | 16 +++
...he.eventmesh.api.storage.StorageResourceService | 16 +++
.../src/main/resources/kafka-client.properties | 135 +++++++++++++++++++
.../kafka/config/ClientConfigurationTest.java | 70 ++++++++++
.../org.apache.io.openmessaging.producer.Producer | 20 +++
.../src/test/resources/kafka-client.properties | 48 +++++++
settings.gradle | 2 +
19 files changed, 1172 insertions(+)
diff --git a/eventmesh-storage/eventmesh-storage-kafka/build.gradle
b/eventmesh-storage/eventmesh-storage-kafka/build.gradle
new file mode 100644
index 000000000..9f100e129
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-kafka/build.gradle
@@ -0,0 +1,47 @@
+/*
+ * 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.
+ */
+configurations {
+ implementation.exclude group: 'ch.qos.logback', module: 'logback-classic'
+ implementation.exclude group: 'log4j', module: 'log4j'
+}
+
+dependencies {
+ implementation project(":eventmesh-storage:eventmesh-storage-api")
+ implementation project(":eventmesh-common")
+ // https://mavenlibs.com/maven/dependency/io.cloudevents/cloudevents-kafka
+ implementation group: 'io.cloudevents', name: 'cloudevents-kafka',
version: '2.2.1'
+
+ // https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
+ implementation 'org.apache.kafka:kafka-clients:3.0.0'
+
+ testImplementation 'org.junit.jupiter:junit-jupiter-api:5.6.0'
+ testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
+
+ testImplementation project(":eventmesh-storage:eventmesh-storage-api")
+ testImplementation project(":eventmesh-common")
+
+ testImplementation "org.mockito:mockito-core"
+ testImplementation "org.powermock:powermock-module-junit4"
+ testImplementation "org.powermock:powermock-api-mockito2"
+
+ compileOnly 'org.projectlombok:lombok'
+ compileOnly 'com.google.code.findbugs:jsr305:3.0.1'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testCompileOnly 'org.projectlombok:lombok'
+ testAnnotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-storage/eventmesh-storage-kafka/gradle.properties
b/eventmesh-storage/eventmesh-storage-kafka/gradle.properties
new file mode 100644
index 000000000..84cc1ec8d
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-kafka/gradle.properties
@@ -0,0 +1,20 @@
+# 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.
+#
+
+kafka_version=3.2.0
+
+pluginType=storage
+pluginName=kafka
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/Constants.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/Constants.java
new file mode 100644
index 000000000..3e6e4c800
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/Constants.java
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.common;
+
+public class Constants {
+
+ public static final String BROADCAST_PREFIX = "broadcast-";
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
new file mode 100644
index 000000000..2bc2f3c04
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
@@ -0,0 +1,27 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.common;
+
+public class EventMeshConstants {
+
+ public static final String EVENTMESH_CONF_FILE = "kafka-client.properties";
+
+ public static final int DEFAULT_TIMEOUT_IN_MILLISECONDS = 3000;
+
+ public static final String STORE_TIMESTAMP = "storetime";
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
new file mode 100644
index 000000000..3749e8157
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
@@ -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.
+ */
+
+package org.apache.eventmesh.storage.kafka.config;
+
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigFiled;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Config(prefix = "eventMesh.server.kafka", path =
"classPath://kafka-client.properties")
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class ClientConfiguration {
+
+ @ConfigFiled(field = "namesrvAddr", notEmpty = true)
+ @Builder.Default
+ private String namesrvAddr = "";
+
+ @ConfigFiled(field = "username")
+ @Builder.Default
+ private String clientUserName = "username";
+
+ @ConfigFiled(field = "password")
+ @Builder.Default
+ private String clientPass = "password";
+
+ @ConfigFiled(field = "client.consumeThreadMin")
+ @Builder.Default
+ private Integer consumeThreadMin = 2;
+
+ @ConfigFiled(field = "client.consumeThreadMax")
+ @Builder.Default
+ private Integer consumeThreadMax = 2;
+
+ @ConfigFiled(field = "client.consumeThreadPoolQueueSize")
+ @Builder.Default
+ private Integer consumeQueueSize = 10000;
+
+ @ConfigFiled(field = "client.pullBatchSize")
+ @Builder.Default
+ private Integer pullBatchSize = 32;
+
+ @ConfigFiled(field = "client.ackwindow")
+ @Builder.Default
+ private Integer ackWindow = 1000;
+
+ @ConfigFiled(field = "client.pubwindow")
+ @Builder.Default
+ private Integer pubWindow = 100;
+
+ @ConfigFiled(field = "client.comsumeTimeoutInMin")
+ @Builder.Default
+ private long consumeTimeout = 0L;
+
+ @ConfigFiled(field = "client.pollNameServerInterval")
+ @Builder.Default
+ private Integer pollNameServerInterval = 10 * 1000;
+
+ @ConfigFiled(field = "client.heartbeatBrokerInterval")
+ @Builder.Default
+ private Integer heartbeatBrokerInterval = 30 * 1000;
+
+ @ConfigFiled(field = "client.rebalanceInterval")
+ @Builder.Default
+ private Integer rebalanceInterval = 20 * 1000;
+
+ @ConfigFiled(field = "cluster")
+ @Builder.Default
+ private String clusterName = "";
+
+ @ConfigFiled(field = "accessKey")
+ @Builder.Default
+ private String accessKey = "";
+
+ @ConfigFiled(field = "secretKey")
+ @Builder.Default
+ private String secretKey = "";
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/ConsumerImpl.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/ConsumerImpl.java
new file mode 100644
index 000000000..92a5b86ec
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/ConsumerImpl.java
@@ -0,0 +1,148 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.consumer;
+
+import org.apache.eventmesh.api.AbstractContext;
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.kafka.CloudEventDeserializer;
+
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class ConsumerImpl {
+
+ private final KafkaConsumer<String, CloudEvent> kafkaConsumer;
+ private final Properties properties;
+ private AtomicBoolean started = new AtomicBoolean(false);
+ private EventListener eventListener;
+ private KafkaConsumerRunner kafkaConsumerRunner;
+ private ExecutorService executorService;
+ 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();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
properties.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
+ 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");
+
+ this.properties = props;
+ this.kafkaConsumer = new KafkaConsumer<String, CloudEvent>(props);
+ kafkaConsumerRunner = new KafkaConsumerRunner(this.kafkaConsumer);
+ executorService = Executors.newFixedThreadPool(10);
+ topicsSet = new HashSet<>();
+
+ Thread.currentThread().setContextClassLoader(original);
+ }
+
+ public Properties attributes() {
+ return properties;
+ }
+
+ public void start() {
+ if (this.started.compareAndSet(false, true)) {
+ executorService.submit(kafkaConsumerRunner);
+ }
+ }
+
+
+ public synchronized void shutdown() {
+ if (this.started.compareAndSet(true, false)) {
+ // Shutdown the executor and interrupt any running tasks
+ kafkaConsumerRunner.shutdown();
+ executorService.shutdown();
+ }
+ }
+
+
+ public boolean isStarted() {
+ return this.started.get();
+ }
+
+
+ public boolean isClosed() {
+ return !this.isStarted();
+ }
+
+ public KafkaConsumer<String, CloudEvent> getKafkaConsumer() {
+ return kafkaConsumer;
+ }
+
+ public synchronized void subscribe(String topic) {
+ try {
+ // Get the current subscription
+ topicsSet.add(topic);
+ List<String> topics = new ArrayList<>(topicsSet);
+ this.kafkaConsumer.subscribe(topics);
+ } catch (Exception e) {
+ log.error("Error while subscribing the Kafka consumer to topic: ",
e);
+ throw new StorageConnectorRuntimeException(
+ String.format("Kafka consumer can't attach to %s.", topic));
+ }
+ }
+
+ public synchronized void unsubscribe(String topic) {
+ try {
+ // Kafka will unsubscribe *all* topic if calling unsubscribe, so we
+ this.kafkaConsumer.unsubscribe();
+ topicsSet.remove(topic);
+ List<String> topics = new ArrayList<>(topicsSet);
+ this.kafkaConsumer.subscribe(topics);
+ } catch (Exception e) {
+ log.error("Error while unsubscribing the Kafka consumer: ", e);
+ throw new StorageConnectorRuntimeException(String.format("kafka
push consumer fails to unsubscribe topic: %s", topic));
+ }
+ }
+
+ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext
context) {
+ Long maxOffset = cloudEvents.stream().map(cloudEvent ->
this.kafkaConsumerRunner.getOffset(cloudEvent)).max(Long::compare).get();
+ cloudEvents.forEach(cloudEvent ->
this.updateOffset(cloudEvent.getSubject(), maxOffset));
+ }
+
+ public void updateOffset(String topicName, long offset) {
+ this.kafkaConsumer.seek(new TopicPartition(topicName, 1), offset);
+ }
+
+ public void registerEventListener(EventListener listener) {
+ this.eventListener = listener;
+ this.kafkaConsumerRunner.setListener(this.eventListener);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
new file mode 100644
index 000000000..0c88ec934
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerImpl.java
@@ -0,0 +1,97 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.consumer;
+
+import org.apache.eventmesh.api.AbstractContext;
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.consumer.Consumer;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.kafka.config.ClientConfiguration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+import java.util.List;
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+
+@Config(field = "clientConfiguration")
+public class KafkaConsumerImpl implements Consumer {
+
+ private ConsumerImpl consumer;
+
+ /**
+ * Unified configuration class corresponding to kafka-client.properties
+ */
+ private ClientConfiguration clientConfiguration;
+
+ @Override
+ public synchronized void init(Properties props) throws Exception {
+ String namesrvAddr = clientConfiguration.getNamesrvAddr();
+ String consumerGroup = props.getProperty("consumerGroup");
+
+ // Other config props
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, namesrvAddr);
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
+ consumer = new ConsumerImpl(props);
+ }
+
+ @Override
+ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext
context) {
+ consumer.updateOffset(cloudEvents, context);
+ }
+
+ @Override
+ public void subscribe(String topic) throws Exception {
+ consumer.subscribe(topic);
+ }
+
+ @Override
+ public boolean isStarted() {
+ return consumer.isStarted();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return consumer.isClosed();
+ }
+
+ @Override
+ public synchronized void start() {
+ consumer.start();
+ }
+
+ @Override
+ public void unsubscribe(String topic) {
+ consumer.unsubscribe(topic);
+ }
+
+ @Override
+ public void registerEventListener(EventListener listener) {
+ consumer.registerEventListener(listener);
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ consumer.shutdown();
+ }
+
+ public ClientConfiguration getClientConfiguration() {
+ return this.clientConfiguration;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java
new file mode 100644
index 000000000..2ba0e6d34
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/consumer/KafkaConsumerRunner.java
@@ -0,0 +1,119 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.consumer;
+
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
+
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.errors.WakeupException;
+
+import java.time.Duration;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+
+import io.cloudevents.CloudEvent;
+
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class KafkaConsumerRunner implements Runnable {
+
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final KafkaConsumer<String, CloudEvent> consumer;
+ private ConcurrentHashMap<CloudEvent, Long> cloudEventToOffset;
+ private EventListener listener;
+ private AtomicInteger offset;
+
+ public KafkaConsumerRunner(KafkaConsumer<String, CloudEvent>
kafkaConsumer) {
+ this.consumer = kafkaConsumer;
+ cloudEventToOffset = new ConcurrentHashMap<>();
+ }
+
+ public synchronized void setListener(EventListener listener) {
+ this.listener = listener;
+ }
+
+ public long getOffset(CloudEvent cloudEvent) {
+ return cloudEventToOffset.getOrDefault(cloudEvent, 0L);
+ }
+
+ @Override
+ 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 -> {
+ try {
+ CloudEvent cloudEvent = rec.value();
+ String topicName = cloudEvent.getSubject();
+ EventMeshAsyncConsumeContext
eventMeshAsyncConsumeContext = new EventMeshAsyncConsumeContext() {
+ @Override
+ public void commit(EventMeshAction action) {
+ switch (action) {
+ case CommitMessage:
+ // update offset
+ log.info("message commit, topic: {},
current offset:{}", topicName,
+ rec.offset());
+ break;
+ case ReconsumeLater:
+ // don't update offset
+ break;
+ case ManualAck:
+ // update offset
+ log.info("message ack, topic: {},
current offset:{}", topicName, rec.offset());
+ break;
+ default:
+ }
+ }
+ };
+ cloudEventToOffset.put(cloudEvent, rec.offset());
+ if (listener != null) {
+ listener.consume(cloudEvent,
eventMeshAsyncConsumeContext);
+ }
+ } catch (Exception e) {
+ log.info("Error parsing cloudevents: {}",
e.getMessage());
+ }
+ });
+ }
+ } catch (WakeupException e) {
+ // Ignore exception if closing
+ if (!closed.get()) {
+ throw e;
+ }
+ } finally {
+ consumer.close();
+ }
+ }
+
+ // Shutdown hook which can be called from a separate thread
+ public void shutdown() {
+ closed.set(true);
+ consumer.wakeup();
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/producer/KafkaProducerImpl.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/producer/KafkaProducerImpl.java
new file mode 100644
index 000000000..fbc9352a8
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/producer/KafkaProducerImpl.java
@@ -0,0 +1,104 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.producer;
+
+import org.apache.eventmesh.api.RequestReplyCallback;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.kafka.config.ClientConfiguration;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+
+@Config(field = "clientConfiguration")
+public class KafkaProducerImpl implements Producer {
+
+ private ProducerImpl producer;
+
+ /**
+ * Unified configuration class corresponding to kafka-client.properties
+ */
+ private ClientConfiguration clientConfiguration;
+
+ @Override
+ public synchronized void init(Properties keyValue) {
+ String namesrvAddr = clientConfiguration.getNamesrvAddr();
+
+ keyValue.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, namesrvAddr);
+ this.producer = new ProducerImpl(keyValue);
+ }
+
+ @Override
+ public boolean isStarted() {
+ return producer.isStarted();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return producer.isClosed();
+ }
+
+ @Override
+ public void start() {
+ producer.start();
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ producer.shutdown();
+ }
+
+ @Override
+ public void publish(CloudEvent message, SendCallback sendCallback) throws
Exception {
+ producer.sendAsync(message, sendCallback);
+ }
+
+ @Override
+ public void request(CloudEvent message, RequestReplyCallback rrCallback,
long timeout) throws Exception {
+ producer.request(message, rrCallback, timeout);
+ }
+
+ @Override
+ public boolean reply(final CloudEvent message, final SendCallback
sendCallback) throws Exception {
+ producer.reply(message, sendCallback);
+ return true;
+ }
+
+ @Override
+ public void checkTopicExist(String topic) throws Exception {
+ this.producer.checkTopicExist(topic);
+ }
+
+ @Override
+ public void setExtFields() {
+ // producer.setExtFields();
+ }
+
+ @Override
+ public void sendOneway(CloudEvent message) {
+ producer.sendOneway(message);
+ }
+
+ public ClientConfiguration getClientConfiguration() {
+ return this.clientConfiguration;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/producer/ProducerImpl.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/producer/ProducerImpl.java
new file mode 100644
index 000000000..1d661dcc9
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/producer/ProducerImpl.java
@@ -0,0 +1,133 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.producer;
+
+import org.apache.eventmesh.api.RequestReplyCallback;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+import java.util.Objects;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.kafka.CloudEventSerializer;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@SuppressWarnings("deprecation")
+public class ProducerImpl {
+
+ private KafkaProducer<String, CloudEvent> producer;
+ Properties properties;
+
+ private AtomicBoolean isStarted;
+
+ public ProducerImpl(Properties props) {
+ this.isStarted = new AtomicBoolean(false);
+
+ properties = new Properties();
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
+ props.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
+ properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
+ properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
CloudEventSerializer.class);
+ this.producer = new KafkaProducer<>(properties);
+ }
+
+ public boolean isStarted() {
+ return isStarted.get();
+ }
+
+ public boolean isClosed() {
+ return !isStarted.get();
+ }
+
+ public void start() {
+ isStarted.compareAndSet(false, true);
+ }
+
+ public void shutdown() {
+ isStarted.compareAndSet(true, false);
+ }
+
+ public ProducerImpl init(Properties properties) throws Exception {
+ return new ProducerImpl(properties);
+ }
+
+ public void send(CloudEvent cloudEvent) {
+ try {
+ this.producer.send(new
ProducerRecord<>(Objects.requireNonNull(cloudEvent.getSubject()), cloudEvent));
+ } catch (Exception e) {
+ log.error(String.format("Send message oneway Exception, %s",
cloudEvent), e);
+ }
+ }
+
+ public void checkTopicExist(String topic) throws ExecutionException,
InterruptedException, StorageConnectorRuntimeException {
+ Admin admin = Admin.create(properties);
+ Set<String> topicNames = admin.listTopics().names().get();
+ admin.close();
+ boolean exist = topicNames.contains(topic);
+ if (!exist) {
+ throw new StorageConnectorRuntimeException(String.format("topic:%s
is not exist", topic));
+ }
+ }
+
+ public void request(CloudEvent cloudEvent, RequestReplyCallback
rrCallback, long timeout) throws Exception {
+ throw new StorageConnectorRuntimeException("Request is not supported");
+ }
+
+ public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ throw new StorageConnectorRuntimeException("Reply is not supported");
+ }
+
+ public void sendOneway(CloudEvent message) {
+ }
+
+ public void sendAsync(CloudEvent cloudEvent, SendCallback sendCallback) {
+ try {
+ this.producer.send(new ProducerRecord<>(cloudEvent.getSubject(),
cloudEvent), (metadata, exception) -> {
+ if (exception != null) {
+ StorageConnectorRuntimeException onsEx = new
StorageConnectorRuntimeException(exception.getMessage(), exception);
+ OnExceptionContext context = new OnExceptionContext();
+ context.setTopic(cloudEvent.getSubject());
+ context.setException(onsEx);
+ sendCallback.onException(context);
+ } else {
+ SendResult sendResult = new SendResult();
+ sendResult.setTopic(cloudEvent.getSubject());
+ sendResult.setMessageId(cloudEvent.getId());
+ sendCallback.onSuccess(sendResult);
+ }
+ });
+ } catch (Exception e) {
+ log.error(String.format("Send message oneway Exception, %s",
cloudEvent), e);
+ }
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/StorageResourceServiceKafkaImpl.java
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/StorageResourceServiceKafkaImpl.java
new file mode 100644
index 000000000..5ee4b8eb8
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/storage/StorageResourceServiceKafkaImpl.java
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.storage;
+
+import org.apache.eventmesh.api.storage.StorageResourceService;
+
+public class StorageResourceServiceKafkaImpl implements StorageResourceService
{
+
+ @Override
+ public void init() throws Exception {
+
+ }
+
+ @Override
+ public void release() throws Exception {
+
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
new file mode 100644
index 000000000..015cc5c7a
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
@@ -0,0 +1,16 @@
+# 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.
+
+kafka=org.apache.eventmesh.storage.kafka.consumer.KafkaConsumerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
new file mode 100644
index 000000000..15ad33471
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
@@ -0,0 +1,16 @@
+# 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.
+
+kafka=org.apache.eventmesh.storage.kafka.producer.KafkaProducerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
new file mode 100644
index 000000000..6b314f5c5
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
@@ -0,0 +1,16 @@
+# 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.
+
+kafka=org.apache.eventmesh.storage.kafka.storage.StorageResourceServiceKafkaImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
new file mode 100644
index 000000000..31923d73d
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
@@ -0,0 +1,135 @@
+#
+# 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.
+#
+#######################kafka-client##################
+eventMesh.server.kafka.namesrvAddr=127.0.0.1:9092;127.0.0.1:9092
+eventMesh.server.kafka.cluster=DefaultCluster
+eventMesh.server.kafka.accessKey=********
+eventMesh.server.kafka.secretKey=********
+
+# see kafka.server.KafkaConfig for additional details and defaults
+
+############################# Server Basics #############################
+
+# The id of the broker. This must be set to a unique integer for each broker.
+broker.id=0
+
+############################# Socket Server Settings
#############################
+
+# The port the socket server listens on
+eventMesh.server.kafka.port=9092
+eventMesh.server.kafka.bootstrap.servers=localhost:9092
+
+# Hostname the broker will bind to. If not set, the server will bind to all
interfaces
+#host.name=localhost
+
+# Hostname the broker will advertise to producers and consumers. If not set,
it uses the
+# value for "host.name" if configured. Otherwise, it will use the value
returned from
+# java.net.InetAddress.getCanonicalHostName().
+#advertised.host.name=<hostname routable by clients>
+
+# The port to publish to ZooKeeper for clients to use. If this is not set,
+# it will publish the same port that the broker binds to.
+#advertised.port=<port accessible by clients>
+
+# The number of threads handling network requests
+num.network.threads=2
+
+# The number of threads doing disk I/O
+num.io.threads=8
+
+# The send buffer (SO_SNDBUF) used by the socket server
+socket.send.buffer.bytes=1048576
+
+# The receive buffer (SO_RCVBUF) used by the socket server
+socket.receive.buffer.bytes=1048576
+
+# The maximum size of a request that the socket server will accept (protection
against OOM)
+socket.request.max.bytes=104857600
+
+
+############################# Log Basics #############################
+
+# A comma seperated list of directories under which to store log files
+log.dirs=c:/tmp/kafka-logs/broker0
+
+# The default number of log partitions per topic. More partitions allow greater
+# parallelism for consumption, but this will also result in more files across
+# the brokers.
+num.partitions=2
+
+############################# Log Flush Policy #############################
+
+# Messages are immediately written to the filesystem but by default we only
fsync() to sync
+# the OS cache lazily. The following configurations control the flush of data
to disk.
+# There are a few important trade-offs here:
+# 1. Durability: Unflushed data may be lost if you are not using
replication.
+# 2. Latency: Very large flush intervals may lead to latency spikes when
the flush does occur as there will be a lot of data to flush.
+# 3. Throughput: The flush is generally the most expensive operation, and a
small flush interval may lead to exceessive seeks.
+# The settings below allow one to configure the flush policy to flush data
after a period of time or
+# every N messages (or both). This can be done globally and overridden on a
per-topic basis.
+
+# The number of messages to accept before forcing a flush of data to disk
+#log.flush.interval.messages=10000
+
+# The maximum amount of time a message can sit in a log before we force a flush
+#log.flush.interval.ms=1000
+
+############################# Log Retention Policy
#############################
+
+# The following configurations control the disposal of log segments. The
policy can
+# be set to delete segments after a period of time, or after a given size has
accumulated.
+# A segment will be deleted whenever *either* of these criteria are met.
Deletion always happens
+# from the end of the log.
+
+# The minimum age of a log file to be eligible for deletion
+log.retention.hours=168
+
+# A size-based retention policy for logs. Segments are pruned from the log as
long as the remaining
+# segments don't drop below log.retention.bytes.
+#log.retention.bytes=1073741824
+
+# The maximum size of a log segment file. When this size is reached a new log
segment will be created.
+log.segment.bytes=536870912
+
+# The interval at which log segments are checked to see if they can be deleted
according
+# to the retention policies
+log.retention.check.interval.ms=60000
+
+# By default the log cleaner is disabled and the log retention policy will
default to just delete segments after their retention expires.
+# If log.cleaner.enable=true is set the cleaner will be enabled and individual
logs can then be marked for log compaction.
+log.cleaner.enable=false
+
+############################# Zookeeper #############################
+
+# Zookeeper connection string (see zookeeper docs for details).
+# This is a comma separated host:port pairs, each corresponding to a zk
+# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
+# You can also append an optional chroot string to the urls to specify the
+# root directory for all kafka znodes.
+zookeeper.connect=localhost:2181
+
+# Timeout in ms for connecting to zookeeper
+zookeeper.connection.timeout.ms=1000000
+
+############################# Producer Basics #############################
+
+# list of brokers used for bootstrapping knowledge about the rest of the
cluster
+# format: host1:port1,host2:port2 ...
+metadata.broker.list=localhost:9092
+
+# specifies whether the messages are sent asynchronously (async) or
synchronously (sync)
+producer.type=async
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/test/java/org/apache/eventmesh/storage/kafka/config/ClientConfigurationTest.java
b/eventmesh-storage/eventmesh-storage-kafka/src/test/java/org/apache/eventmesh/storage/kafka/config/ClientConfigurationTest.java
new file mode 100644
index 000000000..9b5e494ce
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/test/java/org/apache/eventmesh/storage/kafka/config/ClientConfigurationTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+package org.apache.eventmesh.storage.kafka.config;
+
+import org.apache.eventmesh.api.factory.StoragePluginFactory;
+import org.apache.eventmesh.storage.kafka.consumer.KafkaConsumerImpl;
+import org.apache.eventmesh.storage.kafka.producer.KafkaProducerImpl;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * test config of Kafka SPI Impl
+ */
+public class ClientConfigurationTest {
+
+ /**
+ * test KafkaConsumerImpl config init when ConnectorPluginFactory get it
+ */
+ @Test
+ public void getConfigWhenKafkaConsumerImplInit() {
+ KafkaConsumerImpl consumer = (KafkaConsumerImpl)
StoragePluginFactory.getMeshMQPushConsumer("kafka");
+ ClientConfiguration config = consumer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ /**
+ * test KafkaProducerImpl config init when ConnectorPluginFactory get it
+ */
+ @Test
+ public void getConfigWhenKafkaProducerImplInit() {
+ KafkaProducerImpl producer = (KafkaProducerImpl)
StoragePluginFactory.getMeshMQProducer("kafka");
+ ClientConfiguration config = producer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ private void assertConfig(ClientConfiguration config) {
+ Assert.assertEquals(config.getNamesrvAddr(),
"127.0.0.1:9092;127.0.0.1:9092");
+ Assert.assertEquals(config.getClientUserName(), "username-succeed!!!");
+ Assert.assertEquals(config.getClientPass(), "password-succeed!!!");
+ Assert.assertEquals(config.getConsumeThreadMin(),
Integer.valueOf(1816));
+ Assert.assertEquals(config.getConsumeThreadMax(),
Integer.valueOf(2816));
+ Assert.assertEquals(config.getConsumeQueueSize(),
Integer.valueOf(3816));
+ Assert.assertEquals(config.getPullBatchSize(), Integer.valueOf(4816));
+ Assert.assertEquals(config.getAckWindow(), Integer.valueOf(5816));
+ Assert.assertEquals(config.getPubWindow(), Integer.valueOf(6816));
+ Assert.assertEquals(config.getConsumeTimeout(), 7816);
+ Assert.assertEquals(config.getPollNameServerInterval(),
Integer.valueOf(8816));
+ Assert.assertEquals(config.getHeartbeatBrokerInterval(),
Integer.valueOf(9816));
+ Assert.assertEquals(config.getRebalanceInterval(),
Integer.valueOf(11816));
+ Assert.assertEquals(config.getClusterName(), "cluster-succeed!!!");
+ Assert.assertEquals(config.getAccessKey(), "accessKey-succeed!!!");
+ Assert.assertEquals(config.getSecretKey(), "secretKey-succeed!!!");
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/test/resources/META-INF/services/org.apache.io.openmessaging.producer.Producer
b/eventmesh-storage/eventmesh-storage-kafka/src/test/resources/META-INF/services/org.apache.io.openmessaging.producer.Producer
new file mode 100644
index 000000000..358b84da4
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/test/resources/META-INF/services/org.apache.io.openmessaging.producer.Producer
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.apache.eventmesh.connector.kafka.producer.ProducerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-kafka/src/test/resources/kafka-client.properties
b/eventmesh-storage/eventmesh-storage-kafka/src/test/resources/kafka-client.properties
new file mode 100644
index 000000000..3b0c217e0
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-kafka/src/test/resources/kafka-client.properties
@@ -0,0 +1,48 @@
+#
+# 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.
+#
+#######################kafka-client##################
+eventMesh.server.kafka.namesrvAddr=127.0.0.1:9092;127.0.0.1:9092
+eventMesh.server.kafka.username=username-succeed!!!
+eventMesh.server.kafka.password=password-succeed!!!
+eventMesh.server.kafka.client.consumeThreadMin=1816
+eventMesh.server.kafka.client.consumeThreadMax=2816
+eventMesh.server.kafka.client.consumeThreadPoolQueueSize=3816
+eventMesh.server.kafka.client.pullBatchSize=4816
+eventMesh.server.kafka.client.ackwindow=5816
+eventMesh.server.kafka.client.pubwindow=6816
+eventMesh.server.kafka.client.comsumeTimeoutInMin=7816
+eventMesh.server.kafka.client.pollNameServerInterval=8816
+eventMesh.server.kafka.client.heartbeatBrokerInterval=9816
+eventMesh.server.kafka.client.rebalanceInterval=11816
+eventMesh.server.kafka.cluster=cluster-succeed!!!
+eventMesh.server.kafka.accessKey=accessKey-succeed!!!
+eventMesh.server.kafka.secretKey=secretKey-succeed!!!
+eventMesh.server.kafka.username=username-succeed!!!
+eventMesh.server.kafka.password=password-succeed!!!
+eventMesh.server.kafka.client.consumeThreadMin=1816
+eventMesh.server.kafka.client.consumeThreadMax=2816
+eventMesh.server.kafka.client.consumeThreadPoolQueueSize=3816
+eventMesh.server.kafka.client.pullBatchSize=4816
+eventMesh.server.kafka.client.ackwindow=5816
+eventMesh.server.kafka.client.pubwindow=6816
+eventMesh.server.kafka.client.comsumeTimeoutInMin=7816
+eventMesh.server.kafka.client.pollNameServerInterval=8816
+eventMesh.server.kafka.client.heartbeatBrokerInterval=9816
+eventMesh.server.kafka.client.rebalanceInterval=11816
+eventMesh.server.kafka.cluster=cluster-succeed!!!
+eventMesh.server.kafka.accessKey=accessKey-succeed!!!
+eventMesh.server.kafka.secretKey=secretKey-succeed!!!
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index 28cd917b8..8456e9430 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -25,6 +25,8 @@ include 'eventmesh-examples'
include 'eventmesh-spi'
include 'eventmesh-storage:eventmesh-storage-api'
include 'eventmesh-storage:eventmesh-storage-standalone'
+include 'eventmesh-storage:eventmesh-storage-kafka'
+include 'eventmesh-storage:eventmesh-storage-redis'
include 'eventmesh-storage:eventmesh-storage-rocketmq'
include 'eventmesh-connector-plugin:eventmesh-connector-api'
include 'eventmesh-connector-plugin:eventmesh-connector-knative'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
