diff --git a/java-storage/google-cloud-storage/pom.xml b/java-storage/google-cloud-storage/pom.xml index 3fadd4c12810..7acf33654c0c 100644 --- a/java-storage/google-cloud-storage/pom.xml +++ b/java-storage/google-cloud-storage/pom.xml @@ -459,6 +459,8 @@ 1 false + + 900 ch.qos.logback.core.status.NopStatusListener diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AppendableDiag.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AppendableDiag.java new file mode 100644 index 000000000000..8f33f6d57dc5 --- /dev/null +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/AppendableDiag.java @@ -0,0 +1,114 @@ +/* + * Copyright 2026 Google LLC + * + * Licensed 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 com.google.cloud.storage; + +import com.google.storage.v2.BidiWriteObjectRequest; +import com.google.storage.v2.BidiWriteObjectResponse; +import io.grpc.Status; +import java.util.concurrent.atomic.AtomicLong; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DIAGNOSTIC ONLY (do not submit). Logs every message sent / received on appendable upload streams + * so a CI run shows exactly where an upload waits. + */ +final class AppendableDiag { + static final Logger LOG = LoggerFactory.getLogger("com.google.cloud.storage.DIAG"); + private static final AtomicLong IDS = new AtomicLong(); + + private AppendableDiag() {} + + static long nextId() { + return IDS.incrementAndGet(); + } + + /** Never throws: diagnostics must not change behavior (test stubs may not implement getters). */ + static Object safe(java.util.function.Supplier s) { + try { + return s.get(); + } catch (Throwable t) { + return "n/a"; + } + } + + static String req(BidiWriteObjectRequest r) { + StringBuilder sb = new StringBuilder(); + if (r.hasWriteObjectSpec()) { + sb.append("spec=write(") + .append(r.getWriteObjectSpec().getResource().getName()) + .append(", appendable=") + .append(r.getWriteObjectSpec().getAppendable()) + .append(") "); + } else if (r.hasAppendObjectSpec()) { + sb.append("spec=append(") + .append(r.getAppendObjectSpec().getObject()) + .append(", gen=") + .append(r.getAppendObjectSpec().getGeneration()) + .append(", handle=") + .append(r.getAppendObjectSpec().hasWriteHandle()) + .append(", routingToken=") + .append(r.getAppendObjectSpec().hasRoutingToken()) + .append(") "); + } else if (!r.getUploadId().isEmpty()) { + sb.append("uploadId=").append(r.getUploadId()).append(' '); + } + sb.append("offset=").append(r.getWriteOffset()); + if (r.hasChecksummedData()) { + sb.append(" data=").append(r.getChecksummedData().getContent().size()); + } + sb.append(" flush=").append(r.getFlush()); + sb.append(" lookup=").append(r.getStateLookup()); + sb.append(" finish=").append(r.getFinishWrite()); + if (r.hasObjectChecksums()) { + sb.append(" objectCrc32c=").append(r.getObjectChecksums().hasCrc32C()); + } + return sb.toString(); + } + + static String resp(BidiWriteObjectResponse r) { + StringBuilder sb = new StringBuilder(); + if (r.hasPersistedSize()) { + sb.append("persistedSize=").append(r.getPersistedSize()); + } + if (r.hasResource()) { + sb.append("resource(size=") + .append(r.getResource().getSize()) + .append(", gen=") + .append(r.getResource().getGeneration()) + .append(", finalized=") + .append(r.getResource().hasFinalizeTime()) + .append(")"); + } + sb.append(" writeHandle=").append(r.hasWriteHandle()); + return sb.toString(); + } + + static String err(@Nullable Throwable t) { + if (t == null) { + return "null"; + } + Status s = Status.fromThrowable(t); + String msg = String.valueOf(t.getMessage()); + int nl = msg.indexOf('\n'); + if (nl > -1) { + msg = msg.substring(0, nl); + } + return t.getClass().getSimpleName() + "(grpcStatus=" + s.getCode() + "): " + msg; + } +} diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java index eceee9463935..cd677aa205f2 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiAppendableUnbufferedWritableByteChannel.java @@ -97,13 +97,29 @@ public void close() throws IOException { if (!writeCalledAtLeastOnce) { stream.flush(); } + // DIAGNOSTIC ONLY + long loops = 0; + AppendableDiag.LOG.info( + "DIAG channel CLOSE begin writeOffset={} finalize={} expectedCrc32c={}", + writeOffset, + nextWriteShouldFinalize, + expectedCrc32c != null); if (nextWriteShouldFinalize) { //noinspection StatementWithEmptyBody - while (!stream.finishWrite(writeOffset, expectedCrc32c)) {} + while (!stream.finishWrite(writeOffset, expectedCrc32c)) { + loops++; + } } else { //noinspection StatementWithEmptyBody - while (!stream.closeStream(writeOffset)) {} + while (!stream.closeStream(writeOffset)) { + loops++; + } } + AppendableDiag.LOG.info( + "DIAG channel CLOSE enqueued {} after {} retry loop(s); waiting for ack of {}", + nextWriteShouldFinalize ? "finish_write" : "final flush", + loops, + writeOffset); awaitResultFuture(); } finally { @@ -122,8 +138,16 @@ public void nextWriteShouldFinalize(@Nullable String expectedCrc32c) { } void flush() throws InterruptedException { + // DIAGNOSTIC ONLY + long start = System.nanoTime(); + AppendableDiag.LOG.info("DIAG channel FLUSH begin writeOffset={}", writeOffset); stream.flush(); stream.awaitAckOf(writeOffset); + AppendableDiag.LOG.info( + "DIAG channel FLUSH end writeOffset={} waitedMs={} resultDone={}", + writeOffset, + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), + stream.getResultFuture().isDone()); } private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) throws IOException { @@ -191,9 +215,19 @@ private long internalWrite(ByteBuffer[] srcs, int srcsOffset, int srcsLength) th } private void awaitResultFuture() throws IOException { + // DIAGNOSTIC ONLY + long start = System.nanoTime(); try { stream.awaitAckOf(writeOffset); + AppendableDiag.LOG.info( + "DIAG channel ACK-WAIT end writeOffset={} waitedMs={} resultDone={}", + writeOffset, + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), + stream.getResultFuture().isDone()); stream.getResultFuture().get(10_717, TimeUnit.MILLISECONDS); + AppendableDiag.LOG.info( + "DIAG channel RESULT OK totalMs={}", + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); InterruptedIOException ioe = new InterruptedIOException(); @@ -201,6 +235,10 @@ private void awaitResultFuture() throws IOException { writeThrewError = true; throw ioe; } catch (ExecutionException e) { + AppendableDiag.LOG.info( + "DIAG channel RESULT FAILED totalMs={} cause={}", + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), + AppendableDiag.err(e.getCause())); BaseServiceException coalesce = StorageException.coalesce(e.getCause()); String message = coalesce.getMessage(); String ioExceptionMessage = message; @@ -215,6 +253,9 @@ private void awaitResultFuture() throws IOException { writeThrewError = true; throw ioException; } catch (TimeoutException e) { + AppendableDiag.LOG.info( + "DIAG channel RESULT TIMEOUT (ack arrived, result not set in 10.7s) totalMs={}", + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)); writeThrewError = true; throw new IOException(e); } diff --git a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java index b5c7b7af58e1..85e7b693f197 100644 --- a/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java +++ b/java-storage/google-cloud-storage/src/main/java/com/google/cloud/storage/BidiUploadStreamingStream.java @@ -71,6 +71,9 @@ final class BidiUploadStreamingStream { private volatile @Nullable StreamTuple stream; private volatile @Nullable ApiFuture pendingReconciliation; + // DIAGNOSTIC ONLY (do not submit) + private final long diagId = AppendableDiag.nextId(); + BidiUploadStreamingStream( BidiUploadState state, ScheduledExecutorService executor, @@ -81,11 +84,16 @@ final class BidiUploadStreamingStream { this.executor = executor; this.write = write; this.lock = new ReentrantLock(); - this.retryContext = new StreamRetryContextDecorator(retryContext, lock, this::reset); + this.retryContext = new StreamRetryContextDecorator(retryContext, lock, this::reset, diagId); this.onSuccess = this::restart; this.onFailure = t -> { SettableApiFuture resultFuture = state.getResultFuture(); + AppendableDiag.LOG.info( + "DIAG upload#{} TERMINAL FAILURE resultFutureAlreadyDone={} cause={}", + diagId, + resultFuture.isDone(), + AppendableDiag.err(t)); if (!resultFuture.isDone()) { this.state.terminalError(); BaseServiceException coalesced = StorageException.coalesce(t); @@ -248,6 +256,12 @@ long availableCapacity() { void restart() { lock.lock(); try { + AppendableDiag.LOG.info( + "DIAG upload#{} RESTART (reconnect) state={} confirmedBytes={} totalSentBytes={}", + diagId, + AppendableDiag.safe(state::getState), + AppendableDiag.safe(state::getConfirmedBytes), + AppendableDiag.safe(state::getTotalSentBytes)); checkState(stream == null, "attempting to restart stream when stream is already active"); state.retrying(); ApiFuture reconciliation = state.beginReconciliation(); @@ -267,6 +281,8 @@ void restart() { new ApiFutureCallback() { @Override public void onFailure(Throwable t) { + AppendableDiag.LOG.info( + "DIAG upload#{} RECONCILIATION FAILED {}", diagId, AppendableDiag.err(t)); lock.lock(); try { BidiUploadStreamingStream.this.pendingReconciliation = null; @@ -278,6 +294,10 @@ public void onFailure(Throwable t) { @Override public void onSuccess(Void result) { + AppendableDiag.LOG.info( + "DIAG upload#{} RECONCILIATION OK confirmedBytes={}", + diagId, + AppendableDiag.safe(state::getConfirmedBytes)); lock.lock(); try { BidiUploadStreamingStream.this.pendingReconciliation = null; @@ -299,6 +319,8 @@ public void onSuccess(Void result) { } stream = tmp; } catch (Throwable t) { + AppendableDiag.LOG.info( + "DIAG upload#{} RESTART threw {}", diagId, AppendableDiag.err(t)); retryContext.recordError(t, onSuccess, onFailure); } finally { lock.unlock(); @@ -310,6 +332,11 @@ void reset() { lock.lock(); try { StreamTuple tmp = stream; + AppendableDiag.LOG.info( + "DIAG upload#{} RESET (drop current stream before retry) hadStream={} stream#{}", + diagId, + tmp != null, + tmp != null ? tmp.streamDiagId : -1); if (tmp != null) { tmp.in.flagTombstoned(); tmp.closeSend(); @@ -337,8 +364,16 @@ void reset() { private StreamTuple initStreamTuple() { GrpcCallContext grpcCallContext = state.enqueueFirstMessageAndGetGrpcCallContext(); + long streamDiagId = AppendableDiag.nextId(); + AppendableDiag.LOG.info( + "DIAG upload#{} OPEN stream#{} state={} confirmedBytes={} totalSentBytes={}", + diagId, + streamDiagId, + AppendableDiag.safe(state::getState), + AppendableDiag.safe(state::getConfirmedBytes), + AppendableDiag.safe(state::getTotalSentBytes)); StreamingResponseObserver streamResponseObserver = - new StreamingResponseObserver(state, retryContext, onSuccess, onFailure); + new StreamingResponseObserver(state, retryContext, onSuccess, onFailure, streamDiagId); RedirectHandlingResponseObserver responseObserver = new RedirectHandlingResponseObserver( state, @@ -347,11 +382,12 @@ private StreamTuple initStreamTuple() { maxRedirectsAllowed, this::reset, () -> executor.execute(this::restart)); + responseObserver.streamDiagId = streamDiagId; ClientStream clientStream = write.splitCall(responseObserver, grpcCallContext); GracefulOutboundStream out = new GracefulOutboundStream(clientStream); - return new StreamTuple(out, responseObserver); + return new StreamTuple(out, responseObserver, streamDiagId); } private void internalSend() { @@ -364,18 +400,59 @@ private void internalSend() { private static final class StreamTuple implements Consumer { private final ClientStream out; private final RedirectHandlingResponseObserver in; - - StreamTuple(ClientStream out, RedirectHandlingResponseObserver in) { + private final long streamDiagId; + // DIAGNOSTIC ONLY: aggregate plain data messages so 5MB uploads don't flood the log. + private long pendingPlainDataMsgs; + private long pendingPlainDataBytes; + + StreamTuple( + ClientStream out, + RedirectHandlingResponseObserver in, + long streamDiagId) { this.out = out; this.in = in; + this.streamDiagId = streamDiagId; } @Override public void accept(BidiWriteObjectRequest bidiWriteObjectRequest) { + // DIAGNOSTIC ONLY + boolean plainData = + bidiWriteObjectRequest.hasChecksummedData() + && !bidiWriteObjectRequest.getFlush() + && !bidiWriteObjectRequest.getStateLookup() + && !bidiWriteObjectRequest.getFinishWrite() + && !bidiWriteObjectRequest.hasWriteObjectSpec() + && !bidiWriteObjectRequest.hasAppendObjectSpec(); + if (plainData) { + pendingPlainDataMsgs++; + pendingPlainDataBytes += bidiWriteObjectRequest.getChecksummedData().getContent().size(); + if (pendingPlainDataMsgs >= 500) { + logPlainData(); + } + } else { + logPlainData(); + AppendableDiag.LOG.info( + "DIAG stream#{} SEND {}", streamDiagId, AppendableDiag.req(bidiWriteObjectRequest)); + } out.send(bidiWriteObjectRequest); } + private void logPlainData() { + if (pendingPlainDataMsgs > 0) { + AppendableDiag.LOG.info( + "DIAG stream#{} SEND {} plain data message(s), {} bytes", + streamDiagId, + pendingPlainDataMsgs, + pendingPlainDataBytes); + pendingPlainDataMsgs = 0; + pendingPlainDataBytes = 0; + } + } + public void closeSend() { + logPlainData(); + AppendableDiag.LOG.info("DIAG stream#{} CLOSE_SEND (half-close)", streamDiagId); in.flagTombstoned(); out.closeSend(); } @@ -388,6 +465,7 @@ static final class StreamingResponseObserver private final RetryContext retryContext; private final OnSuccess onSuccess; private final OnFailure onFailure; + private final long streamDiagId; @MonotonicNonNull private StreamController controller; @@ -396,15 +474,26 @@ static final class StreamingResponseObserver RetryContext retryContext, OnSuccess onSuccess, OnFailure onFailure) { + this(state, retryContext, onSuccess, onFailure, -1); + } + + StreamingResponseObserver( + BidiUploadState state, + RetryContext retryContext, + OnSuccess onSuccess, + OnFailure onFailure, + long streamDiagId) { this.state = state; this.retryContext = retryContext; this.onSuccess = onSuccess; this.onFailure = onFailure; + this.streamDiagId = streamDiagId; } @EnsuresNonNull("controller") @Override public void onStart(StreamController controller) { + AppendableDiag.LOG.info("DIAG stream#{} ON_START request(1)", streamDiagId); this.controller = controller; controller.disableAutoInboundFlowControl(); controller.request(1); @@ -415,13 +504,22 @@ public void onStart(StreamController controller) { public void onResponse(BidiWriteObjectResponse response) { try { controller.request(1); + long confirmedBytesBefore = state.getConfirmedBytes(); @Nullable StorageException se = state.onResponse(response); + long confirmedBytesAfter = state.getConfirmedBytes(); + boolean progress = confirmedBytesAfter > confirmedBytesBefore; + logRecv(response, confirmedBytesBefore, confirmedBytesAfter, se); if (se != null) { retryContext.recordError(se, onSuccess, onFailure); - } else { + } else if (progress) { + // Only refresh the retry budget when the server confirms new bytes. After a reconnect, + // the state lookup reply repeats the previously confirmed offset; treating it as + // progress would let a persistent retryable error (e.g. 429, 408) retry forever. retryContext.reset(); } } catch (Throwable t) { + AppendableDiag.LOG.info( + "DIAG stream#{} RECV processing threw {}", streamDiagId, AppendableDiag.err(t)); // catch an error that might happen while processing and forward it to our retry context retryContext.recordError(t, onSuccess, onFailure); } @@ -429,13 +527,54 @@ public void onResponse(BidiWriteObjectResponse response) { @Override public void onError(Throwable t) { + try { + AppendableDiag.LOG.info( + "DIAG stream#{} ON_ERROR {} confirmedBytes={} totalSentBytes={}", + streamDiagId, + AppendableDiag.err(t), + safe(state::getConfirmedBytes), + safe(state::getTotalSentBytes)); + } catch (Throwable ignore) { + // diagnostics must never affect behavior + } retryContext.recordError(t, onSuccess, onFailure); } @Override public void onComplete() { + AppendableDiag.LOG.info("DIAG stream#{} ON_COMPLETE (server closed stream OK)", streamDiagId); // ignore } + + // DIAGNOSTIC ONLY: never let logging change behavior (test stubs may not implement getters). + private void logRecv( + BidiWriteObjectResponse response, + long confirmedBytesBefore, + long confirmedBytesAfter, + @Nullable StorageException se) { + try { + AppendableDiag.LOG.info( + "DIAG stream#{} RECV {} confirmedBytes {}->{} totalSentBytes={} resultDone={}" + + " stateError={}", + streamDiagId, + AppendableDiag.resp(response), + confirmedBytesBefore, + confirmedBytesAfter, + safe(state::getTotalSentBytes), + safe(() -> state.getResultFuture().isDone()), + se == null ? "none" : AppendableDiag.err(se)); + } catch (Throwable ignore) { + // diagnostics must never affect behavior + } + } + + private static Object safe(java.util.function.Supplier s) { + try { + return s.get(); + } catch (Throwable t) { + return "n/a"; + } + } } static final class RedirectHandlingResponseObserver @@ -446,6 +585,8 @@ static final class RedirectHandlingResponseObserver private final int maxRedirectsAllowed; private final Runnable beforeRedirect; private final Runnable onRedirect; + // DIAGNOSTIC ONLY + volatile long streamDiagId = -1; private volatile boolean tombstoned; @@ -480,6 +621,7 @@ void flagTombstoned() { @Override public void onStart(StreamController controller) { if (tombstoned) { + AppendableDiag.LOG.info("DIAG stream#{} DROPPED onStart (tombstoned)", streamDiagId); return; } delegate.onStart(controller); @@ -488,6 +630,10 @@ public void onStart(StreamController controller) { @Override public void onResponse(BidiWriteObjectResponse response) { if (tombstoned) { + AppendableDiag.LOG.info( + "DIAG stream#{} DROPPED RECV (tombstoned) {}", + streamDiagId, + AppendableDiag.resp(response)); return; } redirectCounter.set(0); @@ -497,6 +643,8 @@ public void onResponse(BidiWriteObjectResponse response) { @Override public void onError(Throwable t) { if (tombstoned) { + AppendableDiag.LOG.info( + "DIAG stream#{} DROPPED ON_ERROR (tombstoned) {}", streamDiagId, AppendableDiag.err(t)); return; } BidiWriteObjectRedirectedError error = GrpcUtils.getBidiWriteObjectRedirectedError(t); @@ -505,6 +653,14 @@ public void onError(Throwable t) { return; } int redirectCount = redirectCounter.incrementAndGet(); + AppendableDiag.LOG.info( + "DIAG stream#{} REDIRECT #{} (max {}) routingToken={} writeHandle={} generation={}", + streamDiagId, + redirectCount, + maxRedirectsAllowed, + error.hasRoutingToken(), + error.hasWriteHandle(), + error.hasGeneration() ? error.getGeneration() : -1); if (redirectCount > maxRedirectsAllowed) { // attach the fact we're ignoring the redirect to the original exception as a suppressed // Exception. The lower level handler can then perform its usual handling, but if things @@ -521,6 +677,7 @@ public void onError(Throwable t) { @Override public void onComplete() { if (tombstoned) { + AppendableDiag.LOG.info("DIAG stream#{} DROPPED ON_COMPLETE (tombstoned)", streamDiagId); return; } delegate.onComplete(); @@ -580,13 +737,20 @@ static final class StreamRetryContextDecorator implements RetryContext { private final RetryContext retryContext; private final ReentrantLock lock; private final Runnable onRecordError; + private final long diagId; @VisibleForTesting StreamRetryContextDecorator( RetryContext retryContext, ReentrantLock lock, Runnable onRecordError) { + this(retryContext, lock, onRecordError, -1); + } + + StreamRetryContextDecorator( + RetryContext retryContext, ReentrantLock lock, Runnable onRecordError, long diagId) { this.retryContext = retryContext; this.lock = lock; this.onRecordError = onRecordError; + this.diagId = diagId; } @Override @@ -596,22 +760,39 @@ public boolean inBackoff() { @Override public void reset() { + AppendableDiag.LOG.info("DIAG upload#{} RETRY BUDGET RESET (progress confirmed)", diagId); retryContext.reset(); } @Override public void recordError( T t, OnSuccess onSuccess, OnFailure onFailure) { + AppendableDiag.LOG.info("DIAG upload#{} RECORD_ERROR {}", diagId, AppendableDiag.err(t)); + // DIAGNOSTIC ONLY: log the retry decision made by the underlying retry context. + OnSuccess loggingOnSuccess = + () -> { + AppendableDiag.LOG.info("DIAG upload#{} BACKOFF DONE -> retrying", diagId); + onSuccess.onSuccess(); + }; + OnFailure loggingOnFailure = + e -> { + AppendableDiag.LOG.info( + "DIAG upload#{} RETRY GAVE UP -> failing upload: {}", + diagId, + AppendableDiag.err(e), + e); + onFailure.onFailure(e); + }; lock.lock(); try { try { onRecordError.run(); } catch (Throwable tt) { t.addSuppressed(tt); - onFailure.onFailure(t); + loggingOnFailure.onFailure(t); return; } - retryContext.recordError(t, onSuccess, onFailure); + retryContext.recordError(t, loggingOnSuccess, loggingOnFailure); } finally { lock.unlock(); } diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/BidiUploadTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/BidiUploadTest.java index 2774981ecd62..3244ebe13b86 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/BidiUploadTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/BidiUploadTest.java @@ -2126,6 +2126,11 @@ public void onResponse_stateErrorForwardedToRetryContext() { StreamingResponseObserver obs = new StreamingResponseObserver( new BidiUploadState(name.getMethodName()) { + @Override + long getConfirmedBytes() { + return 0; + } + @Override StorageException onResponse(BidiWriteObjectResponse response) { return new StorageException(0, "test-error", null); @@ -2163,6 +2168,11 @@ public void onResponse_exceptionFromStateOnResponseForwardedToRetryContext() { StreamingResponseObserver obs = new StreamingResponseObserver( new BidiUploadState(name.getMethodName()) { + @Override + long getConfirmedBytes() { + return 0; + } + @Override StorageException onResponse(BidiWriteObjectResponse response) { //noinspection DataFlowIssue @@ -2200,11 +2210,19 @@ public void recordError( public void onResponse_resetsRetryContextToEnsureRetriesArePossibleForLongWrites() { AtomicBoolean resetCalled = new AtomicBoolean(false); AtomicBoolean recordErrorCalled = new AtomicBoolean(false); + AtomicLong confirmedBytes = new AtomicLong(0); StreamingResponseObserver obs = new StreamingResponseObserver( new BidiUploadState(name.getMethodName()) { + @Override + long getConfirmedBytes() { + return confirmedBytes.get(); + } + @Override StorageException onResponse(BidiWriteObjectResponse response) { + // the server confirmed new bytes + confirmedBytes.set(10); return null; } }, @@ -2229,11 +2247,60 @@ public void recordError( RetryContextTest.failOnFailure()); obs.onStart(TestUtils.nullStreamController()); - obs.onResponse(resourceWithSize(0)); + obs.onResponse(resourceWithSize(10)); assertThat(resetCalled.get()).isTrue(); assertThat(recordErrorCalled.get()).isFalse(); } + + /** + * After a reconnect the state lookup reply repeats the previously confirmed offset. That is not + * progress, and must not refresh the retry budget; otherwise a persistent retryable error + * (e.g. 429, 408) is retried forever and callers hang. + */ + @Test + public void onResponse_doesNotResetRetryContextWhenNoNewBytesAreConfirmed() { + AtomicBoolean resetCalled = new AtomicBoolean(false); + AtomicBoolean recordErrorCalled = new AtomicBoolean(false); + StreamingResponseObserver obs = + new StreamingResponseObserver( + new BidiUploadState(name.getMethodName()) { + @Override + long getConfirmedBytes() { + return 10; + } + + @Override + StorageException onResponse(BidiWriteObjectResponse response) { + return null; + } + }, + new RetryContext() { + @Override + public boolean inBackoff() { + return false; + } + + @Override + public void reset() { + resetCalled.set(true); + } + + @Override + public void recordError( + T t, OnSuccess onSuccess, OnFailure onFailure) { + recordErrorCalled.set(true); + } + }, + RetryContextTest.failOnSuccess(), + RetryContextTest.failOnFailure()); + + obs.onStart(TestUtils.nullStreamController()); + obs.onResponse(resourceWithSize(10)); + + assertThat(resetCalled.get()).isFalse(); + assertThat(recordErrorCalled.get()).isFalse(); + } } static BidiWriteObjectRequest flushOffset(long offset) { diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java index e83cbccefdcf..93ea3cd37cc5 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/ITAppendableUploadTest.java @@ -49,13 +49,19 @@ import java.nio.channels.SeekableByteChannel; import java.nio.file.Paths; import java.util.List; +import java.util.Map; import java.util.UUID; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import org.junit.Ignore; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; @RunWith(StorageITRunner.class) @CrossRun( @@ -76,6 +82,150 @@ public final class ITAppendableUploadTest { @Parameter public Param p; + // --- DIAGNOSTIC ONLY (do not submit) --------------------------------------- + // 1. Every 5 minutes, dump ALL threads (incl. gRPC/netty) with the running test's name. + // 2. Log every test's start, end, duration and outcome; failures include the full exception + // (the retry context's exception lists every attempt and previous failure). + // 3. Enable gRPC HTTP/2 frame logging (inbound frames, outbound headers/control frames) for this + // test class's JVM only, so we can see when GCS replies arrive on the wire. + // 4. Log the client config once per backend. + // Storage-level send/recv/retry logs come from the "com.google.cloud.storage.DIAG" logger. + private static final Logger WATCHDOG_LOG = + LoggerFactory.getLogger(ITAppendableUploadTest.class); + private static volatile String currentTest = ""; + private static volatile long currentTestStartNanos = System.nanoTime(); + private static final java.util.Set CONFIG_LOGGED = + java.util.concurrent.ConcurrentHashMap.newKeySet(); + + static { + enableGrpcFrameLogging(); + // One daemon thread per JVM (each IT class runs in its own JVM in CI). + Executors.newSingleThreadScheduledExecutor( + r -> { + Thread t = new Thread(r, "it-watchdog"); + t.setDaemon(true); + return t; + }) + .scheduleAtFixedRate(ITAppendableUploadTest::dumpAllThreads, 5, 5, TimeUnit.MINUTES); + } + + // Not a Timeout rule, so the test body stays on the runner thread. + @Rule + public final TestWatcher watcher = + new TestWatcher() { + @Override + protected void starting(Description d) { + currentTest = d.getDisplayName(); + currentTestStartNanos = System.nanoTime(); + WATCHDOG_LOG.info("TEST START {}", currentTest); + logConfigOnce(); + } + + @Override + protected void succeeded(Description d) { + WATCHDOG_LOG.info("TEST PASSED in {}ms {}", elapsedMs(), d.getDisplayName()); + } + + @Override + protected void failed(Throwable e, Description d) { + WATCHDOG_LOG.warn("TEST FAILED in {}ms {}", elapsedMs(), d.getDisplayName(), e); + } + + @Override + protected void skipped( + org.junit.AssumptionViolatedException e, Description d) { + WATCHDOG_LOG.info( + "TEST SKIPPED in {}ms {} ({})", elapsedMs(), d.getDisplayName(), e.getMessage()); + } + }; + + private static long elapsedMs() { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - currentTestStartNanos); + } + + private void logConfigOnce() { + try { + if (storage != null && backend != null && CONFIG_LOGGED.add(backend)) { + StorageOptions o = storage.getOptions(); + WATCHDOG_LOG.info( + "CONFIG backend={} optionsClass={} host={} projectId={} credentials={} retry={}" + + " bucket={} location={} placement={}", + backend, + o.getClass().getSimpleName(), + o.getHost(), + o.getProjectId(), + o.getCredentials() == null ? "null" : o.getCredentials().getClass().getName(), + o.getRetrySettings(), + bucket == null ? "null" : bucket.getName(), + bucket == null ? "null" : bucket.getLocation(), + bucket == null ? "null" : bucket.getCustomPlacementConfig()); + } + } catch (Throwable t) { + WATCHDOG_LOG.warn("CONFIG logging failed", t); + } + } + + private static void enableGrpcFrameLogging() { + try { + ch.qos.logback.classic.LoggerContext ctx = + (ch.qos.logback.classic.LoggerContext) LoggerFactory.getILoggerFactory(); + ctx.getLogger("io.grpc.netty.shaded.io.grpc.netty.NettyClientHandler") + .setLevel(ch.qos.logback.classic.Level.DEBUG); + // Drop the high-volume frames: outbound DATA (upload bytes; already summarized by the DIAG + // SEND logs) and WINDOW_UPDATE. Keep inbound DATA (GCS replies), HEADERS (stream start / + // grpc-status trailers), RST_STREAM, GOAWAY, PING, SETTINGS. + ch.qos.logback.classic.turbo.TurboFilter frameFilter = + new ch.qos.logback.classic.turbo.TurboFilter() { + @Override + public ch.qos.logback.core.spi.FilterReply decide( + org.slf4j.Marker marker, + ch.qos.logback.classic.Logger logger, + ch.qos.logback.classic.Level level, + String format, + Object[] params, + Throwable t) { + if (logger != null + && format != null + && logger.getName().endsWith("NettyClientHandler")) { + boolean outbound = false; + if (params != null) { + for (Object p : params) { + if ("OUTBOUND".equals(String.valueOf(p))) { + outbound = true; + break; + } + } + } + if (format.contains("WINDOW_UPDATE") || (outbound && format.contains("DATA:"))) { + return ch.qos.logback.core.spi.FilterReply.DENY; + } + } + return ch.qos.logback.core.spi.FilterReply.NEUTRAL; + } + }; + frameFilter.setContext(ctx); + frameFilter.start(); + ctx.addTurboFilter(frameFilter); + WATCHDOG_LOG.info("gRPC HTTP/2 frame logging enabled for this JVM"); + } catch (Throwable t) { + WATCHDOG_LOG.warn("Unable to enable gRPC frame logging", t); + } + } + + private static void dumpAllThreads() { + WATCHDOG_LOG.warn( + "WATCHDOG current test={} running for {}s", currentTest, elapsedMs() / 1000); + for (Map.Entry e : Thread.getAllStackTraces().entrySet()) { + StringBuilder sb = new StringBuilder(); + for (StackTraceElement el : e.getValue()) { + sb.append("\n at ").append(el); + } + WATCHDOG_LOG.warn( + "WATCHDOG thread {} ({}):{}", e.getKey().getName(), e.getKey().getState(), sb); + } + } + // --------------------------------------------------------------------------- + @Test public void appendableUpload_emptyObject() throws IOException, ExecutionException, InterruptedException, TimeoutException { @@ -103,7 +253,6 @@ public void appendableUpload_emptyObject() assertThat(xxd(actualBytes)).isEqualTo(xxd(new byte[0])); } - @Ignore("Flaky test: b/574588650") @Test public void appendableUpload_bytes() throws IOException, ExecutionException, InterruptedException, TimeoutException { @@ -136,7 +285,6 @@ public void appendableUpload_bytes() assertThat(xxd(actualBytes)).isEqualTo(xxd(a1_a2.getBytes())); } - @Ignore("Flaky test: b/574588650") @Test public void explicitFlush() throws IOException, ExecutionException, InterruptedException, TimeoutException { diff --git a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java index bfe9a5768754..86ddcaccc72e 100644 --- a/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java +++ b/java-storage/google-cloud-storage/src/test/java/com/google/cloud/storage/it/runner/registry/BackendResources.java @@ -262,7 +262,7 @@ static BackendResources of( .setLocation("us-central1") .setCustomPlacementConfig( CustomPlacementConfig.newBuilder() - .setDataLocations(ImmutableList.of("us-central1-c")) + .setDataLocations(ImmutableList.of("us-central1-a")) .build()) .setStorageClass(StorageClass.valueOf("RAPID")) .setHierarchicalNamespace(