Skip to content

Commit c286628

Browse files
merlimatmerlimat
authored
Migrate stream non-gRPC protos to LightProto (#4781)
* Fix LedgerHandle.batchReadUnconfirmedAsync: use slog log instead of LOG The batchReadUnconfirmedAsync method added in #4739 calls LOG.error(...), but LedgerHandle was migrated to slog and only has a lowercase `log` field. Master fails to compile. Convert the call to the slog builder style used elsewhere in the file. * Migrate stream non-gRPC protos to LightProto Migrates the stream module's non-gRPC proto definitions to LightProto: - stream/statelib/src/main/proto/kv.proto (KV state-store commands) - stream/proto/src/main/proto/cluster.proto (cluster metadata/assignment) Both files use proto3, including oneof (kv.proto) and map (cluster.proto), which LightProto handles directly. For stream/proto, cluster.proto is moved to a parallel proto-lightproto/ directory so the existing protobuf-maven-plugin keeps generating the rest of the protos (which still depend on the gRPC service stubs). Java sources that touched the migrated types are updated to use the LightProto mutable-instance API instead of the protobuf-java builder pattern. This change is independent of the other in-flight LightProto work in bookkeeper-proto and bookkeeper-server. gRPC service migration is left for a follow-up. * Fix checkstyle: remove unused ServerAssignmentData import
1 parent a0231ed commit c286628

14 files changed

Lines changed: 214 additions & 240 deletions

File tree

‎stream/proto/pom.xml‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,15 @@
6969
</extension>
7070
</extensions>
7171
<plugins>
72+
<plugin>
73+
<groupId>org.apache.rat</groupId>
74+
<artifactId>apache-rat-plugin</artifactId>
75+
<configuration>
76+
<excludes combine.children="append">
77+
<exclude>target/generated-sources/lightproto/**</exclude>
78+
</excludes>
79+
</configuration>
80+
</plugin>
7281
<plugin>
7382
<groupId>org.apache.maven.plugins</groupId>
7483
<artifactId>maven-compiler-plugin</artifactId>
@@ -96,6 +105,24 @@
96105
</execution>
97106
</executions>
98107
</plugin>
108+
<plugin>
109+
<groupId>io.streamnative.lightproto</groupId>
110+
<artifactId>lightproto-maven-plugin</artifactId>
111+
<version>${lightproto-maven-plugin.version}</version>
112+
<configuration>
113+
<sources>
114+
<source>${project.basedir}/src/main/proto-lightproto/cluster.proto</source>
115+
</sources>
116+
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
117+
</configuration>
118+
<executions>
119+
<execution>
120+
<goals>
121+
<goal>generate</goal>
122+
</goals>
123+
</execution>
124+
</executions>
125+
</plugin>
99126
<plugin>
100127
<groupId>org.apache.maven.plugins</groupId>
101128
<artifactId>maven-jar-plugin</artifactId>

stream/proto/src/main/proto/cluster.proto renamed to stream/proto/src/main/proto-lightproto/cluster.proto

File renamed without changes.

‎stream/statelib/pom.xml‎

Lines changed: 16 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -61,8 +61,8 @@
6161
<artifactId>rocksdbjni</artifactId>
6262
</dependency>
6363
<dependency>
64-
<groupId>com.google.protobuf</groupId>
65-
<artifactId>protobuf-java</artifactId>
64+
<groupId>io.netty</groupId>
65+
<artifactId>netty-buffer</artifactId>
6666
</dependency>
6767
<dependency>
6868
<groupId>org.apache.distributedlog</groupId>
@@ -104,25 +104,27 @@
104104
</dependency>
105105
</dependencies>
106106
<build>
107-
<extensions>
108-
<extension>
109-
<groupId>kr.motd.maven</groupId>
110-
<artifactId>os-maven-plugin</artifactId>
111-
<version>${os-maven-plugin.version}</version>
112-
</extension>
113-
</extensions>
114107
<plugins>
115108
<plugin>
116-
<groupId>org.xolstice.maven.plugins</groupId>
117-
<artifactId>protobuf-maven-plugin</artifactId>
118-
<version>${protobuf-maven-plugin.version}</version>
109+
<groupId>org.apache.rat</groupId>
110+
<artifactId>apache-rat-plugin</artifactId>
119111
<configuration>
120-
<protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
112+
<excludes combine.children="append">
113+
<exclude>target/generated-sources/lightproto/**</exclude>
114+
</excludes>
115+
</configuration>
116+
</plugin>
117+
<plugin>
118+
<groupId>io.streamnative.lightproto</groupId>
119+
<artifactId>lightproto-maven-plugin</artifactId>
120+
<version>${lightproto-maven-plugin.version}</version>
121+
<configuration>
122+
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
121123
</configuration>
122124
<executions>
123125
<execution>
124126
<goals>
125-
<goal>compile</goal>
127+
<goal>generate</goal>
126128
</goals>
127129
</execution>
128130
</executions>

‎stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVCommandProcessor.java‎

Lines changed: 7 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020

2121
import static org.apache.bookkeeper.statelib.impl.kv.KVUtils.newCommand;
2222

23-
import com.google.protobuf.InvalidProtocolBufferException;
2423
import io.netty.buffer.ByteBuf;
2524
import io.netty.buffer.ByteBufUtil;
2625
import lombok.AccessLevel;
@@ -48,8 +47,8 @@ public void applyCommand(long revision, ByteBuf cmdBuf, RocksdbKVStore<byte[], b
4847
Command command;
4948
try {
5049
command = newCommand(cmdBuf);
51-
} catch (InvalidProtocolBufferException e) {
52-
log.error().attr("buffer", cmdBuf).attr("txid", revision)
50+
} catch (RuntimeException e) {
51+
log.error().exception(e).attr("buffer", cmdBuf).attr("txid", revision)
5352
.log("Invalid kv command found");
5453
// TODO: better to handle this
5554
return;
@@ -74,25 +73,25 @@ public void applyCommand(long revision, ByteBuf cmdBuf, RocksdbKVStore<byte[], b
7473

7574
private void applyPutCommand(long revision, Command command, RocksdbKVStore<byte[], byte[]> store) {
7675
PutRequest putReq = command.getPutReq();
77-
byte[] keyBytes = putReq.getKey().toByteArray();
78-
byte[] valBytes = putReq.getValue().toByteArray();
76+
byte[] keyBytes = putReq.getKey();
77+
byte[] valBytes = putReq.getValue();
7978
@Cleanup("release") ByteBuf serializedValBuf = KVUtils.serialize(valBytes, revision);
8079
byte[] serializedValBytes = ByteBufUtil.getBytes(serializedValBuf);
8180
store.put(keyBytes, serializedValBytes, revision);
8281
}
8382

8483
private void applyPutIfAbsentCommand(long revision, Command command, RocksdbKVStore<byte[], byte[]> store) {
8584
PutRequest putReq = command.getPutReq();
86-
byte[] keyBytes = putReq.getKey().toByteArray();
87-
byte[] valBytes = putReq.getValue().toByteArray();
85+
byte[] keyBytes = putReq.getKey();
86+
byte[] valBytes = putReq.getValue();
8887
@Cleanup("release") ByteBuf serializedValBuf = KVUtils.serialize(valBytes, revision);
8988
byte[] serializedValBytes = ByteBufUtil.getBytes(serializedValBuf);
9089
store.putIfAbsent(keyBytes, serializedValBytes, revision);
9190
}
9291

9392
private void applyDeleteCommand(long revision, Command command, RocksdbKVStore<byte[], byte[]> store) {
9493
DeleteRequest delReq = command.getDelReq();
95-
byte[] keyBytes = delReq.getKey().toByteArray();
94+
byte[] keyBytes = delReq.getKey();
9695
store.delete(keyBytes, revision);
9796
}
9897
}

‎stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/KVUtils.java‎

Lines changed: 14 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -18,22 +18,14 @@
1818
*/
1919
package org.apache.bookkeeper.statelib.impl.kv;
2020

21-
import com.google.protobuf.InvalidProtocolBufferException;
22-
import com.google.protobuf.UnsafeByteOperations;
2321
import io.netty.buffer.ByteBuf;
24-
import io.netty.buffer.ByteBufOutputStream;
2522
import io.netty.buffer.PooledByteBufAllocator;
2623
import io.netty.util.ReferenceCountUtil;
27-
import java.io.IOException;
2824
import lombok.AccessLevel;
2925
import lombok.CustomLog;
3026
import lombok.NoArgsConstructor;
3127
import org.apache.bookkeeper.common.coder.Coder;
3228
import org.apache.bookkeeper.proto.statestore.kv.Command;
33-
import org.apache.bookkeeper.proto.statestore.kv.DeleteRequest;
34-
import org.apache.bookkeeper.proto.statestore.kv.NopRequest;
35-
import org.apache.bookkeeper.proto.statestore.kv.PutIfAbsentRequest;
36-
import org.apache.bookkeeper.proto.statestore.kv.PutRequest;
3729

3830
/**
3931
* Utils for kv stores.
@@ -42,9 +34,13 @@
4234
@NoArgsConstructor(access = AccessLevel.PRIVATE)
4335
final class KVUtils {
4436

45-
static final Command NOP_CMD = Command.newBuilder()
46-
.setNopReq(NopRequest.newBuilder().build())
47-
.build();
37+
static final Command NOP_CMD = newNopCommand();
38+
39+
private static Command newNopCommand() {
40+
Command cmd = new Command();
41+
cmd.setNopReq();
42+
return cmd;
43+
}
4844

4945
static ByteBuf serialize(ByteBuf valBuf, long revision) {
5046
int serializedSize = valBuf.readableBytes() + Long.BYTES;
@@ -68,41 +64,21 @@ static <V> V deserialize(Coder<V> valCoder,
6864
return valCoder.decode(valBuf);
6965
}
7066

71-
static Command newCommand(ByteBuf cmdBuf) throws InvalidProtocolBufferException {
72-
return Command.parseFrom(cmdBuf.nioBuffer());
67+
static Command newCommand(ByteBuf cmdBuf) {
68+
Command cmd = new Command();
69+
cmd.parseFrom(cmdBuf, cmdBuf.readableBytes());
70+
return cmd;
7371
}
7472

75-
static ByteBuf newCommandBuf(Command cmd) throws IOException {
73+
static ByteBuf newCommandBuf(Command cmd) {
7674
ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(cmd.getSerializedSize());
7775
try {
78-
cmd.writeTo(new ByteBufOutputStream(buf));
79-
} catch (IOException e) {
76+
cmd.writeTo(buf);
77+
} catch (RuntimeException e) {
8078
ReferenceCountUtil.release(buf);
8179
throw e;
8280
}
8381
return buf;
8482
}
8583

86-
static PutRequest newPutRequest(byte[] keyBytes,
87-
byte[] valBytes) {
88-
return PutRequest.newBuilder()
89-
.setKey(UnsafeByteOperations.unsafeWrap(keyBytes))
90-
.setValue(UnsafeByteOperations.unsafeWrap(valBytes))
91-
.build();
92-
}
93-
94-
static PutIfAbsentRequest newPutIfAbsentRequest(byte[] keyBytes,
95-
byte[] valBytes) {
96-
return PutIfAbsentRequest.newBuilder()
97-
.setKey(UnsafeByteOperations.unsafeWrap(keyBytes))
98-
.setValue(UnsafeByteOperations.unsafeWrap(valBytes))
99-
.build();
100-
}
101-
102-
static DeleteRequest newDeleteRequest(byte[] keyBytes) {
103-
return DeleteRequest.newBuilder()
104-
.setKey(UnsafeByteOperations.unsafeWrap(keyBytes))
105-
.build();
106-
}
107-
10884
}

‎stream/statelib/src/main/java/org/apache/bookkeeper/statelib/impl/kv/RocksdbKVAsyncStore.java‎

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import io.netty.buffer.ByteBufUtil;
2323
import io.netty.buffer.Unpooled;
2424
import io.netty.util.ReferenceCountUtil;
25-
import java.io.IOException;
2625
import java.util.concurrent.CompletableFuture;
2726
import java.util.function.Supplier;
2827
import org.apache.bookkeeper.common.coder.Coder;
@@ -91,9 +90,8 @@ public CompletableFuture<Void> put(K key, V value) {
9190
byte[] keyBytes = keyCoder.encode(key);
9291
byte[] valBytes = valCoder.encode(value);
9392

94-
Command command = Command.newBuilder()
95-
.setPutReq(KVUtils.newPutRequest(keyBytes, valBytes))
96-
.build();
93+
Command command = new Command();
94+
command.setPutReq().setKey(keyBytes).setValue(valBytes);
9795
return writeCommandReturnTxId(command).thenApplyAsync((revision) -> {
9896
ByteBuf serializedBuf = KVUtils.serialize(valBytes, revision);
9997
try {
@@ -112,9 +110,8 @@ public CompletableFuture<V> putIfAbsent(K key, V value) {
112110
byte[] keyBytes = keyCoder.encode(key);
113111
byte[] valBytes = valCoder.encode(value);
114112

115-
Command command = Command.newBuilder()
116-
.setPutIfAbsentReq(KVUtils.newPutIfAbsentRequest(keyBytes, valBytes))
117-
.build();
113+
Command command = new Command();
114+
command.setPutIfAbsentReq().setKey(keyBytes).setValue(valBytes);
118115
return writeCommandReturnTxId(command).thenApplyAsync((revision) -> {
119116
ByteBuf serializedBuf = KVUtils.serialize(valBytes, revision);
120117
try {
@@ -135,9 +132,8 @@ public CompletableFuture<V> putIfAbsent(K key, V value) {
135132
public CompletableFuture<V> delete(K key) {
136133
byte[] keyBytes = keyCoder.encode(key);
137134

138-
Command command = Command.newBuilder()
139-
.setDelReq(KVUtils.newDeleteRequest(keyBytes))
140-
.build();
135+
Command command = new Command();
136+
command.setDelReq().setKey(keyBytes);
141137
return writeCommandReturnTxId(command).thenApplyAsync((revision) -> {
142138
byte[] prevValue = localStore.delete(keyBytes, revision);
143139
if (null == prevValue) {
@@ -171,7 +167,7 @@ private CompletableFuture<Long> writeCommandReturnTxId(Command command) {
171167
try {
172168
ByteBuf cmdBuf = KVUtils.newCommandBuf(command);
173169
return writeCommandBufReturnTxId(cmdBuf);
174-
} catch (IOException e) {
170+
} catch (RuntimeException e) {
175171
return FutureUtils.exception(e);
176172
}
177173
}

‎stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/InMemClusterMetadataStore.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -56,10 +56,8 @@ synchronized int getNumWatchers() {
5656
@Override
5757
public synchronized boolean initializeCluster(int numStorageContainers,
5858
Optional<String> segmentStorePath) {
59-
this.metadata = ClusterMetadata.newBuilder()
60-
.setNumStorageContainers(numStorageContainers)
61-
.build();
62-
this.assignmentData = ClusterAssignmentData.newBuilder().build();
59+
this.metadata = new ClusterMetadata().setNumStorageContainers(numStorageContainers);
60+
this.assignmentData = new ClusterAssignmentData();
6361
return true;
6462
}
6563

‎stream/storage/impl/src/main/java/org/apache/bookkeeper/stream/storage/impl/cluster/ZkClusterMetadataStore.java‎

Lines changed: 22 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import static org.apache.bookkeeper.stream.storage.StorageConstants.getStoragePath;
2626
import static org.apache.bookkeeper.stream.storage.StorageConstants.getWritableServersPath;
2727

28-
import com.google.protobuf.InvalidProtocolBufferException;
2928
import java.io.IOException;
3029
import java.util.HashMap;
3130
import java.util.Map;
@@ -95,11 +94,8 @@ public void close() {
9594

9695
@Override
9796
public boolean initializeCluster(int numStorageContainers, Optional<String> segmentStorePath) {
98-
ClusterMetadata metadata = ClusterMetadata.newBuilder()
99-
.setNumStorageContainers(numStorageContainers)
100-
.build();
101-
ClusterAssignmentData assignmentData = ClusterAssignmentData.newBuilder()
102-
.build();
97+
ClusterMetadata metadata = new ClusterMetadata().setNumStorageContainers(numStorageContainers);
98+
ClusterAssignmentData assignmentData = new ClusterAssignmentData();
10399
try {
104100
// we are using dlog for the storage backend, so we need to initialize the dlog namespace
105101
BKDLConfig dlogConfig = new BKDLConfig(
@@ -130,10 +126,16 @@ public boolean initializeCluster(int numStorageContainers, Optional<String> segm
130126
public ClusterAssignmentData getClusterAssignmentData() {
131127
try {
132128
byte[] data = client.getData().forPath(zkClusterAssignmentPath);
133-
return ClusterAssignmentData.parseFrom(data);
134-
} catch (InvalidProtocolBufferException ie) {
135-
throw new StorageRuntimeException("The cluster assignment data from zookeeper @"
136-
+ zkClusterAssignmentPath + " is corrupted", ie);
129+
ClusterAssignmentData assignmentData = new ClusterAssignmentData();
130+
try {
131+
assignmentData.parseFrom(data);
132+
} catch (RuntimeException ie) {
133+
throw new StorageRuntimeException("The cluster assignment data from zookeeper @"
134+
+ zkClusterAssignmentPath + " is corrupted", ie);
135+
}
136+
return assignmentData;
137+
} catch (StorageRuntimeException e) {
138+
throw e;
137139
} catch (Exception e) {
138140
throw new StorageRuntimeException("Failed to fetch cluster assignment data from zookeeper @"
139141
+ zkClusterAssignmentPath, e);
@@ -193,10 +195,16 @@ public void unwatchClusterAssignmentData(Consumer<Void> watcher) {
193195
public ClusterMetadata getClusterMetadata() {
194196
try {
195197
byte[] data = client.getData().forPath(zkClusterMetadataPath);
196-
return ClusterMetadata.parseFrom(data);
197-
} catch (InvalidProtocolBufferException ie) {
198-
throw new StorageRuntimeException("The cluster metadata from zookeeper @"
199-
+ zkClusterMetadataPath + " is corrupted", ie);
198+
ClusterMetadata metadata = new ClusterMetadata();
199+
try {
200+
metadata.parseFrom(data);
201+
} catch (RuntimeException ie) {
202+
throw new StorageRuntimeException("The cluster metadata from zookeeper @"
203+
+ zkClusterMetadataPath + " is corrupted", ie);
204+
}
205+
return metadata;
206+
} catch (StorageRuntimeException e) {
207+
throw e;
200208
} catch (Exception e) {
201209
throw new StorageRuntimeException("Failed to fetch cluster metadata from zookeeper @"
202210
+ zkClusterMetadataPath, e);

0 commit comments

Comments
 (0)