feat(inkless:storage): bound and shorten the GCS upload path - #783

* feat(inkless:storage): bound each GCS request attempt by default

GCS requests were bounded only by the google-cloud client defaults (20s connect, 20s read), and neither was configurable. GCS can stall for several seconds on response time-to-first-byte even on an established connection, with no retry and no TLS handshake involved, and with produce.max.upload.attempts=3 a single stalled upload could hold an uploader thread for a minute. That shows up as FileUploadTime maxima and cascades into produce latency through FileCommitWaitTime.

Add gcs.connect.timeout and gcs.read.timeout, mirroring the existing s3.api.call.timeout and s3.api.call.attempt.timeout knobs, and default them to 2s and 1s. Connect keeps the wider bound on purpose: it only applies on a connection-pool miss, so a generous ceiling costs nothing in steady state and leaves headroom for the connection churn that accompanies a slowdown, while the read bound applies to every stalled request and is the one worth keeping tight.

The read value bounds each socket read of a response rather than a whole attempt, so connection setup falls under gcs.connect.timeout, request-body writes are not covered, and a response arriving in several reads can exceed it. An upload response is one small object, so in practice the attempt ends on the first stalled read.

Whether a timed out read is retried depends on the operation. An object insert carries no IF_GENERATION_MATCH precondition, so the client classifies it non-idempotent and its non-idempotent handler retries nothing (HttpRetryAlgorithmManager.getForObjectsCreate, DefaultStorageRetryStrategy, and BaseServiceException.isRetryable, which gates every retryable IOException on idempotency). An upload's total is therefore produce.max.upload.attempts stalled reads: about 3s, down from about 60s. A read is idempotent and the client does retry it, under its default settings of 6 attempts and a 50s total timeout, so a stalled fetch is bounded by that budget rather than by this value alone.

Batch deletes need the read bound applied differently. The client splits a delete into HTTP batches of at most 100 sub-requests and the server answers only once it has processed all of them, so the response wait grows with the batch and a batch is not retried by the client either. Deleting in chunks of half that maximum keeps the wait clear of the timeout, and returning the keys confirmed so far means a slow batch costs one chunk rather than a whole cleaner cycle. FileCleaner already treats deletion as partial, so this is the contract it expects and matches what the S3 backend does.

Timeouts surface as StorageBackendTimeoutException from upload, single-key delete, and a fetch's metadata request. A fetch's body is streamed from the channel it returns and read after the method exits, so a stall mid-download reaches the caller as a plain IOException instead; ReadableByteChannel.read cannot throw StorageBackendException, which does not extend IOException. S3Storage covers its whole fetch only because it materializes the response into a buffer. The batch delete stops the pass rather than classifying, as S3 does. FileUploadJob's existing timeout handling therefore applies to GCS as it already does to S3. Exhausting the attempts fails the file, and every produce request in it gets KAFKA_STORAGE_ERROR, which is retriable at the client.

Testing
-------

GcsStorageConfigTest covers the new defaults, that both values parse into Durations, and that a non-positive value is rejected. GcsErrorHandlingTest drives a delayed response through WireMock and asserts an upload surfaces StorageBackendTimeoutException with SocketTimeoutException as its root cause, which pins the cause-chain walk the classification depends on. GcsStorageTest covers a delete spanning several batches, which the shared suite's ten-key case does not reach.

* feat(inkless:storage): upload GCS objects in a single request

storage.createFrom always opens a resumable session: a POST to create it, then a PUT to upload and finalize. Both round trips are paid on every file and either one can stall. Measured in isolation on ~2 KiB objects, that fixed cost is ~100ms per upload on GCS against ~25ms for the equivalent single PUT on S3, and it does not vary with payload size.

Use storage.create for objects that fit in one array. It sends a single request, and it has the client supply MD5 and CRC32C so the server verifies content rather than only the length checked here. The hashing costs a few milliseconds per megabyte, well below the round trip it replaces.

The threshold is 1 MiB because the trade inverts with size. A single request saves a fixed round trip but cannot stream, so its cost per megabyte is higher, and the measured break-even is near 2 MiB. Raising the threshold without re-measuring makes large uploads slower, not faster. Objects above it keep the resumable path, which also keeps an arbitrarily large one from being buffered whole.

The single-request API takes a byte array, so the direct path copies the payload once more than streaming it would, bounded by the threshold.

createFrom read the stream to EOF, so the existing size check caught a stream that did not match its declared length in either direction. Reading a fixed count drops that, so the upload now probes for a trailing byte the way S3Storage already does, and a stream longer than its length is reported instead of being silently truncated.

Add an object-upload counter: resumable-upload-initiate and resumable-chunk-upload stop advancing for direct uploads, so without it per-request upload activity would no longer be visible. The upload-path interceptor also stops assuming the uploadType parameter is present, which it is not for a direct upload.

Testing
-------

GcsStorageTest overrides testUploadOversizeStream, since the surplus is now reported by the probe rather than by the size check, the way S3StorageTest already does.

* amend! feat(inkless:storage): bound each GCS request attempt by default

feat(inkless:storage): bound each GCS request attempt by default

GCS requests were bounded only by the google-cloud client defaults (20s connect, 20s read), and neither was configurable. GCS can stall for several seconds on response time-to-first-byte even on an established connection, with no retry and no TLS handshake involved, and a single stalled upload could hold an uploader thread for minutes. That shows up as FileUploadTime maxima and cascades into produce latency through FileCommitWaitTime.

Add gcs.connect.timeout and gcs.read.timeout, mirroring the existing s3.api.call.timeout and s3.api.call.attempt.timeout knobs, and default them to 2s and 1s. Connect keeps the wider bound on purpose: it only applies on a connection-pool miss, so a generous ceiling costs nothing in steady state and leaves headroom for the connection churn that accompanies a slowdown, while the read bound applies to every stalled request and is the one worth keeping tight.

The read value bounds each socket read of a response rather than a whole attempt, so connection setup falls under gcs.connect.timeout, request-body writes are not covered, and a response arriving in several reads can exceed it.

Whether a timed out read ends the operation depends on how the client classifies the request, and for an upload the answer is not the one the shape of the code suggests. Creating a resumable session carries no IF_GENERATION_MATCH precondition, so the client classifies it non-idempotent and its non-idempotent handler retries nothing (HttpRetryAlgorithmManager.getForResumableUploadSessionCreate, DefaultStorageRetryStrategy, and BaseServiceException.isRetryable, which gates every retryable IOException on idempotency). Writing to a session that is already active is classified idempotent whether or not the session started that way (getForResumableUploadSessionWrite), so the PUT is retried under the client's default settings of 6 attempts and a 50s total timeout. Since createFrom buffers 15 MiB by default, a file below that size is one such PUT, and its ceiling is that budget rather than produce.max.upload.attempts stalled reads. Lowering this value changes how many attempts fit inside the 50s, not how long the upload can take. A read is idempotent for the same reason and carries the same budget. What the bound buys on every path is that a stalled request fails its attempt in a bounded, operator-set time instead of holding a connection for 20s.

Batch deletes need the read bound applied differently. The client splits a delete into HTTP batches of at most 100 sub-requests and the server answers only once it has processed all of them, so the response wait grows with the batch and a batch is not retried by the client either. Deleting in chunks of half that maximum keeps the wait clear of the timeout, and returning the keys confirmed so far means a slow batch costs one chunk rather than a whole cleaner cycle. FileCleaner already treats deletion as partial, so this is the contract it expects and matches what the S3 backend does.

Chunking alone would still confirm the wrong keys. Storage.delete(Iterable) reports a failed sub-request as the same false its callback uses for an absent blob (StorageImpl.delete, whose per-blob error callback records Boolean.FALSE without rethrowing), so the return value cannot distinguish them and ignoring it confirmed every key in the batch. A 403 or 5xx on one key then dereferenced a live file in the control plane and left an orphan nothing would collect. Driving Storage.batch() directly keeps the three outcomes apart, because StorageBatch maps a 404 to false and anything else to a thrown StorageException: deleted and already-absent are confirmed, a failure is not and is counted by response code, at INFO for throttling and WARN otherwise as in the S3 backend. object-delete now counts confirmed deletions rather than attempted ones.

Timeouts surface as StorageBackendTimeoutException from upload, single-key delete, and a fetch's metadata request. A fetch's body is streamed from the channel it returns and read after the method exits, so a stall mid-download reaches the caller as a plain IOException instead; ReadableByteChannel.read cannot throw StorageBackendException, which does not extend IOException. S3Storage covers its whole fetch only because it materializes the response into a buffer. The batch delete stops the pass rather than classifying, as S3 does. FileUploadJob's existing timeout handling therefore applies to GCS as it already does to S3. Exhausting the attempts fails the file, and every produce request in it gets KAFKA_STORAGE_ERROR, which is retriable at the client.

Testing
-------

GcsStorageConfigTest covers the new defaults, that both values parse into Durations, and that a non-positive value is rejected. GcsStorageTest covers a delete spanning several batches, which the shared suite's ten-key case does not reach.

GcsErrorHandlingTest uses WireMock for the failures fake-gcs-server cannot produce. A delayed response pins the timeout classification: an upload surfaces StorageBackendTimeoutException with SocketTimeoutException as its root cause, which exercises the cause-chain walk every classification site depends on. A hand-built multipart batch response pins the delete confirmation on both sides: with one deleted sub-response and two refused, only one key comes back confirmed, and with every sub-response a 404, both keys do.

* feat(inkless:storage): count GCS error responses by type

The GCS backend reported how many requests it made and nothing about how many failed. The S3 backend counts failed attempts by type, and those counters are what let an operator tell a throttled or faulting provider from a slow one. Without them on GCS, a bucket that starts refusing requests shows up only as latency and as FileCleaner drain arithmetic.

Add throttling-errors, server-errors, and other-errors to gcs-client-metrics under the S3 backend's names, so dashboards built for one backend apply to the other. Each counts per request attempt, so a failure the client retries is counted once per attempt, as the AWS SDK does for S3. The response interceptor the request counters already use is the hook: it runs once per attempt for every request the client builds with its initializer, including the resumable upload's PUT and the object download's GET, and it sees non-2xx responses as well as successes. 429 and 503 count as throttling, as SlowDown does for S3, since GCS returns 503 under load and asks the client to back off; other 5xx are server errors; other non-2xx, such as the 404 for a missing key, are other errors.

The batch request is built without the request initializer (AbstractGoogleClient.batch passes null), so the interceptor never sees it. GcsStorage.delete(Set) reports its failure directly; a batch is one request and one attempt, so the count is exact there too. The comment on recordBatchDeleteObjects said the interceptor saw the batch request but could not identify its operations, which was the wrong half of the explanation.

Requests that fail without a response, a configured timeout or a connection reset, are not counted. The AWS SDK classifies and publishes those itself; the GCS client offers no once-per-attempt hook that carries the exception. The response interceptor never runs on an IOException, HttpIOExceptionHandler receives the request but not the exception, the transport cannot be wrapped because NetHttpTransport is final, and the retry strategy is consulted up to three times per attempt by gax, so counting there needs a dedupe that does not belong in a metrics collector. configured-timeout-errors and io-errors are left for a change that settles where that hook should live.

Testing
-------

GcsErrorHandlingTest asserts each sensor end to end through WireMock, which pins the interceptor against the client's real call pattern: a 429 on upload counts one throttling error, a 500 one server error, a 404 on fetch one other error, and a 503 on the batch endpoint one throttling error through the caller-side path. Every assertion is an exact 1.0, so a client upgrade that starts retrying at the HTTP layer fails the suite instead of skewing the counters.

* feat(inkless:storage): bound each GCS request attempt by default

GCS requests were bounded only by the google-cloud client defaults (20s connect, 20s read), and neither was configurable. GCS can stall for several seconds on response time-to-first-byte even on an established connection, with no retry and no TLS handshake involved, and with produce.max.upload.attempts=3 a single stalled upload could hold an uploader thread for a minute. That shows up as FileUploadTime maxima and cascades into produce latency through FileCommitWaitTime.

Add gcs.connect.timeout and gcs.read.timeout, mirroring the existing s3.api.call.timeout and s3.api.call.attempt.timeout knobs, and default them to 2s and 1s. Connect keeps the wider bound on purpose: it only applies on a connection-pool miss, so a generous ceiling costs nothing in steady state and leaves headroom for the connection churn that accompanies a slowdown, while the read bound applies to every stalled request and is the one worth keeping tight.

The read value bounds each socket read of a response rather than a whole attempt, so connection setup falls under gcs.connect.timeout, request-body writes are not covered, and a response arriving in several reads can exceed it. An upload response is one small object, so in practice the attempt ends on the first stalled read.

Whether a timed out read is retried depends on the operation. An object insert carries no IF_GENERATION_MATCH precondition, so the client classifies it non-idempotent and its non-idempotent handler retries nothing (HttpRetryAlgorithmManager.getForObjectsCreate, DefaultStorageRetryStrategy, and BaseServiceException.isRetryable, which gates every retryable IOException on idempotency). An upload's total is therefore produce.max.upload.attempts stalled reads: about 3s, down from about 60s. A read is idempotent and the client does retry it, under its default settings of 6 attempts and a 50s total timeout, so a stalled fetch is bounded by that budget rather than by this value alone.

Batch deletes need the read bound applied differently. The client splits a delete into HTTP batches of at most 100 sub-requests and the server answers only once it has processed all of them, so the response wait grows with the batch and a batch is not retried by the client either. Deleting in chunks of half that maximum keeps the wait clear of the timeout, and returning the keys confirmed so far means a slow batch costs one chunk rather than a whole cleaner cycle. FileCleaner already treats deletion as partial, so this is the contract it expects and matches what the S3 backend does.

Timeouts surface as StorageBackendTimeoutException from upload, single-key delete, and a fetch's metadata request. A fetch's body is streamed from the channel it returns and read after the method exits, so a stall mid-download reaches the caller as a plain IOException instead; ReadableByteChannel.read cannot throw StorageBackendException, which does not extend IOException. S3Storage covers its whole fetch only because it materializes the response into a buffer. The batch delete stops the pass rather than classifying, as S3 does. FileUploadJob's existing timeout handling therefore applies to GCS as it already does to S3. Exhausting the attempts fails the file, and every produce request in it gets KAFKA_STORAGE_ERROR, which is retriable at the client.

Testing
-------

GcsStorageConfigTest covers the new defaults, that both values parse into Durations, and that a non-positive value is rejected. GcsErrorHandlingTest drives a delayed response through WireMock and asserts an upload surfaces StorageBackendTimeoutException with SocketTimeoutException as its root cause, which pins the cause-chain walk the classification depends on. GcsStorageTest covers a delete spanning several batches, which the shared suite's ten-key case does not reach.

* feat(inkless:storage): upload GCS objects in a single request

storage.createFrom always opens a resumable session: a POST to create it, then a PUT to upload and finalize. Both round trips are paid on every file and either one can stall. Measured in isolation on ~2 KiB objects, that fixed cost is ~100ms per upload on GCS against ~25ms for the equivalent single PUT on S3, and it does not vary with payload size.

Use storage.create for objects that fit in one array. It sends a single request, and it has the client supply MD5 and CRC32C so the server verifies content rather than only the length checked here. The hashing costs a few milliseconds per megabyte, well below the round trip it replaces.

The threshold is 1 MiB because the trade inverts with size. A single request saves a fixed round trip but cannot stream, so its cost per megabyte is higher, and the measured break-even is near 2 MiB. Raising the threshold without re-measuring makes large uploads slower, not faster. Objects above it keep the resumable path, which also keeps an arbitrarily large one from being buffered whole.

The single-request API takes a byte array, so the direct path copies the payload once more than streaming it would, bounded by the threshold.

createFrom read the stream to EOF, so the existing size check caught a stream that did not match its declared length in either direction. Reading a fixed count drops that, so the upload now probes for a trailing byte the way S3Storage already does, and a stream longer than its length is reported instead of being silently truncated.

Add an object-upload counter: resumable-upload-initiate and resumable-chunk-upload stop advancing for direct uploads, so without it per-request upload activity would no longer be visible. The upload-path interceptor also stops assuming the uploadType parameter is present, which it is not for a direct upload.

Testing
-------

GcsStorageTest overrides testUploadOversizeStream, since the surplus is now reported by the probe rather than by the size check, the way S3StorageTest already does.

* amend! feat(inkless:storage): bound each GCS request attempt by default

feat(inkless:storage): bound each GCS request attempt by default

GCS requests were bounded only by the google-cloud client defaults (20s connect, 20s read), and neither was configurable. GCS can stall for several seconds on response time-to-first-byte even on an established connection, with no retry and no TLS handshake involved, and a single stalled upload could hold an uploader thread for minutes. That shows up as FileUploadTime maxima and cascades into produce latency through FileCommitWaitTime.

Add gcs.connect.timeout and gcs.read.timeout, mirroring the existing s3.api.call.timeout and s3.api.call.attempt.timeout knobs, and default them to 2s and 1s. Connect keeps the wider bound on purpose: it only applies on a connection-pool miss, so a generous ceiling costs nothing in steady state and leaves headroom for the connection churn that accompanies a slowdown, while the read bound applies to every stalled request and is the one worth keeping tight.

The read value bounds each socket read of a response rather than a whole attempt, so connection setup falls under gcs.connect.timeout, request-body writes are not covered, and a response arriving in several reads can exceed it.

Whether a timed out read ends the operation depends on how the client classifies the request, and for an upload the answer is not the one the shape of the code suggests. Creating a resumable session carries no IF_GENERATION_MATCH precondition, so the client classifies it non-idempotent and its non-idempotent handler retries nothing (HttpRetryAlgorithmManager.getForResumableUploadSessionCreate, DefaultStorageRetryStrategy, and BaseServiceException.isRetryable, which gates every retryable IOException on idempotency). Writing to a session that is already active is classified idempotent whether or not the session started that way (getForResumableUploadSessionWrite), so the PUT is retried under the client's default settings of 6 attempts and a 50s total timeout. Since createFrom buffers 15 MiB by default, a file below that size is one such PUT, and its ceiling is that budget rather than produce.max.upload.attempts stalled reads. Lowering this value changes how many attempts fit inside the 50s, not how long the upload can take. A read is idempotent for the same reason and carries the same budget. What the bound buys on every path is that a stalled request fails its attempt in a bounded, operator-set time instead of holding a connection for 20s.

Batch deletes need the read bound applied differently. The client splits a delete into HTTP batches of at most 100 sub-requests and the server answers only once it has processed all of them, so the response wait grows with the batch and a batch is not retried by the client either. Deleting in chunks of half that maximum keeps the wait clear of the timeout, and returning the keys confirmed so far means a slow batch costs one chunk rather than a whole cleaner cycle. FileCleaner already treats deletion as partial, so this is the contract it expects and matches what the S3 backend does.

Chunking alone would still confirm the wrong keys. Storage.delete(Iterable) reports a failed sub-request as the same false its callback uses for an absent blob (StorageImpl.delete, whose per-blob error callback records Boolean.FALSE without rethrowing), so the return value cannot distinguish them and ignoring it confirmed every key in the batch. A 403 or 5xx on one key then dereferenced a live file in the control plane and left an orphan nothing would collect. Driving Storage.batch() directly keeps the three outcomes apart, because StorageBatch maps a 404 to false and anything else to a thrown StorageException: deleted and already-absent are confirmed, a failure is not and is counted by response code, at INFO for throttling and WARN otherwise as in the S3 backend. object-delete now counts confirmed deletions rather than attempted ones.

Timeouts surface as StorageBackendTimeoutException from upload, single-key delete, and a fetch's metadata request. A fetch's body is streamed from the channel it returns and read after the method exits, so a stall mid-download reaches the caller as a plain IOException instead; ReadableByteChannel.read cannot throw StorageBackendException, which does not extend IOException. S3Storage covers its whole fetch only because it materializes the response into a buffer. The batch delete stops the pass rather than classifying, as S3 does. FileUploadJob's existing timeout handling therefore applies to GCS as it already does to S3. Exhausting the attempts fails the file, and every produce request in it gets KAFKA_STORAGE_ERROR, which is retriable at the client.

Testing
-------

GcsStorageConfigTest covers the new defaults, that both values parse into Durations, and that a non-positive value is rejected. GcsStorageTest covers a delete spanning several batches, which the shared suite's ten-key case does not reach.

GcsErrorHandlingTest uses WireMock for the failures fake-gcs-server cannot produce. A delayed response pins the timeout classification: an upload surfaces StorageBackendTimeoutException with SocketTimeoutException as its root cause, which exercises the cause-chain walk every classification site depends on. A hand-built multipart batch response pins the delete confirmation on both sides: with one deleted sub-response and two refused, only one key comes back confirmed, and with every sub-response a 404, both keys do.

* feat(inkless:storage): count GCS error responses by type

The GCS backend reported how many requests it made and nothing about how many failed. The S3 backend counts failed attempts by type, and those counters are what let an operator tell a throttled or faulting provider from a slow one. Without them on GCS, a bucket that starts refusing requests shows up only as latency and as FileCleaner drain arithmetic.

Add throttling-errors, server-errors, and other-errors to gcs-client-metrics under the S3 backend's names, so dashboards built for one backend apply to the other. Each counts per request attempt, so a failure the client retries is counted once per attempt, as the AWS SDK does for S3. The response interceptor the request counters already use is the hook: it runs once per attempt for every request the client builds with its initializer, including the resumable upload's PUT and the object download's GET, and it sees non-2xx responses as well as successes. 429 and 503 count as throttling, as SlowDown does for S3, since GCS returns 503 under load and asks the client to back off; other 5xx are server errors; other non-2xx, such as the 404 for a missing key, are other errors.

The batch request is built without the request initializer (AbstractGoogleClient.batch passes null), so the interceptor never sees it. GcsStorage.delete(Set) reports its failure directly; a batch is one request and one attempt, so the count is exact there too. The comment on recordBatchDeleteObjects said the interceptor saw the batch request but could not identify its operations, which was the wrong half of the explanation.

Requests that fail without a response, a configured timeout or a connection reset, are not counted. The AWS SDK classifies and publishes those itself; the GCS client offers no once-per-attempt hook that carries the exception. The response interceptor never runs on an IOException, HttpIOExceptionHandler receives the request but not the exception, the transport cannot be wrapped because NetHttpTransport is final, and the retry strategy is consulted up to three times per attempt by gax, so counting there needs a dedupe that does not belong in a metrics collector. configured-timeout-errors and io-errors are left for a change that settles where that hook should live.

Testing
-------

GcsErrorHandlingTest asserts each sensor end to end through WireMock, which pins the interceptor against the client's real call pattern: a 429 on upload counts one throttling error, a 500 one server error, a 404 on fetch one other error, and a 503 on the batch endpoint one throttling error through the caller-side path. Every assertion is an exact 1.0, so a client upgrade that starts retrying at the HTTP layer fails the suite instead of skewing the counters.

* feat(inkless:storage): bound each GCS request attempt by default

GCS requests were bounded only by the google-cloud client defaults (20s connect, 20s read), and neither was configurable. GCS can stall for several seconds on response time-to-first-byte even on an established connection, with no retry and no TLS handshake involved, and with produce.max.upload.attempts=3 a single stalled upload could hold an uploader thread for a minute. That shows up as FileUploadTime maxima and cascades into produce latency through FileCommitWaitTime.

Add gcs.connect.timeout and gcs.read.timeout, mirroring the existing s3.api.call.timeout and s3.api.call.attempt.timeout knobs, and default them to 2s and 1s. Connect keeps the wider bound on purpose: it only applies on a connection-pool miss, so a generous ceiling costs nothing in steady state and leaves headroom for the connection churn that accompanies a slowdown, while the read bound applies to every stalled request and is the one worth keeping tight.

The read value bounds each socket read of a response rather than a whole attempt, so connection setup falls under gcs.connect.timeout, request-body writes are not covered, and a response arriving in several reads can exceed it. An upload response is one small object, so in practice the attempt ends on the first stalled read.

Whether a timed out read is retried depends on the operation. An object insert carries no IF_GENERATION_MATCH precondition, so the client classifies it non-idempotent and its non-idempotent handler retries nothing (HttpRetryAlgorithmManager.getForObjectsCreate, DefaultStorageRetryStrategy, and BaseServiceException.isRetryable, which gates every retryable IOException on idempotency). An upload's total is therefore produce.max.upload.attempts stalled reads: about 3s, down from about 60s. A read is idempotent and the client does retry it, under its default settings of 6 attempts and a 50s total timeout, so a stalled fetch is bounded by that budget rather than by this value alone.

Batch deletes need the read bound applied differently. The client splits a delete into HTTP batches of at most 100 sub-requests and the server answers only once it has processed all of them, so the response wait grows with the batch and a batch is not retried by the client either. Deleting in chunks of half that maximum keeps the wait clear of the timeout, and returning the keys confirmed so far means a slow batch costs one chunk rather than a whole cleaner cycle. FileCleaner already treats deletion as partial, so this is the contract it expects and matches what the S3 backend does.

Timeouts surface as StorageBackendTimeoutException from upload, single-key delete, and a fetch's metadata request. A fetch's body is streamed from the channel it returns and read after the method exits, so a stall mid-download reaches the caller as a plain IOException instead; ReadableByteChannel.read cannot throw StorageBackendException, which does not extend IOException. S3Storage covers its whole fetch only because it materializes the response into a buffer. The batch delete stops the pass rather than classifying, as S3 does. FileUploadJob's existing timeout handling therefore applies to GCS as it already does to S3. Exhausting the attempts fails the file, and every produce request in it gets KAFKA_STORAGE_ERROR, which is retriable at the client.

Testing
-------

GcsStorageConfigTest covers the new defaults, that both values parse into Durations, and that a non-positive value is rejected. GcsErrorHandlingTest drives a delayed response through WireMock and asserts an upload surfaces StorageBackendTimeoutException with SocketTimeoutException as its root cause, which pins the cause-chain walk the classification depends on. GcsStorageTest covers a delete spanning several batches, which the shared suite's ten-key case does not reach.

* feat(inkless:storage): upload GCS objects in a single request

storage.createFrom always opens a resumable session: a POST to create it, then a PUT to upload and finalize. Both round trips are paid on every file and either one can stall. Measured in isolation on ~2 KiB objects, that fixed cost is ~100ms per upload on GCS against ~25ms for the equivalent single PUT on S3, and it does not vary with payload size.

Use storage.create for objects that fit in one array. It sends a single request, and it has the client supply MD5 and CRC32C so the server verifies content rather than only the length checked here. The hashing costs a few milliseconds per megabyte, well below the round trip it replaces.

The threshold is 1 MiB because the trade inverts with size. A single request saves a fixed round trip but cannot stream, so its cost per megabyte is higher, and the measured break-even is near 2 MiB. Raising the threshold without re-measuring makes large uploads slower, not faster. Objects above it keep the resumable path, which also keeps an arbitrarily large one from being buffered whole.

The single-request API takes a byte array, so the direct path copies the payload once more than streaming it would, bounded by the threshold.

createFrom read the stream to EOF, so the existing size check caught a stream that did not match its declared length in either direction. Reading a fixed count drops that, so the upload now probes for a trailing byte the way S3Storage already does, and a stream longer than its length is reported instead of being silently truncated.

Add an object-upload counter: resumable-upload-initiate and resumable-chunk-upload stop advancing for direct uploads, so without it per-request upload activity would no longer be visible. The upload-path interceptor also stops assuming the uploadType parameter is present, which it is not for a direct upload.

Testing
-------

GcsStorageTest overrides testUploadOversizeStream, since the surplus is now reported by the probe rather than by the size check, the way S3StorageTest already does.

* amend! feat(inkless:storage): bound each GCS request attempt by default

feat(inkless:storage): bound each GCS request attempt by default

GCS requests were bounded only by the google-cloud client defaults (20s connect, 20s read), and neither was configurable. GCS can stall for several seconds on response time-to-first-byte even on an established connection, with no retry and no TLS handshake involved, and a single stalled upload could hold an uploader thread for minutes. That shows up as FileUploadTime maxima and cascades into produce latency through FileCommitWaitTime.

Add gcs.connect.timeout and gcs.read.timeout, mirroring the existing s3.api.call.timeout and s3.api.call.attempt.timeout knobs, and default them to 2s and 1s. Connect keeps the wider bound on purpose: it only applies on a connection-pool miss, so a generous ceiling costs nothing in steady state and leaves headroom for the connection churn that accompanies a slowdown, while the read bound applies to every stalled request and is the one worth keeping tight.

The read value bounds each socket read of a response rather than a whole attempt, so connection setup falls under gcs.connect.timeout, request-body writes are not covered, and a response arriving in several reads can exceed it.

Whether a timed out read ends the operation depends on how the client classifies the request, and for an upload the answer is not the one the shape of the code suggests. Creating a resumable session carries no IF_GENERATION_MATCH precondition, so the client classifies it non-idempotent and its non-idempotent handler retries nothing (HttpRetryAlgorithmManager.getForResumableUploadSessionCreate, DefaultStorageRetryStrategy, and BaseServiceException.isRetryable, which gates every retryable IOException on idempotency). Writing to a session that is already active is classified idempotent whether or not the session started that way (getForResumableUploadSessionWrite), so the PUT is retried under the client's default settings of 6 attempts and a 50s total timeout. Since createFrom buffers 15 MiB by default, a file below that size is one such PUT, and its ceiling is that budget rather than produce.max.upload.attempts stalled reads. Lowering this value changes how many attempts fit inside the 50s, not how long the upload can take. A read is idempotent for the same reason and carries the same budget. What the bound buys on every path is that a stalled request fails its attempt in a bounded, operator-set time instead of holding a connection for 20s.

Batch deletes need the read bound applied differently. The client splits a delete into HTTP batches of at most 100 sub-requests and the server answers only once it has processed all of them, so the response wait grows with the batch and a batch is not retried by the client either. Deleting in chunks of half that maximum keeps the wait clear of the timeout, and returning the keys confirmed so far means a slow batch costs one chunk rather than a whole cleaner cycle. FileCleaner already treats deletion as partial, so this is the contract it expects and matches what the S3 backend does.

Chunking alone would still confirm the wrong keys. Storage.delete(Iterable) reports a failed sub-request as the same false its callback uses for an absent blob (StorageImpl.delete, whose per-blob error callback records Boolean.FALSE without rethrowing), so the return value cannot distinguish them and ignoring it confirmed every key in the batch. A 403 or 5xx on one key then dereferenced a live file in the control plane and left an orphan nothing would collect. Driving Storage.batch() directly keeps the three outcomes apart, because StorageBatch maps a 404 to false and anything else to a thrown StorageException: deleted and already-absent are confirmed, a failure is not and is counted by response code, at INFO for throttling and WARN otherwise as in the S3 backend. object-delete now counts confirmed deletions rather than attempted ones.

Timeouts surface as StorageBackendTimeoutException from upload, single-key delete, and a fetch's metadata request. A fetch's body is streamed from the channel it returns and read after the method exits, so a stall mid-download reaches the caller as a plain IOException instead; ReadableByteChannel.read cannot throw StorageBackendException, which does not extend IOException. S3Storage covers its whole fetch only because it materializes the response into a buffer. The batch delete stops the pass rather than classifying, as S3 does. FileUploadJob's existing timeout handling therefore applies to GCS as it already does to S3. Exhausting the attempts fails the file, and every produce request in it gets KAFKA_STORAGE_ERROR, which is retriable at the client.

Testing
-------

GcsStorageConfigTest covers the new defaults, that both values parse into Durations, and that a non-positive value is rejected. GcsStorageTest covers a delete spanning several batches, which the shared suite's ten-key case does not reach.

GcsErrorHandlingTest uses WireMock for the failures fake-gcs-server cannot produce. A delayed response pins the timeout classification: an upload surfaces StorageBackendTimeoutException with SocketTimeoutException as its root cause, which exercises the cause-chain walk every classification site depends on. A hand-built multipart batch response pins the delete confirmation on both sides: with one deleted sub-response and two refused, only one key comes back confirmed, and with every sub-response a 404, both keys do.

* feat(inkless:storage): count GCS error responses by type

The GCS backend reported how many requests it made and nothing about how many failed. The S3 backend counts failed attempts by type, and those counters are what let an operator tell a throttled or faulting provider from a slow one. Without them on GCS, a bucket that starts refusing requests shows up only as latency and as FileCleaner drain arithmetic.

Add throttling-errors, server-errors, and other-errors to gcs-client-metrics under the S3 backend's names, so dashboards built for one backend apply to the other. Each counts per request attempt, so a failure the client retries is counted once per attempt, as the AWS SDK does for S3. The response interceptor the request counters already use is the hook: it runs once per attempt for every request the client builds with its initializer, including the resumable upload's PUT and the object download's GET, and it sees non-2xx responses as well as successes. 429 and 503 count as throttling, as SlowDown does for S3, since GCS returns 503 under load and asks the client to back off; other 5xx are server errors; other non-2xx, such as the 404 for a missing key, are other errors.

The batch request is built without the request initializer (AbstractGoogleClient.batch passes null), so the interceptor never sees it. GcsStorage.delete(Set) reports its failure directly; a batch is one request and one attempt, so the count is exact there too. The comment on recordBatchDeleteObjects said the interceptor saw the batch request but could not identify its operations, which was the wrong half of the explanation.

Requests that fail without a response, a configured timeout or a connection reset, are not counted. The AWS SDK classifies and publishes those itself; the GCS client offers no once-per-attempt hook that carries the exception. The response interceptor never runs on an IOException, HttpIOExceptionHandler receives the request but not the exception, the transport cannot be wrapped because NetHttpTransport is final, and the retry strategy is consulted up to three times per attempt by gax, so counting there needs a dedupe that does not belong in a metrics collector. configured-timeout-errors and io-errors are left for a change that settles where that hook should live.

Testing
-------

GcsErrorHandlingTest asserts each sensor end to end through WireMock, which pins the interceptor against the client's real call pattern: a 429 on upload counts one throttling error, a 500 one server error, a 404 on fetch one other error, and a 503 on the batch endpoint one throttling error through the caller-side path. Every assertion is an exact 1.0, so a client upgrade that starts retrying at the HTTP layer fails the suite instead of skewing the counters.