This is an automated email from the ASF dual-hosted git repository.
wqliang 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 93940d37f [ISSUE #3379] Add eventmesh-storage-pravega module
new a6b0a0abb Merge pull request #3381 from xwm1992/add-storage-pravega
93940d37f is described below
commit 93940d37f9300580f54c13625a5b16a24d5124f9
Author: xwm1992 <[email protected]>
AuthorDate: Thu Mar 9 16:41:51 2023 +0800
[ISSUE #3379] Add eventmesh-storage-pravega module
---
.../eventmesh-storage-pravega/README.md | 19 ++
.../eventmesh-storage-pravega/build.gradle | 39 ++++
.../eventmesh-storage-pravega/gradle.properties | 20 ++
.../eventmesh-storage-pravega/pravega-storage.jpg | Bin 0 -> 143684 bytes
.../storage/pravega/PravegaConsumerImpl.java | 109 ++++++++++
.../storage/pravega/PravegaProducerImpl.java | 121 +++++++++++
.../pravega/PravegaStorageResourceServiceImpl.java | 37 ++++
.../storage/pravega/client/PravegaClient.java | 233 +++++++++++++++++++++
.../pravega/client/PravegaCloudEventWriter.java | 70 +++++++
.../storage/pravega/client/PravegaEvent.java | 77 +++++++
.../storage/pravega/client/SubscribeTask.java | 89 ++++++++
.../pravega/config/PravegaStorageConfig.java | 67 ++++++
.../pravega/exception/PravegaStorageException.java | 31 +++
.../org.apache.eventmesh.api.consumer.Consumer | 16 ++
.../org.apache.eventmesh.api.producer.Producer | 16 ++
...he.eventmesh.api.storage.StorageResourceService | 16 ++
.../src/main/resources/pravega-storage.properties | 26 +++
.../storage/pravega/client/PravegaClientTest.java | 188 +++++++++++++++++
.../pravega/config/PravegaStorageConfigTest.java | 60 ++++++
.../src/test/resources/pravega-storage.properties | 26 +++
settings.gradle | 1 +
21 files changed, 1261 insertions(+)
diff --git a/eventmesh-storage/eventmesh-storage-pravega/README.md
b/eventmesh-storage/eventmesh-storage-pravega/README.md
new file mode 100644
index 000000000..92306965b
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-pravega/README.md
@@ -0,0 +1,19 @@
+# Pravega Connector
+Pravega connector of EventMesh supports taking Pravega as event storage.
[Pravega](https://cncf.pravega.io/) is a storage system that exposes Stream as
the main primitive for continuous and unbounded data and meets the requirement
of EventMesh connector module.
+
+## The Design of Pravega Connector
+Pravega has 3 level storage concepts, which are **Event**, **Stream**,
**Scope**. An **Event** is represented as a set of bytes within a Stream. A
**Stream** is a durable, elastic, append-only, unbounded sequence of bytes
having good performance and strong consistency. A **Scope** acts as a namespace
for Stream names and all Stream names are unique within their Scope.
+
+Naturally, Pravega connector makes each topic correspond to a pravega stream.
Because `EventStreamWriter` and `EventStreamReader` must assign to a special
stream in a special scope, each topic corresponds to an `EventStreamWriter` and
a `ReaderGroup`(only contains one `EventStreamReader`) as well. What's more,
`SubscribeTask` thread will read every topic stream continuously.
+
+**Note**:
+- Publish success will respond `-1` as messageId since writing to Pravega will
not return message offset of the stream. If you want to use EventMesh with
Pravega, please ensure messageId is unimportant.
+- Name and ID in Pravega must be `0-9`, `A-Z`, `a-z`, `.`, `-`.
+ - Topic must be legal in Pravega.
+ - ReaderGroup name format is `String.format("%s-%s", consumerGroup, topic)`
for cluster message model, while random UUID for broadcast message model.
+ - Reader name format is `String.format("%s-reader",
instanceName).replaceAll("\\(", "-").replaceAll("\\)", "-")`. The
`instanceName` consists of `consumerGroup`, config `eventMesh.server.cluster`,
EventMesh version and PID.
+
+
+
+## Active Pravega Connector
+Referring to doc [_Build and Load
Plugins_](https://eventmesh.apache.org/docs/installation/runtime#13-build-and-load-plugins)
to load `:eventmesh-connector-plugin:eventmesh-connector-pravega` plugin
module, and edit `eventMesh.connector.plugin.type=pravega` in config file
`eventmesh.properties` to active pravega connector.
\ No newline at end of file
diff --git a/eventmesh-storage/eventmesh-storage-pravega/build.gradle
b/eventmesh-storage/eventmesh-storage-pravega/build.gradle
new file mode 100644
index 000000000..f894a1120
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-pravega/build.gradle
@@ -0,0 +1,39 @@
+/*
+ * 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-common")
+ implementation project(":eventmesh-storage:eventmesh-storage-api")
+ implementation("io.pravega:pravega-client:$pravega_version") {
+ exclude group: 'io.netty', module: 'netty-codec-http2:4.1.30.Final'
+ }
+ // use newer version to avoid CVE
+ implementation 'io.netty:netty-codec-http2'
+
+ testImplementation "org.testcontainers:testcontainers:1.17.3"
+
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testCompileOnly 'org.projectlombok:lombok'
+ testAnnotationProcessor 'org.projectlombok:lombok'
+}
diff --git a/eventmesh-storage/eventmesh-storage-pravega/gradle.properties
b/eventmesh-storage/eventmesh-storage-pravega/gradle.properties
new file mode 100644
index 000000000..0bd75ae16
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-pravega/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.
+#
+
+pravega_version=0.11.0
+
+pluginType=storage
+pluginName=pravega
\ No newline at end of file
diff --git a/eventmesh-storage/eventmesh-storage-pravega/pravega-storage.jpg
b/eventmesh-storage/eventmesh-storage-pravega/pravega-storage.jpg
new file mode 100644
index 000000000..20cb82960
Binary files /dev/null and
b/eventmesh-storage/eventmesh-storage-pravega/pravega-storage.jpg differ
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
new file mode 100644
index 000000000..528be1302
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaConsumerImpl.java
@@ -0,0 +1,109 @@
+/*
+ * 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.pravega;
+
+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.pravega.client.PravegaClient;
+import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
+import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Config(field = "pravegaConnectorConfig")
+public class PravegaConsumerImpl implements Consumer {
+
+ private static final AtomicBoolean started = new AtomicBoolean(false);
+
+ /**
+ * Unified configuration class corresponding to pravega-storage.properties
+ */
+ private PravegaStorageConfig pravegaConnectorConfig;
+
+ private boolean isBroadcast;
+ private String instanceName;
+ private String consumerGroup;
+ private PravegaClient client;
+ private EventListener eventListener;
+
+ @Override
+ public void init(Properties keyValue) throws Exception {
+ isBroadcast = Boolean.parseBoolean(keyValue.getProperty("isBroadcast",
"false"));
+ instanceName = keyValue.getProperty("instanceName", "");
+ consumerGroup = keyValue.getProperty("consumerGroup", "");
+
+ client = PravegaClient.getInstance(pravegaConnectorConfig);
+ }
+
+ @Override
+ public void start() {
+ started.compareAndSet(false, true);
+ }
+
+ @Override
+ public void shutdown() {
+ started.compareAndSet(true, false);
+ }
+
+ @Override
+ public boolean isStarted() {
+ return started.get();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !started.get();
+ }
+
+ @Override
+ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext
context) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void subscribe(String topic) throws Exception {
+ if (!client.subscribe(topic, isBroadcast, consumerGroup, instanceName,
eventListener)) {
+ throw new PravegaStorageException(String.format("subscribe
topic[%s] fail.", topic));
+ }
+ }
+
+ @Override
+ public void unsubscribe(String topic) {
+ if (!client.unsubscribe(topic, isBroadcast, consumerGroup)) {
+ throw new PravegaStorageException(String.format("unsubscribe
topic[%s] fail.", topic));
+ }
+ }
+
+ @Override
+ public void registerEventListener(EventListener listener) {
+ this.eventListener = listener;
+ }
+
+ public PravegaStorageConfig getClientConfiguration() {
+ return this.pravegaConnectorConfig;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaProducerImpl.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaProducerImpl.java
new file mode 100644
index 000000000..03eef5b4e
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaProducerImpl.java
@@ -0,0 +1,121 @@
+/*
+ * 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.pravega;
+
+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.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.pravega.client.PravegaClient;
+import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
+import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
+
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Config(field = "pravegaConnectorConfig")
+public class PravegaProducerImpl implements Producer {
+
+ private final AtomicBoolean started = new AtomicBoolean(false);
+ private PravegaClient client;
+
+ /**
+ * Unified configuration class corresponding to pravega-storage.properties
+ */
+ private PravegaStorageConfig pravegaConnectorConfig;
+
+ @Override
+ public void init(Properties properties) throws Exception {
+ client = PravegaClient.getInstance(pravegaConnectorConfig);
+ }
+
+ @Override
+ public void start() {
+ started.compareAndSet(false, true);
+ }
+
+ @Override
+ public void shutdown() {
+ started.compareAndSet(true, false);
+ }
+
+ @Override
+ public boolean isStarted() {
+ return started.get();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !started.get();
+ }
+
+ @Override
+ public void publish(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ try {
+ SendResult sendResult = client.publish(cloudEvent.getSubject(),
cloudEvent);
+ sendCallback.onSuccess(sendResult);
+ } catch (Exception e) {
+ log.error("send message error, topic: {}",
cloudEvent.getSubject());
+ OnExceptionContext onExceptionContext =
OnExceptionContext.builder()
+ .messageId("-1")
+ .topic(cloudEvent.getSubject())
+ .exception(new StorageConnectorRuntimeException(e))
+ .build();
+ sendCallback.onException(onExceptionContext);
+ }
+ }
+
+ @Override
+ public void sendOneway(CloudEvent cloudEvent) {
+ client.publish(cloudEvent.getSubject(), cloudEvent);
+ }
+
+ @Override
+ public void request(CloudEvent cloudEvent, RequestReplyCallback
rrCallback, long timeout) throws Exception {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void checkTopicExist(String topic) throws Exception {
+ if (!client.checkTopicExist(topic)) {
+ throw new PravegaStorageException(String.format("topic:%s is not
exist", topic));
+ }
+ }
+
+ @Override
+ public void setExtFields() {
+
+ }
+
+ public PravegaStorageConfig getClientConfiguration() {
+ return this.pravegaConnectorConfig;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaStorageResourceServiceImpl.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaStorageResourceServiceImpl.java
new file mode 100644
index 000000000..ab8225c91
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/PravegaStorageResourceServiceImpl.java
@@ -0,0 +1,37 @@
+/*
+ * 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.pravega;
+
+import org.apache.eventmesh.api.storage.StorageResourceService;
+import org.apache.eventmesh.storage.pravega.client.PravegaClient;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PravegaStorageResourceServiceImpl implements
StorageResourceService {
+
+ @Override
+ public void init() throws Exception {
+ PravegaClient.getInstance().start();
+ }
+
+ @Override
+ public void release() throws Exception {
+ PravegaClient.getInstance().shutdown();
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaClient.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaClient.java
new file mode 100644
index 000000000..7c0cf7829
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaClient.java
@@ -0,0 +1,233 @@
+/*
+ * 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.pravega.client;
+
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
+import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
+
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import io.cloudevents.CloudEvent;
+import io.pravega.client.ClientConfig;
+import io.pravega.client.EventStreamClientFactory;
+import io.pravega.client.admin.ReaderGroupManager;
+import io.pravega.client.admin.StreamManager;
+import io.pravega.client.stream.EventStreamReader;
+import io.pravega.client.stream.EventStreamWriter;
+import io.pravega.client.stream.EventWriterConfig;
+import io.pravega.client.stream.ReaderConfig;
+import io.pravega.client.stream.ReaderGroupConfig;
+import io.pravega.client.stream.StreamConfiguration;
+import io.pravega.client.stream.impl.ByteArraySerializer;
+import io.pravega.shared.NameUtils;
+import io.pravega.shared.security.auth.DefaultCredentials;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PravegaClient {
+
+ private final PravegaStorageConfig config;
+ private final StreamManager streamManager;
+ private final EventStreamClientFactory clientFactory;
+ private final ReaderGroupManager readerGroupManager;
+ private final Map<String, EventStreamWriter<byte[]>> writerMap = new
ConcurrentHashMap<>();
+ private final Map<String, SubscribeTask> subscribeTaskMap = new
ConcurrentHashMap<>();
+
+ private static PravegaClient instance;
+
+ public static PravegaClient getInstance() {
+ return instance;
+ }
+
+ public static PravegaClient getInstance(PravegaStorageConfig config) {
+ if (instance == null) {
+ instance = new PravegaClient(config);
+ }
+
+ return instance;
+ }
+
+ private PravegaClient(PravegaStorageConfig config) {
+ this.config = config;
+ streamManager = StreamManager.create(config.getControllerURI());
+ ClientConfig.ClientConfigBuilder clientConfigBuilder =
ClientConfig.builder().controllerURI(config.getControllerURI());
+ if (config.isAuthEnabled()) {
+ clientConfigBuilder.credentials(new
DefaultCredentials(config.getPassword(), config.getUsername()));
+ }
+ if (config.isTlsEnable()) {
+
clientConfigBuilder.trustStore(config.getTruststore()).validateHostName(false);
+ }
+ ClientConfig clientConfig = clientConfigBuilder.build();
+ clientFactory = EventStreamClientFactory.withScope(config.getScope(),
clientConfig);
+ readerGroupManager = ReaderGroupManager.withScope(config.getScope(),
clientConfig);
+ }
+
+ protected static PravegaClient getNewInstance(PravegaStorageConfig config)
{
+ return new PravegaClient(config);
+ }
+
+ public void start() {
+ if (createScope()) {
+ log.info("Create Pravega scope[{}] success.", config.getScope());
+ } else {
+ log.info("Pravega scope[{}] has already been created.",
config.getScope());
+ }
+ }
+
+ public void shutdown() {
+ subscribeTaskMap.forEach((topic, task) -> task.stopRead());
+ subscribeTaskMap.clear();
+ writerMap.forEach((topic, writer) -> writer.close());
+ writerMap.clear();
+ readerGroupManager.close();
+ clientFactory.close();
+ streamManager.close();
+ }
+
+ /**
+ * Publish CloudEvent to Pravega stream named topic. Note that the
messageId in SendResult is always -1 since {@link
+ * EventStreamWriter#writeEvent(Object)} just return {@link
java.util.concurrent.CompletableFuture} with {@link Void} which couldn't get
+ * messageId.
+ *
+ * @param topic topic
+ * @param cloudEvent cloudEvent
+ * @return SendResult whose messageId is always -1
+ */
+ public SendResult publish(String topic, CloudEvent cloudEvent) {
+ if (!createStream(topic)) {
+ log.debug("stream[{}] has already been created.", topic);
+ }
+ EventStreamWriter<byte[]> writer = writerMap.computeIfAbsent(topic, k
-> createWrite(topic));
+ PravegaCloudEventWriter cloudEventWriter = new
PravegaCloudEventWriter(topic);
+ PravegaEvent pravegaEvent = cloudEventWriter.writeBinary(cloudEvent);
+ try {
+ writer.writeEvent(PravegaEvent.toByteArray(pravegaEvent)).get(5,
TimeUnit.SECONDS);
+ } catch (ExecutionException | InterruptedException | TimeoutException
e) {
+ log.error(String.format("Write topic[%s] fail.", topic), e);
+ throw new PravegaStorageException(String.format("Write topic[%s]
fail.", topic));
+ }
+ SendResult sendResult = new SendResult();
+ sendResult.setTopic(topic);
+ // set -1 as messageId since writeEvent method doesn't return it.
+ sendResult.setMessageId("-1");
+ return sendResult;
+
+ }
+
+ public boolean subscribe(String topic, boolean isBroadcast, String
consumerGroup, String instanceName, EventListener listener) {
+ if (subscribeTaskMap.containsKey(topic)) {
+ return true;
+ }
+ String readerGroupName = buildReaderGroupName(isBroadcast,
consumerGroup, topic);
+ createReaderGroup(topic, readerGroupName);
+ String readerId = buildReaderId(instanceName);
+ EventStreamReader<byte[]> reader = createReader(readerId,
readerGroupName);
+ SubscribeTask subscribeTask = new SubscribeTask(topic, reader,
listener);
+ subscribeTask.start();
+ subscribeTaskMap.put(topic, subscribeTask);
+ return true;
+ }
+
+ public boolean unsubscribe(String topic, boolean isBroadcast, String
consumerGroup) {
+ if (!subscribeTaskMap.containsKey(topic)) {
+ return true;
+ }
+ if (!isBroadcast) {
+ deleteReaderGroup(buildReaderGroupName(false, consumerGroup,
topic));
+ }
+ subscribeTaskMap.remove(topic).stopRead();
+ return true;
+ }
+
+ public boolean checkTopicExist(String topic) {
+ return streamManager.checkStreamExists(config.getScope(), topic);
+ }
+
+ private boolean createScope() {
+ return streamManager.createScope(config.getScope());
+ }
+
+ private boolean createStream(String topic) {
+ StreamConfiguration streamConfiguration =
StreamConfiguration.builder().build();
+ return streamManager.createStream(config.getScope(), topic,
streamConfiguration);
+ }
+
+ private EventStreamWriter<byte[]> createWrite(String topic) {
+ return clientFactory.createEventWriter(topic, new
ByteArraySerializer(), EventWriterConfig.builder().build());
+ }
+
+ private String buildReaderGroupName(boolean isBroadcast, String
consumerGroup, String topic) {
+ if (isBroadcast) {
+ return UUID.randomUUID().toString();
+ } else {
+ return String.format("%s-%s", consumerGroup, topic);
+ }
+ }
+
+ private String buildReaderId(String instanceName) {
+ return String.format("%s-reader", instanceName).replaceAll("\\(",
"-").replaceAll("\\)", "-");
+ }
+
+ private void createReaderGroup(String topic, String readerGroupName) {
+ if (!checkTopicExist(topic)) {
+ createStream(topic);
+ }
+ ReaderGroupConfig readerGroupConfig =
+ ReaderGroupConfig.builder()
+ .stream(NameUtils.getScopedStreamName(config.getScope(),
topic))
+
.retentionType(ReaderGroupConfig.StreamDataRetention.AUTOMATIC_RELEASE_AT_LAST_CHECKPOINT)
+ .build();
+ readerGroupManager.createReaderGroup(readerGroupName,
readerGroupConfig);
+ }
+
+ private void deleteReaderGroup(String readerGroup) {
+ readerGroupManager.deleteReaderGroup(readerGroup);
+ }
+
+ private EventStreamReader<byte[]> createReader(String readerId, String
readerGroup) {
+ return clientFactory.createReader(readerId, readerGroup, new
ByteArraySerializer(), ReaderConfig.builder().build());
+ }
+
+ protected StreamManager getStreamManager() {
+ return streamManager;
+ }
+
+ protected EventStreamClientFactory getClientFactory() {
+ return clientFactory;
+ }
+
+ protected ReaderGroupManager getReaderGroupManager() {
+ return readerGroupManager;
+ }
+
+ public Map<String, EventStreamWriter<byte[]>> getWriterMap() {
+ return writerMap;
+ }
+
+ protected Map<String, SubscribeTask> getSubscribeTaskMap() {
+ return subscribeTaskMap;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaCloudEventWriter.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaCloudEventWriter.java
new file mode 100644
index 000000000..ebc2c70e3
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaCloudEventWriter.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.pravega.client;
+
+import java.nio.charset.StandardCharsets;
+
+import io.cloudevents.CloudEventData;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.rw.CloudEventContextWriter;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventWriter;
+
+public class PravegaCloudEventWriter
+ implements MessageWriter<CloudEventWriter<PravegaEvent>, PravegaEvent>,
CloudEventWriter<PravegaEvent> {
+
+ private final PravegaEvent pravegaEvent;
+
+ public PravegaCloudEventWriter(String topic) {
+ pravegaEvent = new PravegaEvent();
+ pravegaEvent.setTopic(topic);
+ pravegaEvent.setCreateTimestamp(System.currentTimeMillis());
+ }
+
+ @Override
+ public PravegaEvent setEvent(EventFormat format, byte[] value) throws
CloudEventRWException {
+ pravegaEvent.setData(new String(value, StandardCharsets.UTF_8));
+ return pravegaEvent;
+ }
+
+ @Override
+ public PravegaEvent end(CloudEventData data) throws CloudEventRWException {
+ pravegaEvent.setData(new String(data.toBytes(),
StandardCharsets.UTF_8));
+ return pravegaEvent;
+ }
+
+ @Override
+ public PravegaEvent end() throws CloudEventRWException {
+ pravegaEvent.setData("");
+ return pravegaEvent;
+ }
+
+ @Override
+ public CloudEventContextWriter withContextAttribute(String name, String
value) throws CloudEventRWException {
+ pravegaEvent.getExtensions().put(name, value);
+ return this;
+ }
+
+ @Override
+ public CloudEventWriter<PravegaEvent> create(SpecVersion version) throws
CloudEventRWException {
+ pravegaEvent.setVersion(version);
+ return this;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaEvent.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaEvent.java
new file mode 100644
index 000000000..4ae85de92
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/PravegaEvent.java
@@ -0,0 +1,77 @@
+/*
+ * 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.pravega.client;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.storage.pravega.exception.PravegaStorageException;
+
+import java.io.Serializable;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class PravegaEvent implements Serializable {
+
+ private static final long serialVersionUID = 0L;
+
+ private SpecVersion version;
+ private String topic;
+ private String data;
+ private Map<String, String> extensions = new HashMap<>();
+ private long createTimestamp;
+
+ public static byte[] toByteArray(PravegaEvent pravegaEvent) {
+ return
JsonUtils.toJSONString(pravegaEvent).getBytes(StandardCharsets.UTF_8);
+ }
+
+ public static PravegaEvent getFromByteArray(byte[] body) {
+ return JsonUtils.parseObject(new String(body, StandardCharsets.UTF_8),
PravegaEvent.class);
+ }
+
+ public CloudEvent convertToCloudEvent() {
+ CloudEventBuilder builder;
+ switch (version) {
+ case V03:
+ builder = CloudEventBuilder.v03();
+ break;
+ case V1:
+ builder = CloudEventBuilder.v1();
+ break;
+ default:
+ throw new PravegaStorageException(String.format("CloudEvent
version %s does not support.", version));
+ }
+ builder.withData(data.getBytes(StandardCharsets.UTF_8))
+ .withId(extensions.remove("id"))
+ .withSource(URI.create(extensions.remove("source")))
+ .withType(extensions.remove("type"))
+ .withDataContentType(extensions.remove("datacontenttype"))
+ .withSubject(extensions.remove("subject"));
+ extensions.forEach(builder::withExtension);
+ return builder.build();
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/SubscribeTask.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/SubscribeTask.java
new file mode 100644
index 000000000..34ea6661b
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/client/SubscribeTask.java
@@ -0,0 +1,89 @@
+/*
+ * 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.pravega.client;
+
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+import io.pravega.client.stream.EventRead;
+import io.pravega.client.stream.EventStreamReader;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SubscribeTask extends Thread {
+
+ private final EventStreamReader<byte[]> reader;
+ private final EventListener listener;
+ private final AtomicBoolean running = new AtomicBoolean(true);
+ private final AtomicBoolean continueRead = new AtomicBoolean(true);
+
+ public SubscribeTask(String name, EventStreamReader<byte[]> reader,
EventListener listener) {
+ super(name);
+ this.reader = reader;
+ this.listener = listener;
+ }
+
+ @Override
+ public void run() {
+ CloudEvent cloudEvent = null;
+ while (running.get()) {
+ if (continueRead.get()) {
+ EventRead<byte[]> event = reader.readNextEvent(2000);
+ if (event == null) {
+ continue;
+ }
+ byte[] eventByteArray = event.getEvent();
+ if (eventByteArray == null) {
+ continue;
+ }
+ PravegaEvent pravegaEvent =
PravegaEvent.getFromByteArray(eventByteArray);
+ cloudEvent = pravegaEvent.convertToCloudEvent();
+
+ listener.consume(cloudEvent, new
PravegaEventMeshAsyncConsumeContext());
+ } else {
+ listener.consume(cloudEvent, new
PravegaEventMeshAsyncConsumeContext());
+ }
+ }
+ }
+
+ public void stopRead() {
+ running.compareAndSet(true, false);
+ }
+
+ private class PravegaEventMeshAsyncConsumeContext extends
EventMeshAsyncConsumeContext {
+
+ @Override
+ public void commit(EventMeshAction action) {
+ switch (action) {
+ case CommitMessage:
+ case ReconsumeLater:
+ continueRead.set(false);
+ break;
+ case ManualAck:
+ continueRead.set(true);
+ break;
+ default:
+ }
+ }
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfig.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfig.java
new file mode 100644
index 000000000..646d021f0
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfig.java
@@ -0,0 +1,67 @@
+/*
+ * 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.pravega.config;
+
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigFiled;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.net.URI;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@Config(prefix = "eventMesh.server.pravega", path =
"classPath://pravega-storage.properties")
+public class PravegaStorageConfig {
+
+ @ConfigFiled(field = "controller.uri")
+ private URI controllerURI = URI.create("tcp://127.0.0.1:9090");
+
+ @ConfigFiled(field = "scope")
+ private String scope = "eventmesh-pravega";
+
+ @ConfigFiled(field = "clientPool.size")
+ private int clientPoolSize = 8;
+
+ @ConfigFiled(field = "queue.size")
+ private int queueSize = 512;
+
+ @ConfigFiled(field = "authEnabled", reload = true)
+ private boolean authEnabled = false;
+
+ @ConfigFiled(field = "username")
+ private String username = "";
+
+ @ConfigFiled(field = "password")
+ private String password = "";
+
+ @ConfigFiled(field = "tlsEnabled")
+ private boolean tlsEnable = false;
+
+ @ConfigFiled(field = "truststore")
+ private String truststore = "";
+
+ public void reload() {
+ if (!authEnabled && StringUtils.isNotBlank(username) &&
StringUtils.isNotBlank(password)) {
+ authEnabled = true;
+ }
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/exception/PravegaStorageException.java
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/exception/PravegaStorageException.java
new file mode 100644
index 000000000..85cfa710e
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/java/org/apache/eventmesh/storage/pravega/exception/PravegaStorageException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.pravega.exception;
+
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+
+public class PravegaStorageException extends StorageConnectorRuntimeException {
+
+ public PravegaStorageException(String message) {
+ super(message);
+ }
+
+ public PravegaStorageException(Throwable throwable) {
+ super(throwable);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
b/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
new file mode 100644
index 000000000..2d68877d3
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/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.
+
+pravega=org.apache.eventmesh.storage.pravega.PravegaConsumerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
b/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
new file mode 100644
index 000000000..b35a24cd4
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/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.
+
+pravega=org.apache.eventmesh.storage.pravega.PravegaProducerImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
b/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
new file mode 100644
index 000000000..64222b4dc
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/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.
+
+pravega=org.apache.eventmesh.storage.pravega.PravegaStorageResourceServiceImpl
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/pravega-storage.properties
b/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/pravega-storage.properties
new file mode 100644
index 000000000..25fa6f0af
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/main/resources/pravega-storage.properties
@@ -0,0 +1,26 @@
+#
+# 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.
+#
+#######################pravega-client##################
+eventMesh.server.pravega.controller.uri=tcp://127.0.0.1:9090
+eventMesh.server.pravega.scope=eventmesh-pravega
+eventMesh.server.pravega.authEnabled=false
+eventMesh.server.pravega.username=
+eventMesh.server.pravega.password=
+eventMesh.server.pravega.tlsEnabled=false
+eventMesh.server.pravega.truststore=
+eventMesh.server.pravega.clientPool.size=8
+eventMesh.server.pravega.queue.size=512
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/client/PravegaClientTest.java
b/eventmesh-storage/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/client/PravegaClientTest.java
new file mode 100644
index 000000000..594d1e0fd
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/client/PravegaClientTest.java
@@ -0,0 +1,188 @@
+/*
+ * 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.pravega.client;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.apache.eventmesh.api.AsyncConsumeContext;
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.storage.pravega.config.PravegaStorageConfig;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import io.pravega.client.admin.StreamManager;
+import io.pravega.client.stream.ReaderGroupNotFoundException;
+
+import com.github.dockerjava.api.command.CreateContainerCmd;
+import com.github.dockerjava.api.model.PortBinding;
+
+@Ignore
+public class PravegaClientTest {
+
+ private StreamManager streamManager;
+ private PravegaStorageConfig config;
+ private URI controllerURI;
+
+ @Rule
+ public GenericContainer pravega =
+ new GenericContainer(DockerImageName.parse("pravega/pravega:0.11.0"))
+ .withExposedPorts(9090, 12345).withEnv("HOST_IP",
"127.0.0.1").withCommand("standalone")
+ .withCreateContainerCmdModifier((Consumer<CreateContainerCmd>)
createContainerCmd ->
+ createContainerCmd.getHostConfig()
+ .withPortBindings(PortBinding.parse("9090:9090"),
PortBinding.parse("12345:12345")));
+
+ @Before
+ public void setUp() {
+ String host = pravega.getHost();
+ int port = pravega.getFirstMappedPort();
+ controllerURI = URI.create(String.format("tcp://%s:%d", host, port));
+ streamManager = StreamManager.create(controllerURI);
+ }
+
+ @Test
+ public void startTest() {
+ PravegaClient pravegaClient = getNewPravegaClient();
+ pravegaClient.start();
+ assertTrue(streamManager.checkScopeExists(config.getScope()));
+ }
+
+ @Test
+ public void shutdownTest() {
+ PravegaClient pravegaClient = getNewPravegaClient();
+ pravegaClient.start();
+ pravegaClient.publish("test1", createCloudEvent());
+ pravegaClient.publish("test2", createCloudEvent());
+ pravegaClient.publish("test3", createCloudEvent());
+
+ pravegaClient.shutdown();
+ assertTrue(pravegaClient.getSubscribeTaskMap().isEmpty());
+ assertTrue(pravegaClient.getWriterMap().isEmpty());
+ }
+
+ @Test
+ public void publishTest() {
+ PravegaClient pravegaClient = getNewPravegaClient();
+ pravegaClient.start();
+ pravegaClient.publish("test1", createCloudEvent());
+ pravegaClient.publish("test2", createCloudEvent());
+ assertTrue(streamManager.checkStreamExists(config.getScope(),
"test1"));
+ assertTrue(streamManager.checkStreamExists(config.getScope(),
"test2"));
+ }
+
+ @Test
+ public void subscribeTest() {
+ PravegaClient pravegaClient = getNewPravegaClient();
+ pravegaClient.start();
+ pravegaClient.subscribe("test1", false, "consumerGroup",
"instanceName", new EventListener() {
+ @Override
+ public void consume(CloudEvent cloudEvent, AsyncConsumeContext
context) {
+ // do nothing
+ }
+ });
+
assertNotNull(pravegaClient.getReaderGroupManager().getReaderGroup("test1-consumerGroup"));
+ assertTrue(pravegaClient.getSubscribeTaskMap().containsKey("test1"));
+ }
+
+ @Test
+ public void unsubscribeTest() {
+ PravegaClient pravegaClient = getNewPravegaClient();
+ pravegaClient.unsubscribe("test1", false, "consumerGroup");
+
+ pravegaClient.start();
+ pravegaClient.subscribe("test1", false, "consumerGroup",
"instanceName", new EventListener() {
+ @Override
+ public void consume(CloudEvent cloudEvent, AsyncConsumeContext
context) {
+ // do nothing
+ }
+ });
+ pravegaClient.unsubscribe("test1", false, "consumerGroup");
+
+ assertFalse(pravegaClient.getSubscribeTaskMap().containsKey("test1"));
+ try {
+
pravegaClient.getReaderGroupManager().getReaderGroup("test1-consumerGroup");
+ } catch (Exception e) {
+ assertTrue(e instanceof ReaderGroupNotFoundException);
+ return;
+ }
+ fail();
+ }
+
+ @Test
+ public void checkTopicExistTest() {
+ PravegaClient pravegaClient = getNewPravegaClient();
+ pravegaClient.start();
+ assertFalse(pravegaClient.checkTopicExist("test1"));
+
+ pravegaClient.publish("test1", createCloudEvent());
+ assertTrue(pravegaClient.checkTopicExist("test1"));
+ }
+
+ public PravegaClient getNewPravegaClient() {
+ ConfigService configService = ConfigService.getInstance();
+
+ this.config =
configService.buildConfigInstance(PravegaStorageConfig.class);
+ this.config.setControllerURI(controllerURI);
+ return PravegaClient.getNewInstance(this.config);
+ }
+
+ private CloudEvent createCloudEvent() {
+ String data =
"{\"headers\":{\"content-length\":\"36\",\"Accept\":\"*/*\",\"ip\":\"127.0.0.1:51226\",\"User-Agent\":\"curl/7.83.1\","
+ +
"\"Host\":\"127.0.0.1:10105\",\"source\":\"127.0.0.1:51226\",\"Content-Type\":\"application/json\"},"
+ +
"\"path\":\"/eventmesh/publish/TEST-TOPIC-HTTP-ASYNC\",\"method\":\"POST\",\"body\":{\"pass\":\"12345678\",\"name\":\"admin\"}}";
+ return CloudEventBuilder.v1()
+ .withId("c61039e1-7884-4d7f-b72f-6c61160a64fc")
+ .withSource(URI.create("source:127.0.0.1:51226"))
+ .withType("http_request")
+ .withDataContentType("application/json")
+ .withSubject("TEST-TOPIC-HTTP-ASYNC")
+ .withData(data.getBytes(StandardCharsets.UTF_8))
+ .withExtension("reqeventmesh2mqtimestamp", "1659342713460")
+ .withExtension("ip", "127.0.0.1:51226")
+ .withExtension("idc", "idc")
+ .withExtension("protocoldesc", "http")
+ .withExtension("pid", 2376)
+ .withExtension("env", "env")
+ .withExtension("sys", 1234)
+ .withExtension("ttl", 4000)
+ .withExtension("producergroup", "em-http-producer")
+ .withExtension("consumergroup", "em-http-consumer")
+ .withExtension("passwd", "pass")
+ .withExtension("bizseqno", "249695004068274968410952665702")
+ .withExtension("protocoltype", "http")
+ .withExtension("msgtype", "persistent")
+ .withExtension("uniqueid", "866012286651006371403062105469")
+ .withExtension("username", "eventmesh")
+ .withExtension("reqc2eventmeshtimestamp", "1659342713460")
+ .build();
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfigTest.java
b/eventmesh-storage/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfigTest.java
new file mode 100644
index 000000000..18fb7363f
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/test/java/org/apache/eventmesh/storage/pravega/config/PravegaStorageConfigTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.pravega.config;
+
+import org.apache.eventmesh.api.factory.StoragePluginFactory;
+import org.apache.eventmesh.storage.pravega.PravegaConsumerImpl;
+import org.apache.eventmesh.storage.pravega.PravegaProducerImpl;
+
+import java.net.URI;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PravegaStorageConfigTest {
+
+ @Test
+ public void getConfigWhenPravegaConsumerInit() {
+ PravegaConsumerImpl consumer =
+ (PravegaConsumerImpl)
StoragePluginFactory.getMeshMQPushConsumer("pravega");
+
+ PravegaStorageConfig config = consumer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ @Test
+ public void getConfigWhenPravegaProducerInit() {
+ PravegaProducerImpl producer =
+ (PravegaProducerImpl)
StoragePluginFactory.getMeshMQProducer("pravega");
+
+ PravegaStorageConfig config = producer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ private void assertConfig(PravegaStorageConfig config) {
+ Assert.assertEquals(config.getControllerURI(),
URI.create("tcp://127.0.0.1:816"));
+ Assert.assertEquals(config.getScope(), "scope-success!!!");
+ Assert.assertTrue(config.isAuthEnabled());
+ Assert.assertEquals(config.getUsername(), "username-success!!!");
+ Assert.assertEquals(config.getPassword(), "password-success!!!");
+ Assert.assertTrue(config.isTlsEnable());
+ Assert.assertEquals(config.getTruststore(), "truststore-success!!!");
+ Assert.assertEquals(config.getClientPoolSize(), 816);
+ Assert.assertEquals(config.getQueueSize(), 1816);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-pravega/src/test/resources/pravega-storage.properties
b/eventmesh-storage/eventmesh-storage-pravega/src/test/resources/pravega-storage.properties
new file mode 100644
index 000000000..5bb5e5490
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-pravega/src/test/resources/pravega-storage.properties
@@ -0,0 +1,26 @@
+#
+# 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.
+#
+#######################pravega-client##################
+eventMesh.server.pravega.controller.uri=tcp://127.0.0.1:816
+eventMesh.server.pravega.scope=scope-success!!!
+eventMesh.server.pravega.authEnabled=false
+eventMesh.server.pravega.username=username-success!!!
+eventMesh.server.pravega.password=password-success!!!
+eventMesh.server.pravega.tlsEnabled=true
+eventMesh.server.pravega.truststore=truststore-success!!!
+eventMesh.server.pravega.clientPool.size=816
+eventMesh.server.pravega.queue.size=1816
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index 491c6a47e..fbce4a682 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -26,6 +26,7 @@ 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-pravega'
include 'eventmesh-storage:eventmesh-storage-redis'
include 'eventmesh-storage:eventmesh-storage-rocketmq'
include 'eventmesh-storage:eventmesh-storage-rabbitmq'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
