You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by da...@apache.org on 2024/07/30 09:13:05 UTC
(pulsar) 01/01: Introduce gRPC SchemaStorage
This is an automated email from the ASF dual-hosted git repository.
daojun pushed a commit to branch dev/grpc_schema_storage
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit e1d49c1245e164690dfd48a14ff1159ade12d439
Author: DaoJun <da...@apache.org>
AuthorDate: Tue Jul 30 17:12:44 2024 +0800
Introduce gRPC SchemaStorage
---
.../apache/pulsar/broker/ServiceConfiguration.java | 4 +
pulsar-broker/pom.xml | 26 ++-
.../service/schema/grpc/GrpcSchemaStorage.java | 145 ++++++++++++++++
.../schema/grpc/GrpcSchemaStorageFactory.java | 33 ++++
.../broker/service/schema/grpc/package-info.java | 19 +++
.../service/schema/GrpcSchemaStorageTest.java | 185 +++++++++++++++++++++
pulsar-common/pom.xml | 64 +++++++
.../src/main/proto/GrpcSchemaService.proto | 64 +++++++
8 files changed, 535 insertions(+), 5 deletions(-)
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index 2d2765287c0..7f7424d1b5d 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -2989,6 +2989,10 @@ public class ServiceConfiguration implements PulsarConfiguration {
private String schemaRegistryStorageClassName = "org.apache.pulsar.broker.service.schema"
+ ".BookkeeperSchemaStorageFactory";
+ private String schemaRegistryStorageGrpcEndpoint;
+
+ private boolean schemaRegistryStorageGrpcEnableTls = false;
+
@FieldContext(
category = CATEGORY_SCHEMA,
doc = "The list compatibility checkers to be used in schema registry"
diff --git a/pulsar-broker/pom.xml b/pulsar-broker/pom.xml
index 20117ed21db..6f4675feffb 100644
--- a/pulsar-broker/pom.xml
+++ b/pulsar-broker/pom.xml
@@ -33,6 +33,27 @@
<name>Pulsar Broker</name>
<dependencies>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-netty-shaded</artifactId>
+ <version>${grpc.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-protobuf</artifactId>
+ <version>${grpc.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-stub</artifactId>
+ <version>${grpc.version}</version>
+ </dependency>
+ <dependency> <!-- necessary for Java 9+ -->
+ <groupId>org.apache.tomcat</groupId>
+ <artifactId>annotations-api</artifactId>
+ <version>6.0.53</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
@@ -491,11 +512,6 @@
<classifier>shaded</classifier>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>io.grpc</groupId>
- <artifactId>grpc-netty-shaded</artifactId>
- <scope>test</scope>
- </dependency>
<dependency>
<groupId>io.etcd</groupId>
<artifactId>jetcd-test</artifactId>
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/GrpcSchemaStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/GrpcSchemaStorage.java
new file mode 100644
index 00000000000..fd31afd352a
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/GrpcSchemaStorage.java
@@ -0,0 +1,145 @@
+/*
+ * 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.pulsar.broker.service.schema.grpc;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.protobuf.ByteString;
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion;
+import org.apache.pulsar.common.protocol.schema.SchemaStorage;
+import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.protocol.schema.StoredSchema;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
+import org.apache.pulsar.common.schema.grpc.DeleteSchemaRequest;
+import org.apache.pulsar.common.schema.grpc.GetAllVersionSchemaRequest;
+import org.apache.pulsar.common.schema.grpc.GetAllVersionSchemaResponse;
+import org.apache.pulsar.common.schema.grpc.GetVersionSchemaRequest;
+import org.apache.pulsar.common.schema.grpc.GrpcSchemaStorageServiceGrpc;
+import org.apache.pulsar.common.schema.grpc.GrpcSchemaVersion;
+import org.apache.pulsar.common.schema.grpc.PutSchemaRequest;
+import org.apache.pulsar.common.schema.grpc.VersionedSchemaResponse;
+
+public class GrpcSchemaStorage implements SchemaStorage {
+
+ private final ServiceConfiguration conf;
+ private final Executor executor;
+ private ManagedChannel channel;
+ private GrpcSchemaStorageServiceGrpc.GrpcSchemaStorageServiceFutureStub stub;
+
+ public GrpcSchemaStorage(PulsarService pulsar) {
+ this.conf = pulsar.getConfiguration();
+ this.executor = pulsar.getIoEventLoopGroup();
+ }
+
+ @Override
+ public CompletableFuture<SchemaVersion> put(String key, byte[] value, byte[] hash) {
+ PutSchemaRequest request = PutSchemaRequest.newBuilder()
+ .setKey(key).setValue(ByteString.copyFrom(value)).setHash(ByteString.copyFrom(hash)).build();
+ ListenableFuture<GrpcSchemaVersion> f = stub.put(request);
+ return fromListenableFuture(f).thenApply(v -> BytesSchemaVersion.of(v.getSchemaVersion().toByteArray()));
+ }
+
+ @Override
+ public CompletableFuture<StoredSchema> get(String key, SchemaVersion version) {
+ GetVersionSchemaRequest request = GetVersionSchemaRequest.newBuilder()
+ .setKey(key).setVersion(GrpcSchemaVersion.newBuilder()
+ .setSchemaVersion(ByteString.copyFrom(version.bytes()))).build();
+
+ ListenableFuture<VersionedSchemaResponse> f = stub.get(request);
+ return fromListenableFuture(f).thenApply(v ->
+ new StoredSchema(v.getData().toByteArray(), BytesSchemaVersion.of(v.getVersion().toByteArray())));
+ }
+
+ @Override
+ public CompletableFuture<List<CompletableFuture<StoredSchema>>> getAll(String key) {
+ GetAllVersionSchemaRequest request = GetAllVersionSchemaRequest
+ .newBuilder().setKey(key).build();
+ ListenableFuture<GetAllVersionSchemaResponse> future = stub.getAll(request);
+ return fromListenableFuture(future).thenApply(values -> {
+ if (values.getSchemasCount() == 0) {
+ return List.of();
+ }
+ List<CompletableFuture<StoredSchema>> futures = new ArrayList<>();
+ values.getSchemasList().forEach(schema -> {
+ byte[] data = schema.getData().toByteArray();
+ SchemaVersion version = BytesSchemaVersion.of(schema.getVersion().toByteArray());
+ futures.add(CompletableFuture.completedFuture(new StoredSchema(data, version)));
+ });
+ return futures;
+ });
+ }
+
+ @Override
+ public CompletableFuture<SchemaVersion> delete(String key, boolean forcefully) {
+ DeleteSchemaRequest request = DeleteSchemaRequest.newBuilder()
+ .setKey(key).setForce(forcefully).build();
+
+ ListenableFuture<GrpcSchemaVersion> f = stub.delete(request);
+ return fromListenableFuture(f).thenApply(v -> BytesSchemaVersion.of(v.getSchemaVersion().toByteArray()));
+ }
+
+ @Override
+ public CompletableFuture<SchemaVersion> delete(String key) {
+ return delete(key, false);
+ }
+
+ @Override
+ public SchemaVersion versionFromBytes(byte[] version) {
+ ByteBuffer bb = ByteBuffer.wrap(version);
+ return new LongSchemaVersion(bb.getLong());
+ }
+
+ @Override
+ public void start() throws Exception {
+ String address = conf.getSchemaRegistryStorageGrpcEndpoint();
+ channel = ManagedChannelBuilder.forTarget(address).build();
+ stub = GrpcSchemaStorageServiceGrpc.newFutureStub(channel);
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (null != stub) {
+ stub = null;
+ }
+ if (null != channel) {
+ channel.shutdown();
+ }
+ }
+
+ private <T> CompletableFuture<T> fromListenableFuture(ListenableFuture<T> future) {
+ CompletableFuture<T> ret = new CompletableFuture<>();
+ future.addListener(() -> {
+ try {
+ T v = future.get();
+ ret.complete(v);
+ } catch (Throwable t) {
+ ret.completeExceptionally(t);
+ }
+ }, executor);
+ return ret;
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/GrpcSchemaStorageFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/GrpcSchemaStorageFactory.java
new file mode 100644
index 00000000000..4c50960b149
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/GrpcSchemaStorageFactory.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.pulsar.broker.service.schema.grpc;
+
+import javax.validation.constraints.NotNull;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.service.schema.SchemaStorageFactory;
+import org.apache.pulsar.common.protocol.schema.SchemaStorage;
+
+@SuppressWarnings("unused")
+public class GrpcSchemaStorageFactory implements SchemaStorageFactory {
+ @Override
+ @NotNull
+ public SchemaStorage create(PulsarService pulsar) throws Exception {
+ return new GrpcSchemaStorage(pulsar);
+ }
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/package-info.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/package-info.java
new file mode 100644
index 00000000000..7b455f916cc
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/schema/grpc/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * 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.pulsar.broker.service.schema.grpc;
\ No newline at end of file
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/GrpcSchemaStorageTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/GrpcSchemaStorageTest.java
new file mode 100644
index 00000000000..3fd3bce828a
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/schema/GrpcSchemaStorageTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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.pulsar.broker.service.schema;
+
+import com.google.protobuf.ByteString;
+import io.grpc.Grpc;
+import io.grpc.InsecureServerCredentials;
+import io.grpc.Server;
+import io.grpc.stub.StreamObserver;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.schema.grpc.GrpcSchemaStorageFactory;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.common.protocol.schema.SchemaStorage;
+import org.apache.pulsar.common.protocol.schema.SchemaVersion;
+import org.apache.pulsar.common.schema.LongSchemaVersion;
+import org.apache.pulsar.common.schema.grpc.*;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class GrpcSchemaStorageTest extends ProducerConsumerBase {
+
+ private Server server;
+
+ @BeforeClass
+ @Override
+ protected void setup() throws Exception {
+ server = Grpc.newServerBuilderForPort(8005, InsecureServerCredentials.create())
+ .addService(new GrpcSchemaStorageServer())
+ .build();
+ conf.setManagedLedgerCacheEvictionIntervalMs(10000);
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ server.shutdown();
+ super.internalCleanup();
+ }
+
+ private SchemaStorage buildSchemaStorage() throws Exception {
+ PulsarService pulsar = Mockito.spy(PulsarService.class);
+ Mockito.doReturn(Executors.newSingleThreadExecutor()).when(pulsar).getIoEventLoopGroup();
+ ServiceConfiguration conf = new ServiceConfiguration();
+ conf.setSchemaRegistryStorageGrpcEndpoint("localhost:8005");
+ conf.setSchemaRegistryStorageGrpcEnableTls(false);
+ Mockito.doReturn(conf).when(pulsar).getConfig();
+ Mockito.doReturn(conf).when(pulsar).getConfiguration();
+ GrpcSchemaStorageFactory factory = new GrpcSchemaStorageFactory();
+ return factory.create(pulsar);
+ }
+
+ @Test
+ public void testPutSchema() throws Exception {
+ SchemaStorage schemaStorage = buildSchemaStorage();
+ byte[] value = UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8);
+ CompletableFuture<SchemaVersion> version = schemaStorage.put("key1", value, new byte[0]);
+ SchemaVersion result = version.get();
+ long version0 = ((LongSchemaVersion) schemaStorage.versionFromBytes(result.bytes())).getVersion();
+ Assert.assertEquals(version0, 0);
+ version = schemaStorage.put("key1", value, new byte[0]);
+ version0 = ((LongSchemaVersion) schemaStorage.versionFromBytes(result.bytes())).getVersion();
+ Assert.assertEquals(version0, 1);
+ version = schemaStorage.put("key1", value, new byte[0]);
+ version0 = ((LongSchemaVersion) schemaStorage.versionFromBytes(result.bytes())).getVersion();
+ Assert.assertEquals(version0, 2);
+ }
+
+
+ static class GrpcSchemaStorageServer extends GrpcSchemaStorageServiceGrpc.GrpcSchemaStorageServiceImplBase {
+ private final Map<String, AtomicInteger> versionMap = new ConcurrentHashMap<>();
+ private final Map<String, Map<Long, byte[]>> schemaStorage = new ConcurrentHashMap<>();
+
+ @Override
+ public void get(GetVersionSchemaRequest request,
+ StreamObserver<VersionedSchemaResponse> responseObserver) {
+ String key = request.getKey();
+ byte[] versionBytes = request.getVersion().getSchemaVersion().toByteArray();
+ long version = ByteBuffer.wrap(versionBytes).getLong();
+
+ VersionedSchemaResponse ret = null;
+ Map<Long, byte[]> versionMap = schemaStorage.get(key);
+ if (versionMap != null && versionMap.containsKey(version)) {
+ byte[] data = versionMap.get(version);
+ ret = VersionedSchemaResponse.newBuilder()
+ .setData(ByteString.copyFrom(data))
+ .setVersion(GrpcSchemaVersion.newBuilder()
+ .setSchemaVersion(ByteString.copyFrom(versionBytes)).build())
+ .build();
+ }
+ responseObserver.onNext(ret);
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void getAll(GetAllVersionSchemaRequest request, StreamObserver<GetAllVersionSchemaResponse> responseObserver) {
+ String key = request.getKey();
+ Map<Long, byte[]> versionMap = schemaStorage.get(key);
+ List<VersionedSchemaResponse> rets = new ArrayList<>();
+ if (versionMap != null && !versionMap.isEmpty()) {
+ versionMap.forEach((version, bytes) -> {
+ byte[] versionBytes = new byte[8];
+ ByteBuffer.wrap(versionBytes).putLong(version);
+ VersionedSchemaResponse ret = VersionedSchemaResponse.newBuilder()
+ .setData(ByteString.copyFrom(bytes))
+ .setVersion(GrpcSchemaVersion.newBuilder()
+ .setSchemaVersion(ByteString.copyFrom(versionBytes)).build())
+ .build();
+ rets.add(ret);
+ });
+ }
+
+ GetAllVersionSchemaResponse allSchemas = GetAllVersionSchemaResponse
+ .newBuilder().addAllSchemas(rets).build();
+
+ responseObserver.onNext(allSchemas);
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void put(PutSchemaRequest request,
+ StreamObserver<GrpcSchemaVersion> responseObserver) {
+ String key = request.getKey();
+ byte[] data = request.getValue().toByteArray();
+ byte[] hash = request.getHash().toByteArray();
+ long version = versionMap.computeIfAbsent(key, k -> new AtomicInteger()).incrementAndGet();
+ Map<Long, byte[]> storage = schemaStorage.computeIfAbsent(key, k -> new HashMap<>());
+ storage.put(version, data);
+ byte[] versionBytes = new byte[8];
+ ByteBuffer.wrap(versionBytes).putLong(version);
+ responseObserver.onNext(GrpcSchemaVersion
+ .newBuilder().setSchemaVersion(ByteString.copyFrom(versionBytes)).build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void delete(DeleteSchemaRequest request,
+ StreamObserver<GrpcSchemaVersion> responseObserver) {
+ String key = request.getKey();
+ Map<Long, byte[]> values = schemaStorage.remove(key);
+ if (values.isEmpty()) {
+ responseObserver.onNext(null);
+ responseObserver.onCompleted();
+ return;
+ }
+
+ long maxVersion = values.keySet().stream().max(Comparator.comparingLong(Long::longValue)).get();
+ byte[] versionBytes = new byte[8];
+ ByteBuffer.wrap(versionBytes).putLong(maxVersion);
+
+ responseObserver.onNext(GrpcSchemaVersion.newBuilder()
+ .setSchemaVersion(ByteString.copyFrom(versionBytes)).build());
+ responseObserver.onCompleted();
+ }
+ }
+}
diff --git a/pulsar-common/pom.xml b/pulsar-common/pom.xml
index aa7e4998e5c..a5b9eba69c7 100644
--- a/pulsar-common/pom.xml
+++ b/pulsar-common/pom.xml
@@ -34,6 +34,31 @@
<description>Common libraries needed by client/broker/tools</description>
<dependencies>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-netty-shaded</artifactId>
+ <version>${grpc.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-protobuf</artifactId>
+ <version>${grpc.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.grpc</groupId>
+ <artifactId>grpc-stub</artifactId>
+ <version>${grpc.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency> <!-- necessary for Java 9+ -->
+ <groupId>org.apache.tomcat</groupId>
+ <artifactId>annotations-api</artifactId>
+ <version>6.0.53</version>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client-api</artifactId>
@@ -286,6 +311,14 @@
<groupId>com.github.splunk.lightproto</groupId>
<artifactId>lightproto-maven-plugin</artifactId>
<version>${lightproto-maven-plugin.version}</version>
+ <configuration>
+ <sources>
+ <source>${project.basedir}/src/main/proto/PulsarApi.proto</source>
+ <source>${project.basedir}/src/main/proto/PulsarMarkers.proto</source>
+ </sources>
+ <targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
+ <targetTestSourcesSubDir>generated-sources/lightproto/java</targetTestSourcesSubDir>
+ </configuration>
<executions>
<execution>
<goals>
@@ -385,6 +418,37 @@
</execution>
</executions>
</plugin>
+
+ <plugin>
+ <groupId>org.xolstice.maven.plugins</groupId>
+ <artifactId>protobuf-maven-plugin</artifactId>
+ <version>${protobuf-maven-plugin.version}</version>
+ <configuration>
+ <protocArtifact>com.google.protobuf:protoc:${protobuf3.version}:exe:${os.detected.classifier}</protocArtifact>
+ <pluginId>grpc-java</pluginId>
+ <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
+ <excludes>
+ <exclude>**/PulsarApi.proto</exclude>
+ <exclude>**/PulsarMarkers.proto</exclude>
+ </excludes>
+ </configuration>
+ <executions>
+ <execution>
+ <goals>
+ <goal>compile</goal>
+ <goal>compile-custom</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
+
+ <extensions>
+ <extension>
+ <groupId>kr.motd.maven</groupId>
+ <artifactId>os-maven-plugin</artifactId>
+ <version>${os-maven-plugin.version}</version>
+ </extension>
+ </extensions>
</build>
</project>
diff --git a/pulsar-common/src/main/proto/GrpcSchemaService.proto b/pulsar-common/src/main/proto/GrpcSchemaService.proto
new file mode 100644
index 00000000000..cf7abe53fe5
--- /dev/null
+++ b/pulsar-common/src/main/proto/GrpcSchemaService.proto
@@ -0,0 +1,64 @@
+/**
+ * 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.
+ */
+syntax = "proto3";
+
+package pulsar.service.schema;
+option java_package = "org.apache.pulsar.common.schema.grpc";
+option optimize_for = SPEED;
+option java_multiple_files=true;
+
+message GrpcSchemaVersion {
+ bytes schemaVersion = 1;
+}
+
+message VersionedSchemaResponse {
+ bytes data = 1;
+ GrpcSchemaVersion version = 2;
+}
+
+message GetAllVersionSchemaResponse {
+ repeated VersionedSchemaResponse schemas = 1;
+}
+
+message PutSchemaRequest {
+ string key = 1;
+ bytes value = 2;
+ bytes hash = 3;
+}
+
+message GetVersionSchemaRequest {
+ string key = 1;
+ GrpcSchemaVersion version = 2;
+}
+
+message GetAllVersionSchemaRequest {
+ string key = 1;
+}
+
+message DeleteSchemaRequest {
+ string key = 1;
+ bool force = 2;
+}
+
+service GrpcSchemaStorageService {
+ rpc Put(PutSchemaRequest) returns (GrpcSchemaVersion) {}
+ rpc Get(GetVersionSchemaRequest) returns (VersionedSchemaResponse) {}
+ rpc GetAll(GetAllVersionSchemaRequest) returns (GetAllVersionSchemaResponse) {}
+ rpc Delete(DeleteSchemaRequest) returns (GrpcSchemaVersion) {}
+}
\ No newline at end of file