This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/29162-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit c2cdd5513d8a8c47548d2f2d3819fc79a7dba0a4 Author: Guillaume Nodet <[email protected]> AuthorDate: Wed Nov 4 15:34:43 2026 +0000 [backport camel-4.18.x] CAMEL-26586: camel-google-storage - close the upload stream after a failure --- .../google/storage/GoogleCloudStorageProducer.java | 74 +++++------ .../unit/ProducerUploadStreamCloseTest.java | 137 +++++++++++++++++++++ 2 files changed, 175 insertions(+), 36 deletions(-) diff --git a/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageProducer.java b/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageProducer.java index e876cf7cb56b..b4484e48ed5f 100644 --- a/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageProducer.java +++ b/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageProducer.java @@ -120,45 +120,47 @@ public class GoogleCloudStorageProducer extends DefaultProducer { } else { is = exchange.getIn().getMandatoryBody(InputStream.class); } - // Set content length from WrappedFile if available and not already set - if (fileLength > 0 && !objectMetadata.containsKey("Content-Length")) { - objectMetadata.put("Content-Length", String.valueOf(fileLength)); - } - // Handle Content-Length if not already set - is = setContentLength(objectMetadata, is); + try { + // Set content length from WrappedFile if available and not already set + if (fileLength > 0 && !objectMetadata.containsKey("Content-Length")) { + objectMetadata.put("Content-Length", String.valueOf(fileLength)); + } + // Handle Content-Length if not already set + is = setContentLength(objectMetadata, is); - Blob createdBlob; - BlobId blobId = BlobId.of(bucketName, objectName); + Blob createdBlob; + BlobId blobId = BlobId.of(bucketName, objectName); - BlobInfo.Builder builder = BlobInfo.newBuilder(blobId); - String ct = objectMetadata.remove("Content-Type"); - if (ct != null) { - builder.setContentType(ct); - } - String cd = objectMetadata.remove("Content-Disposition"); - if (cd != null) { - builder.setContentDisposition(cd); - } - String ce = objectMetadata.remove("Content-Encoding"); - if (ce != null) { - builder.setContentEncoding(ce); - } - String md5 = objectMetadata.remove("Content-Md5"); - if (md5 != null) { - builder.setMd5(md5); - } - String cc = objectMetadata.remove("Cache-Control"); - if (cc != null) { - builder.setCacheControl(cc); + BlobInfo.Builder builder = BlobInfo.newBuilder(blobId); + String ct = objectMetadata.remove("Content-Type"); + if (ct != null) { + builder.setContentType(ct); + } + String cd = objectMetadata.remove("Content-Disposition"); + if (cd != null) { + builder.setContentDisposition(cd); + } + String ce = objectMetadata.remove("Content-Encoding"); + if (ce != null) { + builder.setContentEncoding(ce); + } + String md5 = objectMetadata.remove("Content-Md5"); + if (md5 != null) { + builder.setMd5(md5); + } + String cc = objectMetadata.remove("Cache-Control"); + if (cc != null) { + builder.setCacheControl(cc); + } + BlobInfo blobInfo = builder.setMetadata(objectMetadata).build(); + // According to documentation, this internally uses a WriteChannel + createdBlob = storage.createFrom(blobInfo, is); + LOG.trace("created createdBlob [{}]", createdBlob); + Message message = getMessageForResponse(exchange); + message.setBody(createdBlob); + } finally { + IOHelper.close(is); } - BlobInfo blobInfo = builder.setMetadata(objectMetadata).build(); - // According to documentation, this internally uses a WriteChannel - createdBlob = storage.createFrom(blobInfo, is); - LOG.trace("created createdBlob [{}]", createdBlob); - Message message = getMessageForResponse(exchange); - message.setBody(createdBlob); - - IOHelper.close(is); } /** diff --git a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/unit/ProducerUploadStreamCloseTest.java b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/unit/ProducerUploadStreamCloseTest.java new file mode 100644 index 000000000000..62bf71ddbd1f --- /dev/null +++ b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/unit/ProducerUploadStreamCloseTest.java @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 org.apache.camel.component.google.storage.unit; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Proxy; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicReference; + +import com.google.cloud.storage.Storage; +import org.apache.camel.Exchange; +import org.apache.camel.component.google.storage.GoogleCloudStorageComponent; +import org.apache.camel.component.google.storage.GoogleCloudStorageConfiguration; +import org.apache.camel.component.google.storage.GoogleCloudStorageConstants; +import org.apache.camel.component.google.storage.GoogleCloudStorageEndpoint; +import org.apache.camel.component.google.storage.GoogleCloudStorageProducer; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.DefaultExchange; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ProducerUploadStreamCloseTest { + + @TempDir + Path directory; + + @Test + void closesTheFileStreamWhenTheUploadFails() throws Exception { + Path file = Files.writeString(directory.resolve("upload.txt"), "payload"); + AtomicReference<InputStream> uploaded = new AtomicReference<>(); + IOException failure = new IOException("Upload failed"); + + try (DefaultCamelContext context = new DefaultCamelContext()) { + GoogleCloudStorageProducer producer = producer(context, uploaded, failure); + Exchange exchange = new DefaultExchange(context); + exchange.getIn().setBody(file.toFile()); + + assertThatThrownBy(() -> producer.process(exchange)).isSameAs(failure); + assertThat(uploaded.get()).isNotNull(); + assertThatThrownBy(() -> uploaded.get().read()).isInstanceOf(IOException.class); + } + } + + @Test + void closesTheBodyStreamWhenTheUploadFails() throws Exception { + TrackingStream stream = new TrackingStream(); + IOException failure = new IOException("Upload failed"); + + try (DefaultCamelContext context = new DefaultCamelContext()) { + GoogleCloudStorageProducer producer = producer(context, new AtomicReference<>(), failure); + Exchange exchange = new DefaultExchange(context); + exchange.getIn().setBody(stream); + exchange.getIn().setHeader(GoogleCloudStorageConstants.CONTENT_LENGTH, 7); + + assertThatThrownBy(() -> producer.process(exchange)).isSameAs(failure); + assertThat(stream.closed).isTrue(); + } + } + + @Test + void stillClosesTheBodyStreamAfterASuccessfulUpload() throws Exception { + TrackingStream stream = new TrackingStream(); + AtomicReference<InputStream> uploaded = new AtomicReference<>(); + + try (DefaultCamelContext context = new DefaultCamelContext()) { + GoogleCloudStorageProducer producer = producer(context, uploaded, null); + Exchange exchange = new DefaultExchange(context); + exchange.getIn().setBody(stream); + exchange.getIn().setHeader(GoogleCloudStorageConstants.CONTENT_LENGTH, 7); + + producer.process(exchange); + + assertThat(uploaded.get()).isSameAs(stream); + assertThat(stream.closed).isTrue(); + } + } + + private static GoogleCloudStorageProducer producer( + DefaultCamelContext context, AtomicReference<InputStream> uploaded, IOException failure) { + Storage storage = (Storage) Proxy.newProxyInstance(Storage.class.getClassLoader(), new Class<?>[] { Storage.class }, + (proxy, method, args) -> { + if ("createFrom".equals(method.getName())) { + uploaded.set((InputStream) args[1]); + if (failure != null) { + throw failure; + } + return null; + } + throw new UnsupportedOperationException(method.getName()); + }); + GoogleCloudStorageConfiguration configuration = new GoogleCloudStorageConfiguration(); + configuration.setBucketName("bucket"); + configuration.setObjectName("object"); + GoogleCloudStorageComponent component = new GoogleCloudStorageComponent(context); + GoogleCloudStorageEndpoint endpoint + = new GoogleCloudStorageEndpoint("google-storage://bucket", component, configuration) { + @Override + public Storage getStorageClient() { + return storage; + } + }; + return new GoogleCloudStorageProducer(endpoint); + } + + private static final class TrackingStream extends ByteArrayInputStream { + private boolean closed; + + private TrackingStream() { + super(new byte[7]); + } + + @Override + public void close() { + closed = true; + } + } +}
