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/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 6dfd46d3e [ISSUE #4396] Realize the admin function for Kafka
storage-plugin (#4438)
6dfd46d3e is described below
commit 6dfd46d3ecd3a39fdb0929017ce9cec54eebf9c3
Author: Pil0tXia <[email protected]>
AuthorDate: Tue Sep 12 14:58:37 2023 +0800
[ISSUE #4396] Realize the admin function for Kafka storage-plugin (#4438)
* config: kafka-storage-plugin, RESTORE BEFORE MERGE
* fix: meshMqAdmin not loaded at startup
* feat: implement getTopic
* feat: get all history message count
* feat: implement createTopic
* optimize: remove redundant bootstrap.server config and add new configs
* feat: implement deleteTopic and handle exceptions
* Revert "config: kafka-storage-plugin, RESTORE BEFORE MERGE"
This reverts commit 20dce1af753c13cf447d33690d84231e624bac94.
* chore: move kafka comment to a more suitable place.
* optimize: support topic batch listing
---
.../eventmesh/storage/kafka/admin/KafkaAdmin.java | 150 +++++++++++++++++++++
.../storage/kafka/admin/KafkaAdminAdaptor.java | 85 ++++++++++++
.../storage/kafka/common/EventMeshConstants.java | 2 +
.../storage/kafka/config/ClientConfiguration.java | 8 ++
.../eventmesh/org.apache.eventmesh.api.admin.Admin | 16 +++
.../src/main/resources/kafka-client.properties | 20 ++-
6 files changed, 274 insertions(+), 7 deletions(-)
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/admin/KafkaAdmin.java
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/admin/KafkaAdmin.java
new file mode 100644
index 000000000..f86bfee87
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/admin/KafkaAdmin.java
@@ -0,0 +1,150 @@
+/*
+ * 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.admin;
+
+import static
org.apache.eventmesh.storage.kafka.common.EventMeshConstants.DEFAULT_TIMEOUT_IN_SECONDS;
+
+import org.apache.eventmesh.api.admin.AbstractAdmin;
+import org.apache.eventmesh.api.admin.TopicProperties;
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.storage.kafka.config.ClientConfiguration;
+
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.OffsetSpec;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class KafkaAdmin extends AbstractAdmin {
+
+ // properties for kafka admin client
+ private static final Properties kafkaProps = new Properties();
+
+ // properties for topic management
+ private static final Map<String, Integer> topicProps = new HashMap<>();
+
+ public KafkaAdmin() {
+ super(new AtomicBoolean(false));
+
+ ConfigService configService = ConfigService.getInstance();
+ ClientConfiguration clientConfiguration =
configService.buildConfigInstance(ClientConfiguration.class);
+
+ kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
clientConfiguration.getNamesrvAddr());
+ topicProps.put("partitionNum", clientConfiguration.getPartitions());
+ topicProps.put("replicationFactorNum", (int)
clientConfiguration.getReplicationFactors());
+ }
+
+ @Override
+ public List<TopicProperties> getTopic() {
+ try (Admin client = Admin.create(kafkaProps)) {
+ Set<String> topicList =
client.listTopics().names().get(DEFAULT_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
+ Map<String, KafkaFuture<TopicDescription>> topicDescriptionFutures
= client.describeTopics(topicList).values();
+ List<TopicProperties> result = new ArrayList<>();
+ for (Entry<String, KafkaFuture<TopicDescription>> entry :
topicDescriptionFutures.entrySet()) {
+ String topicName = entry.getKey();
+ TopicDescription topicDescription =
entry.getValue().get(DEFAULT_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
+
+ long messageCount = topicDescription.partitions().stream()
+ .mapToInt(TopicPartitionInfo::partition)
+ .mapToLong(partition -> {
+ try {
+ return getMsgCount(topicName, partition, client);
+ } catch (TimeoutException e) {
+ log.error("Failed to get msg offset when listing
topics. Kafka response timed out in {} seconds.",
+ DEFAULT_TIMEOUT_IN_SECONDS);
+ throw new RuntimeException(e);
+ } catch (ExecutionException | InterruptedException e) {
+ log.error("Failed to get msg offset when listing
topics.", e);
+ throw new RuntimeException(e);
+ }
+ })
+ .sum();
+
+ result.add(new TopicProperties(topicName, messageCount));
+ }
+ result.sort(Comparator.comparing(t -> t.name));
+ return result;
+ } catch (TimeoutException e) {
+ log.error("Failed to list topics. Kafka response timed out in {}
seconds.", DEFAULT_TIMEOUT_IN_SECONDS);
+ throw new RuntimeException(e);
+ } catch (Exception e) {
+ log.error("Failed to list topics.", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private long getMsgCount(String topicName, int partition, Admin client)
throws ExecutionException, InterruptedException, TimeoutException {
+ TopicPartition topicPartition = new TopicPartition(topicName,
partition);
+ long earliestOffset = getOffset(topicPartition, OffsetSpec.earliest(),
client);
+ long latestOffset = getOffset(topicPartition, OffsetSpec.latest(),
client);
+ return latestOffset - earliestOffset;
+ }
+
+ private long getOffset(TopicPartition topicPartition, OffsetSpec
offsetSpec, Admin client)
+ throws ExecutionException, InterruptedException, TimeoutException {
+ Map<TopicPartition, OffsetSpec> offsetSpecMap =
Collections.singletonMap(topicPartition, offsetSpec);
+ Map<TopicPartition, ListOffsetsResultInfo> offsetResultMap =
+
client.listOffsets(offsetSpecMap).all().get(DEFAULT_TIMEOUT_IN_SECONDS,
TimeUnit.SECONDS);
+ return offsetResultMap.get(topicPartition).offset();
+ }
+
+ @Override
+ public void createTopic(String topicName) {
+ try (Admin client = Admin.create(kafkaProps)) {
+ NewTopic newTopic = new NewTopic(topicName,
topicProps.get("partitionNum"),
topicProps.get("replicationFactorNum").shortValue());
+
client.createTopics(Collections.singletonList(newTopic)).all().get(DEFAULT_TIMEOUT_IN_SECONDS,
TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ log.error("Failed to create topic. Kafka response timed out in {}
seconds.", DEFAULT_TIMEOUT_IN_SECONDS);
+ } catch (Exception e) {
+ log.error("Failed to create topic.", e);
+ }
+ }
+
+ @Override
+ public void deleteTopic(String topicName) {
+ try (Admin client = Admin.create(kafkaProps)) {
+
client.deleteTopics(Collections.singletonList(topicName)).all().get(DEFAULT_TIMEOUT_IN_SECONDS,
TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ log.error("Failed to delete topic. Kafka response timed out in {}
seconds.", DEFAULT_TIMEOUT_IN_SECONDS);
+ } catch (Exception e) {
+ log.error("Failed to delete topic.", e);
+ }
+ }
+
+}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/admin/KafkaAdminAdaptor.java
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/admin/KafkaAdminAdaptor.java
new file mode 100644
index 000000000..042756d45
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/admin/KafkaAdminAdaptor.java
@@ -0,0 +1,85 @@
+/*
+ * 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.admin;
+
+import org.apache.eventmesh.api.admin.Admin;
+import org.apache.eventmesh.api.admin.TopicProperties;
+
+import java.util.List;
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+
+public class KafkaAdminAdaptor implements Admin {
+
+ private final KafkaAdmin admin;
+
+ public KafkaAdminAdaptor() {
+ admin = new KafkaAdmin();
+ }
+
+ @Override
+ public boolean isStarted() {
+ return admin.isStarted();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return admin.isClosed();
+ }
+
+ @Override
+ public void start() {
+ admin.start();
+ }
+
+ @Override
+ public void shutdown() {
+ admin.shutdown();
+ }
+
+ @Override
+ public void init(Properties properties) throws Exception {
+ admin.init(properties);
+ }
+
+ @Override
+ public List<TopicProperties> getTopic() throws Exception {
+ return admin.getTopic();
+ }
+
+ @Override
+ public void createTopic(String topicName) throws Exception {
+ admin.createTopic(topicName);
+ }
+
+ @Override
+ public void deleteTopic(String topicName) throws Exception {
+ admin.deleteTopic(topicName);
+ }
+
+ @Override
+ public List<CloudEvent> getEvent(String topicName, int offset, int length)
throws Exception {
+ return admin.getEvent(topicName, offset, length);
+ }
+
+ @Override
+ public void publish(CloudEvent cloudEvent) throws Exception {
+ admin.publish(cloudEvent);
+ }
+}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
index 2bc2f3c04..4b54846ce 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/common/EventMeshConstants.java
@@ -23,5 +23,7 @@ public class EventMeshConstants {
public static final int DEFAULT_TIMEOUT_IN_MILLISECONDS = 3000;
+ public static final int DEFAULT_TIMEOUT_IN_SECONDS = 10;
+
public static final String STORE_TIMESTAMP = "storetime";
}
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
index 3749e8157..2e92877ec 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/java/org/apache/eventmesh/storage/kafka/config/ClientConfiguration.java
@@ -44,6 +44,14 @@ public class ClientConfiguration {
@Builder.Default
private String clientPass = "password";
+ @ConfigFiled(field = "num.partitions")
+ @Builder.Default
+ private int partitions = 1;
+
+ @ConfigFiled(field = "num.replicationFactors")
+ @Builder.Default
+ private short replicationFactors = 1;
+
@ConfigFiled(field = "client.consumeThreadMin")
@Builder.Default
private Integer consumeThreadMin = 2;
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
new file mode 100644
index 000000000..9f8cb52ae
--- /dev/null
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.admin.Admin
@@ -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.admin.KafkaAdminAdaptor
\ No newline at end of file
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
index 31923d73d..cbe0a036f 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
+++
b/eventmesh-storage-plugin/eventmesh-storage-kafka/src/main/resources/kafka-client.properties
@@ -14,13 +14,23 @@
# 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
+
+# Configurations started with 'eventMesh.server.kafka' are required for
EventMesh.
+# You may see kafka.server.KafkaConfig for additional details and defaults of
the rest of the configurations.
+
+############################# EventMesh Kafka Client
#############################
+
+eventMesh.server.kafka.namesrvAddr=localhost:9092;localhost:9092
eventMesh.server.kafka.cluster=DefaultCluster
eventMesh.server.kafka.accessKey=********
eventMesh.server.kafka.secretKey=********
-# see kafka.server.KafkaConfig for additional details and defaults
+############################# EventMesh Log Basics
#############################
+
+# the number of partitions per topic
+eventMesh.server.kafka.num.partitions=1
+# the number of replication factors per topic
+eventMesh.server.kafka.num.replicationFactors=1
############################# Server Basics #############################
@@ -29,10 +39,6 @@ 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
