Skip to content

Commit 47efc0d

Browse files
merlimatmerlimat
authored
Migrate BookkeeperProtocol from protobuf-java to LightProto (apache#4780)
* Fix LedgerHandle.batchReadUnconfirmedAsync: use slog log instead of LOG The batchReadUnconfirmedAsync method added in apache#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 DataFormats and DbLedgerStorageDataFormats from protobuf-java to LightProto Replace Google's protobuf-java with StreamNative LightProto for the storage and metadata formats in `bookkeeper-proto`. The wire protocol (`BookkeeperProtocol`) remains on protobuf-java for now. LightProto generates mutable, reusable, ByteBuf-aware classes with built-in proto2 TextFormat (de)serialization (via `generateTextFormat=true`), so the existing TextFormat-based znode payloads (cookies, auditor votes, lock data, underreplication entries, layout) round-trip byte-identically. Notable behavior changes: - `BookKeeper.DigestType.toProtoDigestType` now returns the LightProto-generated enum (same constants, different package). - v3 ledger metadata uses a hand-rolled length-prefixed delimited writer/reader matching protobuf's `writeDelimitedTo`/`mergeDelimitedFrom`. * Add SpotBugs excludes for lightproto-generated classes The existing exclude `~org.apache.bookkeeper.proto.DataFormats.*` matched protobuf's nested `DataFormats$LedgerMetadataFormat` etc. LightProto generates the same messages as flat top-level classes (`LedgerMetadataFormat` directly in `org.apache.bookkeeper.proto`), so those weren't excluded and triggered 12 bugs in the generated code (bit-twiddling, exposed internal byte arrays, etc.) that aren't actionable. Replace the obsolete `DataFormats.*` exclude with explicit per-message patterns covering both packages and `LightProtoCodec` (also generated per-package). * Fix checkstyle: remove redundant same-package import in MockBookies * Migrate BookkeeperProtocol from protobuf-java to LightProto Migrates the BookkeeperProtocol.proto wire protocol to use LightProto for serialization. Combined with the prior migrations of DataFormats and DbLedgerStorageDataFormats, this drops the protobuf-java runtime dependency from bookkeeper-proto entirely. LightProto produces wire-compatible output with protoc for the same .proto, so on-the-wire bookie/client compatibility is preserved. Notes on lifecycle handling: - LightProto messages parsed from a ByteBuf hold lazy references into that buffer for field access. The decoders now call materialize() on parsed Request/Response/AuthMessage instances so they survive after the source buffer is released. - Server response paths that put entry payloads into ReadLacResponse or ReadResponse now copy the bytes via ByteBufUtil.getBytes(...), matching the previous ByteString.copyFrom semantics. Drive-by fix: processWriteLacRequestV3/processReadLacRequestV3 were ordering work on r.getAddRequest().getLedgerId() instead of the matching WriteLac/ReadLac request. With protobuf this returned a default 0 for the unset field; with LightProto it throws IllegalStateException. * Fix checkstyle: remove redundant same-package imports * Fix shaded-jar tests after protobuf-java removal protobuf-java is no longer pulled in transitively by bookkeeper-server, so the shaded jars no longer contain (shaded) protobuf classes, and the flat lightproto-generated classes have replaced the BookkeeperProtocol outer class. - BookKeeperServerShadedJarTest / DistributedLogCoreShadedJarTest: drop the now-irrelevant testProtobufShadedPath checks and switch the BookkeeperProtocol presence check to a real lightproto class (AddRequest). - Drop the dead com.google.protobuf:protobuf-java <include> from the three shade plugin configs (bookkeeper-server-shaded, bookkeeper-server-tests-shaded, distributedlog-core-shaded). * Use ByteBufList.toByteBuf to avoid copying request bodies Replace ByteBufList.coalesce(...) with a new ByteBufList#toByteBuf method on the WriteLac and AddEntry request paths. coalesce allocates a new buffer and copies all the bytes; toByteBuf wraps the existing buffers in a CompositeByteBuf (or returns the single buffer directly when the list has one entry, or Unpooled.EMPTY_BUFFER when empty), transferring ownership to the caller and releasing the source ByteBufList. Adds unit tests covering the empty / single / multi-buffer paths, including ref-count behaviour for both the list and the underlying buffers. * Fix ByteBufList.toByteBuf to not release the source list The previous implementation released the source ByteBufList when producing the wrapping ByteBuf, but callers in PerChannelBookieClient only wrap a buffer they don't own &mdash; the ByteBufList's lifecycle is managed by the upstream PendingAddOp, which shares the same list across multiple bookies in a quorum write. Releasing in toByteBuf produced a double-release / use-after-free that surfaced as IllegalReferenceCountException in tests like BookieStickyReadsTest. Change toByteBuf to leave the source list's ref count untouched and return a wrapper that holds its own retains (a CompositeByteBuf for multiple buffers, a retainedDuplicate for the single-buffer fast path, or Unpooled.EMPTY_BUFFER for empty). This matches the original ByteBufList.coalesce semantics from the caller's point of view, while still avoiding the byte copy. Tests updated to assert the new ownership semantics. * Don't read required ledgerId/entryId from error responses Four completion handlers (AddCompletion, ForceLedgerCompletion, ReadCompletion, WriteLacCompletion) were always reading the inner *Response's ledgerId/entryId fields, even when the outer Response carried an error status. On error responses (e.g. EUA from a rejected SASL handshake) those required fields are not populated. Under protobuf-java they returned the default 0; under LightProto they throw IllegalStateException("Field 'ledgerId' is not set"), which surfaced as GSSAPIBookKeeperTest.testNotAllowedClientId blowing up on the client side after the server rejected the auth. Read the inner response only when status is EOK and hasXxxResponse() is true. Otherwise fall back to the request's ledgerId/entryId, which the CompletionValue base class already records. * Read response ledgerId/entryId on long-poll reads in ReadCompletion Long-poll reads send entryId=LAST_ADD_CONFIRMED and the bookie fills in the actual entry id (and ledgerId) on the response when an entry is returned. The previous fix in ReadCompletion always used the request's recorded entryId, which made the long-poll path look like an empty piggy-back response on the client side and dropped the entry buffer. Read ledgerId/entryId from the response when status is EOK and the inner ReadResponse is present; fall back to the request's recorded values only on error envelopes (where the inner response may be missing or unpopulated). Fixes TestReadLastConfirmedAndEntry.testRaceOnLastAddConfirmed.
1 parent 9633bb7 commit 47efc0d

57 files changed

Lines changed: 769 additions & 1022 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎bookkeeper-proto/pom.xml‎

Lines changed: 0 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,6 @@
2525
<artifactId>bookkeeper-proto</artifactId>
2626
<name>Apache BookKeeper :: Protocols</name>
2727
<dependencies>
28-
<dependency>
29-
<groupId>com.google.protobuf</groupId>
30-
<artifactId>protobuf-java</artifactId>
31-
</dependency>
3228
<dependency>
3329
<groupId>io.netty</groupId>
3430
<artifactId>netty-buffer</artifactId>
@@ -42,40 +38,16 @@
4238
<configuration>
4339
<excludes>
4440
<!-- exclude generated files //-->
45-
<exclude>**/BookkeeperProtocol.java</exclude>
4641
<exclude>target/generated-sources/lightproto/**</exclude>
4742
<exclude>**/.checkstyle</exclude>
4843
</excludes>
4944
</configuration>
5045
</plugin>
51-
<plugin>
52-
<groupId>org.xolstice.maven.plugins</groupId>
53-
<artifactId>protobuf-maven-plugin</artifactId>
54-
<version>${protobuf-maven-plugin.version}</version>
55-
<configuration>
56-
<protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
57-
<checkStaleness>true</checkStaleness>
58-
<includes>
59-
<include>BookkeeperProtocol.proto</include>
60-
</includes>
61-
</configuration>
62-
<executions>
63-
<execution>
64-
<goals>
65-
<goal>compile</goal>
66-
</goals>
67-
</execution>
68-
</executions>
69-
</plugin>
7046
<plugin>
7147
<groupId>io.streamnative.lightproto</groupId>
7248
<artifactId>lightproto-maven-plugin</artifactId>
7349
<version>${lightproto-maven-plugin.version}</version>
7450
<configuration>
75-
<sources>
76-
<source>${project.basedir}/src/main/proto/DataFormats.proto</source>
77-
<source>${project.basedir}/src/main/proto/DbLedgerStorageDataFormats.proto</source>
78-
</sources>
7951
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
8052
<generateTextFormat>true</generateTextFormat>
8153
</configuration>

‎bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookieInfoReader.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@
3535
import org.apache.bookkeeper.net.BookieId;
3636
import org.apache.bookkeeper.proto.BookieClient;
3737
import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GetBookieInfoCallback;
38-
import org.apache.bookkeeper.proto.BookkeeperProtocol;
38+
import org.apache.bookkeeper.proto.GetBookieInfoRequest;
3939
import org.apache.commons.collections4.CollectionUtils;
4040

4141
/**
@@ -46,8 +46,8 @@
4646
@CustomLog
4747
public class BookieInfoReader {
4848
private static final long GET_BOOKIE_INFO_REQUEST_FLAGS =
49-
BookkeeperProtocol.GetBookieInfoRequest.Flags.TOTAL_DISK_CAPACITY_VALUE
50-
| BookkeeperProtocol.GetBookieInfoRequest.Flags.FREE_DISK_SPACE_VALUE;
49+
GetBookieInfoRequest.Flags.TOTAL_DISK_CAPACITY_VALUE
50+
| GetBookieInfoRequest.Flags.FREE_DISK_SPACE_VALUE;
5151

5252
private final ScheduledExecutorService scheduler;
5353
private final BookKeeper bk;
@@ -329,8 +329,8 @@ synchronized void getReadWriteBookieInfo() {
329329
}
330330

331331
BookieClient bkc = bk.getBookieClient();
332-
final long requested = BookkeeperProtocol.GetBookieInfoRequest.Flags.TOTAL_DISK_CAPACITY_VALUE
333-
| BookkeeperProtocol.GetBookieInfoRequest.Flags.FREE_DISK_SPACE_VALUE;
332+
final long requested = GetBookieInfoRequest.Flags.TOTAL_DISK_CAPACITY_VALUE
333+
| GetBookieInfoRequest.Flags.FREE_DISK_SPACE_VALUE;
334334
totalSent = 0;
335335
completedCnt = 0;
336336
errorCnt = 0;
@@ -413,8 +413,8 @@ Map<BookieId, BookieInfo> getBookieInfo() throws BKException, InterruptedExcepti
413413
final ConcurrentMap<BookieId, BookieInfo> map =
414414
new ConcurrentHashMap<BookieId, BookieInfo>();
415415
final CountDownLatch latch = new CountDownLatch(1);
416-
long requested = BookkeeperProtocol.GetBookieInfoRequest.Flags.TOTAL_DISK_CAPACITY_VALUE
417-
| BookkeeperProtocol.GetBookieInfoRequest.Flags.FREE_DISK_SPACE_VALUE;
416+
long requested = GetBookieInfoRequest.Flags.TOTAL_DISK_CAPACITY_VALUE
417+
| GetBookieInfoRequest.Flags.FREE_DISK_SPACE_VALUE;
418418

419419
Collection<BookieId> bookies;
420420
bookies = bk.bookieWatcher.getBookies();

‎bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AddCompletion.java‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -117,25 +117,30 @@ public void setOutstanding() {
117117

118118
@Override
119119
public void handleV2Response(
120-
long ledgerId, long entryId, BookkeeperProtocol.StatusCode status,
120+
long ledgerId, long entryId, StatusCode status,
121121
BookieProtocol.Response response) {
122122
perChannelBookieClient.addEntryOutstanding.dec();
123123
handleResponse(ledgerId, entryId, status);
124124
}
125125

126126
@Override
127127
public void handleV3Response(
128-
BookkeeperProtocol.Response response) {
128+
Response response) {
129129
perChannelBookieClient.addEntryOutstanding.dec();
130-
BookkeeperProtocol.AddResponse addResponse = response.getAddResponse();
131-
BookkeeperProtocol.StatusCode status = response.getStatus() == BookkeeperProtocol.StatusCode.EOK
132-
? addResponse.getStatus() : response.getStatus();
133-
handleResponse(addResponse.getLedgerId(), addResponse.getEntryId(),
134-
status);
130+
StatusCode status;
131+
if (response.getStatus() == StatusCode.EOK && response.hasAddResponse()) {
132+
status = response.getAddResponse().getStatus();
133+
} else {
134+
// Error responses (e.g. EUA from a rejected auth handshake) may not
135+
// carry an AddResponse with ledgerId/entryId populated. Fall back to
136+
// the values we recorded from the outgoing request.
137+
status = response.getStatus();
138+
}
139+
handleResponse(ledgerId, entryId, status);
135140
}
136141

137142
private void handleResponse(long ledgerId, long entryId,
138-
BookkeeperProtocol.StatusCode status) {
143+
StatusCode status) {
139144
logEvent(status).log("Got response from bookie");
140145

141146
int rc = convertStatus(status, BKException.Code.WriteException);

‎bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/AuthHandler.java‎

Lines changed: 46 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222

2323
import static org.apache.bookkeeper.auth.AuthProviderFactoryFactory.AUTHENTICATION_DISABLED_PLUGIN_NAME;
2424

25-
import com.google.protobuf.ByteString;
2625
import io.netty.buffer.ByteBuf;
2726
import io.netty.channel.Channel;
2827
import io.netty.channel.ChannelDuplexHandler;
@@ -40,7 +39,6 @@
4039
import org.apache.bookkeeper.auth.BookieAuthProvider;
4140
import org.apache.bookkeeper.auth.ClientAuthProvider;
4241
import org.apache.bookkeeper.client.BKException;
43-
import org.apache.bookkeeper.proto.BookkeeperProtocol.AuthMessage;
4442
import org.apache.bookkeeper.util.ByteBufList;
4543
import org.apache.bookkeeper.util.NettyChannelUtil;
4644

@@ -90,10 +88,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
9088
BookieProtocol.AuthRequest req = (BookieProtocol.AuthRequest) msg;
9189
assert (req.getOpCode() == BookieProtocol.AUTH);
9290
if (checkAuthPlugin(req.getAuthMessage(), ctx.channel())) {
93-
byte[] payload = req
94-
.getAuthMessage()
95-
.getPayload()
96-
.toByteArray();
91+
byte[] payload = req.getAuthMessage().getPayload();
9792
authProvider.process(AuthToken.wrap(payload),
9893
new AuthResponseCallbackLegacy(req, ctx.channel()));
9994
} else {
@@ -114,26 +109,23 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
114109
} else {
115110
ctx.channel().close();
116111
}
117-
} else if (msg instanceof BookkeeperProtocol.Request) { // post-PB-client
118-
BookkeeperProtocol.Request req = (BookkeeperProtocol.Request) msg;
119-
if (req.getHeader().getOperation() == BookkeeperProtocol.OperationType.AUTH
112+
} else if (msg instanceof Request) { // post-PB-client
113+
Request req = (Request) msg;
114+
if (req.getHeader().getOperation() == OperationType.AUTH
120115
&& req.hasAuthRequest()
121116
&& checkAuthPlugin(req.getAuthRequest(), ctx.channel())) {
122-
byte[] payload = req
123-
.getAuthRequest()
124-
.getPayload()
125-
.toByteArray();
117+
byte[] payload = req.getAuthRequest().getPayload();
126118
authProvider.process(AuthToken.wrap(payload),
127119
new AuthResponseCallback(req, ctx.channel(), authProviderFactory.getPluginName()));
128-
} else if (req.getHeader().getOperation() == BookkeeperProtocol.OperationType.START_TLS
120+
} else if (req.getHeader().getOperation() == OperationType.START_TLS
129121
&& req.hasStartTLSRequest()) {
130122
super.channelRead(ctx, msg);
131123
} else {
132-
BookkeeperProtocol.Response.Builder builder = BookkeeperProtocol.Response.newBuilder()
133-
.setHeader(req.getHeader())
134-
.setStatus(BookkeeperProtocol.StatusCode.EUA);
124+
Response response = new Response();
125+
response.setHeader().copyFrom(req.getHeader());
126+
response.setStatus(StatusCode.EUA);
135127

136-
NettyChannelUtil.writeAndFlushWithVoidPromise(ctx.channel(), builder.build());
128+
NettyChannelUtil.writeAndFlushWithVoidPromise(ctx.channel(), response);
137129
}
138130
} else {
139131
// close the channel, junk coming over it
@@ -171,45 +163,43 @@ public void operationComplete(int rc, AuthToken newam) {
171163
channel.close();
172164
return;
173165
}
174-
AuthMessage message = AuthMessage.newBuilder().setAuthPluginName(req.authMessage.getAuthPluginName())
175-
.setPayload(ByteString.copyFrom(newam.getData())).build();
166+
AuthMessage message = new AuthMessage()
167+
.setAuthPluginName(req.authMessage.getAuthPluginName())
168+
.setPayload(newam.getData());
176169
final BookieProtocol.AuthResponse response =
177170
new BookieProtocol.AuthResponse(req.getProtocolVersion(), message);
178171
NettyChannelUtil.writeAndFlushWithVoidPromise(channel, response);
179172
}
180173
}
181174

182175
static class AuthResponseCallback implements AuthCallbacks.GenericCallback<AuthToken> {
183-
final BookkeeperProtocol.Request req;
176+
final Request req;
184177
final Channel channel;
185178
final String pluginName;
186179

187-
AuthResponseCallback(BookkeeperProtocol.Request req, Channel channel, String pluginName) {
180+
AuthResponseCallback(Request req, Channel channel, String pluginName) {
188181
this.req = req;
189182
this.channel = channel;
190183
this.pluginName = pluginName;
191184
}
192185

193186
@Override
194187
public void operationComplete(int rc, AuthToken newam) {
195-
BookkeeperProtocol.Response.Builder builder = BookkeeperProtocol.Response.newBuilder()
196-
.setHeader(req.getHeader());
188+
Response response = new Response();
189+
response.setHeader().copyFrom(req.getHeader());
197190

198191
if (rc != BKException.Code.OK) {
199192
log.error("Error processing auth message, closing connection");
200193

201-
builder.setStatus(BookkeeperProtocol.StatusCode.EUA);
202-
NettyChannelUtil.writeAndFlushWithClosePromise(
203-
channel, builder.build()
204-
);
194+
response.setStatus(StatusCode.EUA);
195+
NettyChannelUtil.writeAndFlushWithClosePromise(channel, response);
205196
return;
206197
} else {
207-
AuthMessage message = AuthMessage.newBuilder().setAuthPluginName(pluginName)
208-
.setPayload(ByteString.copyFrom(newam.getData())).build();
209-
builder.setStatus(BookkeeperProtocol.StatusCode.EOK).setAuthResponse(message);
210-
NettyChannelUtil.writeAndFlushWithVoidPromise(
211-
channel, builder.build()
212-
);
198+
response.setStatus(StatusCode.EOK)
199+
.setAuthResponse()
200+
.setAuthPluginName(pluginName)
201+
.setPayload(newam.getData());
202+
NettyChannelUtil.writeAndFlushWithVoidPromise(channel, response);
213203
}
214204
}
215205
}
@@ -272,8 +262,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
272262

273263
if (authenticated) {
274264
super.channelRead(ctx, msg);
275-
} else if (msg instanceof BookkeeperProtocol.Response) {
276-
BookkeeperProtocol.Response resp = (BookkeeperProtocol.Response) msg;
265+
} else if (msg instanceof Response) {
266+
Response resp = (Response) msg;
277267
if (null == resp.getHeader().getOperation()) {
278268
log.info()
279269
.attr("message", msg)
@@ -286,11 +276,11 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
286276
super.channelRead(ctx, msg);
287277
break;
288278
case AUTH:
289-
if (resp.getStatus() != BookkeeperProtocol.StatusCode.EOK) {
290-
authenticationError(ctx, resp.getStatus().getNumber());
279+
if (resp.getStatus() != StatusCode.EOK) {
280+
authenticationError(ctx, resp.getStatus().getValue());
291281
} else {
292282
assert (resp.hasAuthResponse());
293-
BookkeeperProtocol.AuthMessage am = resp.getAuthResponse();
283+
AuthMessage am = resp.getAuthResponse();
294284
if (AUTHENTICATION_DISABLED_PLUGIN_NAME.equals(am.getAuthPluginName())){
295285
SocketAddress remote = ctx.channel().remoteAddress();
296286
log.info().attr("client", remote)
@@ -299,7 +289,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
299289
cb.operationComplete(BKException.Code.OK, null);
300290
return;
301291
}
302-
byte[] payload = am.getPayload().toByteArray();
292+
byte[] payload = am.getPayload();
303293
authProvider.process(AuthToken.wrap(payload), new AuthRequestCallback(ctx,
304294
authProviderFactory.getPluginName()));
305295
}
@@ -321,7 +311,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
321311
if (resp.errorCode != BookieProtocol.EOK) {
322312
authenticationError(ctx, resp.errorCode);
323313
} else {
324-
BookkeeperProtocol.AuthMessage am = ((BookieProtocol.AuthResponse) resp).authMessage;
314+
AuthMessage am = ((BookieProtocol.AuthResponse) resp).authMessage;
325315
if (AUTHENTICATION_DISABLED_PLUGIN_NAME.equals(am.getAuthPluginName())) {
326316
SocketAddress remote = ctx.channel().remoteAddress();
327317
log.info().attr("client", remote)
@@ -330,7 +320,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
330320
cb.operationComplete(BKException.Code.OK, null);
331321
return;
332322
}
333-
byte[] payload = am.getPayload().toByteArray();
323+
byte[] payload = am.getPayload();
334324
authProvider.process(AuthToken.wrap(payload), new AuthRequestCallback(ctx,
335325
authProviderFactory.getPluginName()));
336326
}
@@ -353,12 +343,12 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
353343
if (authenticated) {
354344
super.write(ctx, msg, promise);
355345
super.flush(ctx);
356-
} else if (msg instanceof BookkeeperProtocol.Request) {
346+
} else if (msg instanceof Request) {
357347
// let auth messages through, queue the rest
358-
BookkeeperProtocol.Request req = (BookkeeperProtocol.Request) msg;
348+
Request req = (Request) msg;
359349
if (req.getHeader().getOperation()
360-
== BookkeeperProtocol.OperationType.AUTH
361-
|| req.getHeader().getOperation() == BookkeeperProtocol.OperationType.START_TLS) {
350+
== OperationType.AUTH
351+
|| req.getHeader().getOperation() == OperationType.START_TLS) {
362352
super.write(ctx, msg, promise);
363353
super.flush(ctx);
364354
} else {
@@ -421,22 +411,23 @@ public void operationComplete(int rc, AuthToken newam) {
421411
return;
422412
}
423413

424-
AuthMessage message = AuthMessage.newBuilder().setAuthPluginName(pluginName)
425-
.setPayload(ByteString.copyFrom(newam.getData())).build();
414+
AuthMessage message = new AuthMessage()
415+
.setAuthPluginName(pluginName)
416+
.setPayload(newam.getData());
426417

427418
if (isUsingV2Protocol) {
428419
final BookieProtocol.AuthRequest msg =
429420
new BookieProtocol.AuthRequest(BookieProtocol.CURRENT_PROTOCOL_VERSION, message);
430421
NettyChannelUtil.writeAndFlushWithVoidPromise(channel, msg);
431422
} else {
432423
// V3 protocol
433-
BookkeeperProtocol.BKPacketHeader header = BookkeeperProtocol.BKPacketHeader.newBuilder()
434-
.setVersion(BookkeeperProtocol.ProtocolVersion.VERSION_THREE)
435-
.setOperation(BookkeeperProtocol.OperationType.AUTH).setTxnId(newTxnId()).build();
436-
BookkeeperProtocol.Request.Builder builder = BookkeeperProtocol.Request.newBuilder()
437-
.setHeader(header)
438-
.setAuthRequest(message);
439-
NettyChannelUtil.writeAndFlushWithVoidPromise(channel, builder.build());
424+
Request request = new Request();
425+
request.setHeader()
426+
.setVersion(ProtocolVersion.VERSION_THREE)
427+
.setOperation(OperationType.AUTH)
428+
.setTxnId(newTxnId());
429+
request.setAuthRequest().copyFrom(message);
430+
NettyChannelUtil.writeAndFlushWithVoidPromise(channel, request);
440431
}
441432
}
442433
}

0 commit comments

Comments
 (0)