diff --git a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java new file mode 100644 index 0000000000..283630911a --- /dev/null +++ b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/ComposeAppendOutputStream.java @@ -0,0 +1,121 @@ +/* + * 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.hop.vfs.gs; + +import com.google.cloud.WriteChannel; +import com.google.cloud.storage.BlobId; +import com.google.cloud.storage.BlobInfo; +import com.google.cloud.storage.Storage; +import com.google.cloud.storage.Storage.BlobTargetOption; +import com.google.cloud.storage.Storage.ComposeRequest; +import java.io.IOException; +import java.io.OutputStream; +import org.apache.hop.core.logging.LogChannel; + +/** + * Append output stream for Google Cloud Storage, built on composite objects. + * + *

GCS objects are immutable, so there is no true in-place append. Instead the bytes written to + * this stream are streamed into a short-lived temporary object, and on {@link #close()} + * the target and the temporary object are concatenated back onto the target with a single {@code + * compose} call ({@code target = target + temp}). The temporary object is then deleted. + * + *

The compose is guarded with an {@code ifGenerationMatch} precondition captured when the stream + * was opened, so a concurrent modification of the target fails the append fast instead of silently + * dropping data. One open/close cycle performs exactly one compose regardless of how many times + * {@code write(...)} is called, so a transform that appends many rows still costs a single compose. + * + *

Note that composite objects have no MD5 metadata (CRC32C is still maintained and validated by + * the compose operation). + */ +public class ComposeAppendOutputStream extends OutputStream { + + private final Storage storage; + private final String bucketName; + private final String targetName; + private final String tempName; + private final long targetGeneration; + + /** Streams the appended bytes into the temporary object; reused for its robust write/close. */ + private final WriteChannelOutputStream tempStream; + + private boolean closed = false; + + public ComposeAppendOutputStream( + Storage storage, + String bucketName, + String targetName, + String tempName, + long targetGeneration, + WriteChannel tempChannel) { + this.storage = storage; + this.bucketName = bucketName; + this.targetName = targetName; + this.tempName = tempName; + this.targetGeneration = targetGeneration; + this.tempStream = new WriteChannelOutputStream(tempChannel); + } + + @Override + public void write(int b) throws IOException { + tempStream.write(b); + } + + @Override + public void write(byte[] buf, int off, int len) throws IOException { + tempStream.write(buf, off, len); + } + + @Override + public void close() throws IOException { + if (closed) { + return; + } + closed = true; + try { + // Finish streaming the appended bytes into the temporary object. + tempStream.close(); + // Concatenate: target = target + temp. The generation precondition makes a concurrent + // modification of the target fail here rather than silently lose the appended data. + storage.compose( + ComposeRequest.newBuilder() + .addSource(targetName) + .addSource(tempName) + .setTarget(BlobInfo.newBuilder(bucketName, targetName).build()) + .setTargetOptions(BlobTargetOption.generationMatch(targetGeneration)) + .build()); + } catch (RuntimeException e) { + throw new IOException( + "Unable to append to gs://" + bucketName + "/" + targetName + " using compose", e); + } finally { + deleteTempQuietly(); + } + } + + /** The temporary object is throwaway; a failed delete only leaves a stray object, so log it. */ + private void deleteTempQuietly() { + try { + storage.delete(BlobId.of(bucketName, tempName)); + } catch (RuntimeException e) { + LogChannel.GENERAL.logError( + "Unable to delete temporary append object gs://" + bucketName + "/" + tempName, e); + } + } +} diff --git a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java index f487dd96c7..c43b8e61ae 100644 --- a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java +++ b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileObject.java @@ -20,6 +20,7 @@ import com.google.api.gax.paging.Page; import com.google.api.gax.rpc.NotFoundException; +import com.google.cloud.WriteChannel; import com.google.cloud.storage.Blob; import com.google.cloud.storage.BlobId; import com.google.cloud.storage.BlobInfo; @@ -42,6 +43,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.UUID; import org.apache.commons.vfs2.FileObject; import org.apache.commons.vfs2.FileType; import org.apache.commons.vfs2.provider.AbstractFileName; @@ -342,14 +344,36 @@ protected OutputStream doGetOutputStream(boolean bAppend) throws Exception { throw new IOException("Object needs a path within the bucket"); } Storage storage = getAbstractFileSystem().setupStorage(); + String objectName = stripTrailingSlash(bucketPath); + + if (bAppend) { + Blob current = storage.get(BlobId.of(bucketName, objectName)); + if (current != null && current.getSize() != null && current.getSize() > 0) { + return openComposeAppendStream(storage, current, objectName); + } + } + if (!hasObject()) { - this.blob = - storage.create(BlobInfo.newBuilder(bucket, stripTrailingSlash(bucketPath)).build()); + this.blob = storage.create(BlobInfo.newBuilder(bucket, objectName).build()); } getAbstractFileSystem().invalidateListCacheForParentOf(bucketName, bucketPath); return new WriteChannelOutputStream(storage.writer(blob)); } + /** + * Open a {@link ComposeAppendOutputStream} that streams the appended bytes into a throwaway + * temporary object next to the target and, on close, composes {@code target + temp} back onto the + * target. See GCS + * appends. + */ + private OutputStream openComposeAppendStream(Storage storage, Blob current, String objectName) { + String tempName = objectName + ".hop-append-" + UUID.randomUUID() + ".tmp"; + WriteChannel tempChannel = storage.writer(BlobInfo.newBuilder(bucketName, tempName).build()); + getAbstractFileSystem().invalidateListCacheForParentOf(bucketName, bucketPath); + return new ComposeAppendOutputStream( + storage, bucketName, objectName, tempName, current.getGeneration(), tempChannel); + } + @Override protected FileObject[] doListChildrenResolved() throws Exception { if (!isFolder()) { diff --git a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java index 7e6f97d381..5ccde11bf5 100644 --- a/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java +++ b/plugins/tech/google/src/main/java/org/apache/hop/vfs/gs/GoogleStorageFileProvider.java @@ -52,6 +52,9 @@ public GoogleStorageFileProvider( setServiceAccountCredentials(variables, googleStorageMetadataType); } + // APPEND_CONTENT is required for append: without it commons-vfs2 rejects getOutputStream(true) + // with "does not support append mode". Append itself is emulated with composite objects, see + // ComposeAppendOutputStream. public static final Collection capabilities = Set.of( Capability.CREATE, @@ -63,7 +66,8 @@ public GoogleStorageFileProvider( Capability.LIST_CHILDREN, Capability.READ_CONTENT, Capability.URI, - Capability.WRITE_CONTENT); + Capability.WRITE_CONTENT, + Capability.APPEND_CONTENT); @Override public Collection getCapabilities() { diff --git a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java new file mode 100644 index 0000000000..79b734940e --- /dev/null +++ b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/ComposeAppendOutputStreamTest.java @@ -0,0 +1,160 @@ +/* + * 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.hop.vfs.gs; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.cloud.WriteChannel; +import com.google.cloud.storage.BlobId; +import com.google.cloud.storage.Storage; +import com.google.cloud.storage.Storage.BlobTargetOption; +import com.google.cloud.storage.Storage.ComposeRequest; +import com.google.cloud.storage.StorageException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.InOrder; +import org.mockito.Mockito; + +/** + * Covers the composite-object append behaviour of {@link ComposeAppendOutputStream}: bytes are + * streamed into a temporary object and, on close, the target and temp object are concatenated with + * a single generation-guarded {@code compose}, after which the temp object is deleted. + */ +class ComposeAppendOutputStreamTest { + + private static final String BUCKET = "my-bucket"; + private static final String TARGET = "folder/data.txt"; + private static final String TEMP = "folder/data.txt.hop-append-abc.tmp"; + private static final long GENERATION = 42L; + + /** A WriteChannel mock that consumes every byte offered, like a healthy upload. */ + private static WriteChannel consumingChannel() { + WriteChannel channel = mock(WriteChannel.class); + try { + when(channel.write(any(ByteBuffer.class))) + .thenAnswer( + inv -> { + ByteBuffer b = inv.getArgument(0); + int remaining = b.remaining(); + b.position(b.limit()); + return remaining; + }); + } catch (IOException e) { + throw new IllegalStateException(e); + } + return channel; + } + + @Test + void closeStreamsToTempThenComposesThenDeletesTemp() throws Exception { + Storage storage = mock(Storage.class); + WriteChannel tempChannel = consumingChannel(); + + ComposeAppendOutputStream out = + new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP, GENERATION, tempChannel); + out.write("appended".getBytes(StandardCharsets.UTF_8)); + out.close(); + + // The appended bytes were streamed into the temp object and the channel closed. + verify(tempChannel).close(); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ComposeRequest.class); + verify(storage).compose(captor.capture()); + ComposeRequest request = captor.getValue(); + + // Sources are, in order, the existing target then the temp object (target = target + temp). + assertEquals(2, request.getSourceBlobs().size()); + assertEquals(TARGET, request.getSourceBlobs().get(0).getName()); + assertEquals(TEMP, request.getSourceBlobs().get(1).getName()); + + // The composite is written back onto the target. + assertEquals(BUCKET, request.getTarget().getBucket()); + assertEquals(TARGET, request.getTarget().getName()); + + // Guarded by the generation captured when the stream opened. + assertTrue( + request.getTargetOptions().contains(BlobTargetOption.generationMatch(GENERATION)), + "append must be guarded by ifGenerationMatch to fail fast on concurrent modification"); + + // And the throwaway temp object is cleaned up, after the compose. + InOrder inOrder = Mockito.inOrder(storage); + inOrder.verify(storage).compose(any(ComposeRequest.class)); + inOrder.verify(storage).delete(BlobId.of(BUCKET, TEMP)); + } + + @Test + void composeFailurePropagatesButTempIsStillDeleted() throws Exception { + Storage storage = mock(Storage.class); + when(storage.compose(any(ComposeRequest.class))) + .thenThrow(new StorageException(412, "precondition failed")); + WriteChannel tempChannel = consumingChannel(); + + ComposeAppendOutputStream out = + new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP, GENERATION, tempChannel); + out.write("x".getBytes(StandardCharsets.UTF_8)); + + IOException thrown = assertThrows(IOException.class, out::close); + assertTrue(thrown.getMessage().contains(TARGET), "the error should name the target object"); + // Even on a failed compose the temp object must not be leaked. + verify(storage).delete(BlobId.of(BUCKET, TEMP)); + } + + @Test + void closeIsIdempotent() throws Exception { + Storage storage = mock(Storage.class); + WriteChannel tempChannel = consumingChannel(); + + ComposeAppendOutputStream out = + new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP, GENERATION, tempChannel); + out.close(); + out.close(); + + // A second close must not compose (or delete) again. + verify(storage, times(1)).compose(any(ComposeRequest.class)); + verify(storage, times(1)).delete(eq(BlobId.of(BUCKET, TEMP))); + } + + @Test + void nothingIsComposedBeforeClose() throws Exception { + Storage storage = mock(Storage.class); + WriteChannel tempChannel = consumingChannel(); + + ComposeAppendOutputStream out = + new ComposeAppendOutputStream(storage, BUCKET, TARGET, TEMP, GENERATION, tempChannel); + out.write("streaming".getBytes(StandardCharsets.UTF_8)); + + // Writing alone must not compose; the concatenation happens exactly once, at close. + verify(storage, never()).compose(any(ComposeRequest.class)); + + out.close(); + verify(storage, times(1)).compose(any(ComposeRequest.class)); + } +} diff --git a/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageFileProviderCapabilitiesTest.java b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageFileProviderCapabilitiesTest.java new file mode 100644 index 0000000000..b400d6af3d --- /dev/null +++ b/plugins/tech/google/src/test/java/org/apache/hop/vfs/gs/GoogleStorageFileProviderCapabilitiesTest.java @@ -0,0 +1,40 @@ +/* + * 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.hop.vfs.gs; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.commons.vfs2.Capability; +import org.junit.jupiter.api.Test; + +class GoogleStorageFileProviderCapabilitiesTest { + + /** + * commons-vfs2 rejects {@code getOutputStream(true)} with "does not support append mode" unless + * the file system advertises {@link Capability#APPEND_CONTENT}. Append is implemented with + * composite objects (see {@link ComposeAppendOutputStream}), so the capability must be present or + * every append is blocked before reaching the provider. + */ + @Test + void appendContentIsAdvertised() { + assertTrue( + GoogleStorageFileProvider.capabilities.contains(Capability.APPEND_CONTENT), + "GCS VFS must advertise APPEND_CONTENT so commons-vfs2 allows append mode"); + } +}