Skip to content

Commit 451b6a5

Browse files
lhotarilhotari
authored
Fix table service txn request corruption on silent client retries (#4829)
TableClientTest.testTableAPIServerSideRouting has been flaky since 2018 (#1440), failing at assertTrue(txnResult.isSuccess()) with no error in the client logs. Root cause: table service requests embed ByteBuf slices and serializing a request drains them, so a request instance must not be reused across RPC attempts. Both txn client paths reused the same TxnRequest across silent retries: PByteBufSimpleTableImpl.TxnImpl.commit() passed one pre-populated request into RetryUtils.execute (server-side routing) and TxnRequestProcessor.createRequest() returned the same stored request on every attempt (client-side routing). A transient first-attempt failure (e.g. a cold storage-container proxy channel returning NOT_FOUND by design) made the retry serialize the drained request, sending corrupted bytes. The server then answered gRPC-OK with code=BAD_REQUEST and succeeded=false, which KvUtils.newKvTxnResult conflated with a legitimately failed compare because it never checks the response code. put/get/delete/increment already build a fresh request per attempt; txn was the outlier. Changes: - Store the compare/success/failure ops in both TxnImpl classes and build a fresh TxnRequest for every RPC attempt. - TxnRequestProcessor now takes a Supplier<TxnRequest>. - PByteBufSimpleTableImpl surfaces non-SUCCESS txn response codes as InternalServerException instead of isSuccess()=false, matching TxnRequestProcessor.processResponse. - Regression tests that fail against the previous implementation. Assisted-by: Claude Code
1 parent 15a961b commit 451b6a5

5 files changed

Lines changed: 338 additions & 36 deletions

File tree

‎stream/clients/java/kv/src/main/java/org/apache/bookkeeper/clients/impl/kv/PByteBufSimpleTableImpl.java‎

Lines changed: 47 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -64,14 +64,17 @@
6464
import org.apache.bookkeeper.api.kv.result.PutResult;
6565
import org.apache.bookkeeper.api.kv.result.RangeResult;
6666
import org.apache.bookkeeper.api.kv.result.TxnResult;
67+
import org.apache.bookkeeper.clients.exceptions.InternalServerException;
6768
import org.apache.bookkeeper.clients.utils.RetryUtils;
69+
import org.apache.bookkeeper.common.concurrent.FutureUtils;
6870
import org.apache.bookkeeper.stream.proto.StreamProperties;
6971
import org.apache.bookkeeper.stream.proto.kv.rpc.DeleteRangeRequest;
7072
import org.apache.bookkeeper.stream.proto.kv.rpc.IncrementRequest;
7173
import org.apache.bookkeeper.stream.proto.kv.rpc.PutRequest;
7274
import org.apache.bookkeeper.stream.proto.kv.rpc.RangeRequest;
7375
import org.apache.bookkeeper.stream.proto.kv.rpc.RoutingHeader;
7476
import org.apache.bookkeeper.stream.proto.kv.rpc.TxnRequest;
77+
import org.apache.bookkeeper.stream.proto.storage.StatusCode;
7578

7679
/**
7780
* A {@link PTable} implementation using simple grpc calls.
@@ -260,21 +263,22 @@ public void close() {
260263
class TxnImpl implements Txn<ByteBuf, ByteBuf> {
261264

262265
private final ByteBuf pKey;
263-
private final TxnRequest txnRequest;
264-
private final List<AutoCloseable> resourcesToRelease;
266+
private final List<CompareOp<ByteBuf, ByteBuf>> compareOps;
267+
private final List<Op<ByteBuf, ByteBuf>> successOps;
268+
private final List<Op<ByteBuf, ByteBuf>> failureOps;
265269

266270
TxnImpl(ByteBuf pKey) {
267271
this.pKey = pKey.retain();
268-
this.txnRequest = new TxnRequest();
269-
this.resourcesToRelease = Lists.newArrayList();
272+
this.compareOps = Lists.newArrayList();
273+
this.successOps = Lists.newArrayList();
274+
this.failureOps = Lists.newArrayList();
270275
}
271276

272277
@SuppressWarnings("unchecked")
273278
@Override
274279
public Txn<ByteBuf, ByteBuf> If(CompareOp... cmps) {
275280
for (CompareOp<ByteBuf, ByteBuf> cmp : cmps) {
276-
populateProtoCompare(txnRequest.addCompare(), cmp);
277-
resourcesToRelease.add(cmp);
281+
compareOps.add(cmp);
278282
}
279283
return this;
280284
}
@@ -283,8 +287,7 @@ public Txn<ByteBuf, ByteBuf> If(CompareOp... cmps) {
283287
@Override
284288
public Txn<ByteBuf, ByteBuf> Then(Op... ops) {
285289
for (Op<ByteBuf, ByteBuf> op : ops) {
286-
populateProtoRequest(txnRequest.addSuccess(), op);
287-
resourcesToRelease.add(op);
290+
successOps.add(op);
288291
}
289292
return this;
290293
}
@@ -293,25 +296,54 @@ public Txn<ByteBuf, ByteBuf> Then(Op... ops) {
293296
@Override
294297
public Txn<ByteBuf, ByteBuf> Else(Op... ops) {
295298
for (Op<ByteBuf, ByteBuf> op : ops) {
296-
populateProtoRequest(txnRequest.addFailure(), op);
297-
resourcesToRelease.add(op);
299+
failureOps.add(op);
298300
}
299301
return this;
300302
}
301303

304+
// Serializing a request drains the ByteBuf slices stored in it, so a request instance
305+
// must not be reused across RPC attempts: a retried attempt would send a corrupted
306+
// request. Build a fresh request per attempt, like put/get/delete/increment above.
307+
private TxnRequest newTxnRequest() {
308+
TxnRequest txnRequest = new TxnRequest();
309+
for (CompareOp<ByteBuf, ByteBuf> cmp : compareOps) {
310+
populateProtoCompare(txnRequest.addCompare(), cmp);
311+
}
312+
for (Op<ByteBuf, ByteBuf> op : successOps) {
313+
populateProtoRequest(txnRequest.addSuccess(), op);
314+
}
315+
for (Op<ByteBuf, ByteBuf> op : failureOps) {
316+
populateProtoRequest(txnRequest.addFailure(), op);
317+
}
318+
populateRoutingHeader(txnRequest.setHeader(), pKey);
319+
return txnRequest;
320+
}
321+
302322
@Override
303323
public CompletableFuture<TxnResult<ByteBuf, ByteBuf>> commit() {
304-
populateRoutingHeader(txnRequest.setHeader(), pKey);
305324
return retryUtils.execute(() -> fromListenableFuture(
306325
ClientCalls.futureUnaryCall(
307326
getChannel(pKey).newCall(getTxnMethod(), getCallOptions()),
308-
txnRequest)
327+
newTxnRequest())
309328
))
310-
.thenApply(response -> KvUtils.newKvTxnResult(response, resultFactory, kvFactory))
329+
.thenCompose(response -> {
330+
if (StatusCode.SUCCESS != response.getHeader().getCode()) {
331+
// A server-side error must not be conflated with a failed txn compare.
332+
return FutureUtils.exception(new InternalServerException(
333+
"Encountered internal server exception : code = " + response.getHeader().getCode()));
334+
}
335+
return FutureUtils.value(KvUtils.newKvTxnResult(response, resultFactory, kvFactory));
336+
})
311337
.whenComplete((ignored, cause) -> {
312338
ReferenceCountUtil.release(pKey);
313-
for (AutoCloseable resource : resourcesToRelease) {
314-
closeResource(resource);
339+
for (CompareOp<ByteBuf, ByteBuf> cmp : compareOps) {
340+
closeResource(cmp);
341+
}
342+
for (Op<ByteBuf, ByteBuf> op : successOps) {
343+
closeResource(op);
344+
}
345+
for (Op<ByteBuf, ByteBuf> op : failureOps) {
346+
closeResource(op);
315347
}
316348
});
317349
}

‎stream/clients/java/kv/src/main/java/org/apache/bookkeeper/clients/impl/kv/PByteBufTableRangeImpl.java‎

Lines changed: 36 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -209,21 +209,22 @@ public OpFactory<ByteBuf, ByteBuf> opFactory() {
209209
class TxnImpl implements Txn<ByteBuf, ByteBuf> {
210210

211211
private final ByteBuf pKey;
212-
private final TxnRequest txnRequest;
213-
private final List<AutoCloseable> resourcesToRelease;
212+
private final List<CompareOp<ByteBuf, ByteBuf>> compareOps;
213+
private final List<Op<ByteBuf, ByteBuf>> successOps;
214+
private final List<Op<ByteBuf, ByteBuf>> failureOps;
214215

215216
TxnImpl(ByteBuf pKey) {
216217
this.pKey = pKey.retain();
217-
this.txnRequest = new TxnRequest();
218-
this.resourcesToRelease = Lists.newArrayList();
218+
this.compareOps = Lists.newArrayList();
219+
this.successOps = Lists.newArrayList();
220+
this.failureOps = Lists.newArrayList();
219221
}
220222

221223
@SuppressWarnings("unchecked")
222224
@Override
223225
public Txn<ByteBuf, ByteBuf> If(CompareOp... cmps) {
224226
for (CompareOp<ByteBuf, ByteBuf> cmp : cmps) {
225-
populateProtoCompare(txnRequest.addCompare(), cmp);
226-
resourcesToRelease.add(cmp);
227+
compareOps.add(cmp);
227228
}
228229
return this;
229230
}
@@ -232,8 +233,7 @@ public Txn<ByteBuf, ByteBuf> If(CompareOp... cmps) {
232233
@Override
233234
public Txn<ByteBuf, ByteBuf> Then(Op... ops) {
234235
for (Op<ByteBuf, ByteBuf> op : ops) {
235-
populateProtoRequest(txnRequest.addSuccess(), op);
236-
resourcesToRelease.add(op);
236+
successOps.add(op);
237237
}
238238
return this;
239239
}
@@ -242,25 +242,47 @@ public Txn<ByteBuf, ByteBuf> Then(Op... ops) {
242242
@Override
243243
public Txn<ByteBuf, ByteBuf> Else(Op... ops) {
244244
for (Op<ByteBuf, ByteBuf> op : ops) {
245-
populateProtoRequest(txnRequest.addFailure(), op);
246-
resourcesToRelease.add(op);
245+
failureOps.add(op);
247246
}
248247
return this;
249248
}
250249

250+
// Serializing a request drains the ByteBuf slices stored in it, so a request instance
251+
// must not be reused across RPC attempts: a retried attempt would send a corrupted
252+
// request. Build a fresh request per attempt.
253+
private TxnRequest newTxnRequest() {
254+
TxnRequest txnRequest = new TxnRequest();
255+
for (CompareOp<ByteBuf, ByteBuf> cmp : compareOps) {
256+
populateProtoCompare(txnRequest.addCompare(), cmp);
257+
}
258+
for (Op<ByteBuf, ByteBuf> op : successOps) {
259+
populateProtoRequest(txnRequest.addSuccess(), op);
260+
}
261+
for (Op<ByteBuf, ByteBuf> op : failureOps) {
262+
populateProtoRequest(txnRequest.addFailure(), op);
263+
}
264+
populateRoutingHeader(txnRequest.setHeader(), pKey);
265+
return txnRequest;
266+
}
267+
251268
@Override
252269
public CompletableFuture<TxnResult<ByteBuf, ByteBuf>> commit() {
253-
populateRoutingHeader(txnRequest.setHeader(), pKey);
254270
return TxnRequestProcessor.of(
255-
txnRequest,
271+
this::newTxnRequest,
256272
response -> KvUtils.newKvTxnResult(response, resultFactory, kvFactory),
257273
scChannel,
258274
executor,
259275
backoffPolicy
260276
).process().whenComplete((ignored, cause) -> {
261277
ReferenceCountUtil.release(pKey);
262-
for (AutoCloseable resource : resourcesToRelease) {
263-
closeResource(resource);
278+
for (CompareOp<ByteBuf, ByteBuf> cmp : compareOps) {
279+
closeResource(cmp);
280+
}
281+
for (Op<ByteBuf, ByteBuf> op : successOps) {
282+
closeResource(op);
283+
}
284+
for (Op<ByteBuf, ByteBuf> op : failureOps) {
285+
closeResource(op);
264286
}
265287
});
266288
}

‎stream/clients/java/kv/src/main/java/org/apache/bookkeeper/clients/impl/kv/TxnRequestProcessor.java‎

Lines changed: 9 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import com.google.common.util.concurrent.ListenableFuture;
3636
import java.util.concurrent.ScheduledExecutorService;
3737
import java.util.function.Function;
38+
import java.util.function.Supplier;
3839
import org.apache.bookkeeper.clients.exceptions.InternalServerException;
3940
import org.apache.bookkeeper.clients.impl.channel.StorageServerChannel;
4041
import org.apache.bookkeeper.clients.impl.container.StorageContainerChannel;
@@ -51,30 +52,32 @@ class TxnRequestProcessor<RespT>
5152
extends ListenableFutureRpcProcessor<TxnRequest, TxnResponse, RespT> {
5253

5354
public static <T> TxnRequestProcessor<T> of(
54-
TxnRequest request,
55+
Supplier<TxnRequest> requestSupplier,
5556
Function<TxnResponse, T> responseFunc,
5657
StorageContainerChannel channel,
5758
ScheduledExecutorService executor,
5859
Policy backoffPolicy) {
59-
return new TxnRequestProcessor<>(request, responseFunc, channel, executor, backoffPolicy);
60+
return new TxnRequestProcessor<>(requestSupplier, responseFunc, channel, executor, backoffPolicy);
6061
}
6162

62-
private final TxnRequest request;
63+
private final Supplier<TxnRequest> requestSupplier;
6364
private final Function<TxnResponse, RespT> responseFunc;
6465

65-
private TxnRequestProcessor(TxnRequest request,
66+
private TxnRequestProcessor(Supplier<TxnRequest> requestSupplier,
6667
Function<TxnResponse, RespT> respFunc,
6768
StorageContainerChannel channel,
6869
ScheduledExecutorService executor,
6970
Policy backoffPolicy) {
7071
super(channel, executor, backoffPolicy);
71-
this.request = request;
72+
this.requestSupplier = requestSupplier;
7273
this.responseFunc = respFunc;
7374
}
7475

7576
@Override
7677
protected TxnRequest createRequest() {
77-
return request;
78+
// Serializing a request drains the ByteBuf slices stored in it, so a request instance
79+
// must not be reused across RPC attempts: build a fresh request for every attempt.
80+
return requestSupplier.get();
7881
}
7982

8083
@Override

0 commit comments

Comments
 (0)