You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2025/09/19 08:27:31 UTC
(ignite-3) branch IGNITE-26455 created (now 3fce6b73728)
This is an automated email from the ASF dual-hosted git repository.
mpochatkin pushed a change to branch IGNITE-26455
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at 3fce6b73728 IGNITE-26455 Add support for deploy units with folders structure
This branch includes the following new commits:
new 3fce6b73728 IGNITE-26455 Add support for deploy units with folders structure
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(ignite-3) 01/01: IGNITE-26455 Add support for deploy units with folders structure
Posted by mp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mpochatkin pushed a commit to branch IGNITE-26455
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 3fce6b73728b272a3dd42e70f7989ea995798ff8
Author: Pochatkin Mikhail <m....@gmail.com>
AuthorDate: Fri Sep 19 11:27:16 2025 +0300
IGNITE-26455 Add support for deploy units with folders structure
---
.../ignite/internal/deployment/DeployFile.java | 9 +-
.../ignite/internal/deployment/DeployFiles.java | 121 ++++++++++++++++++---
.../internal/deployment/ItDeploymentUnitTest.java | 31 ++++++
.../deployunit/DeployDeploymentUnitProcessor.java | 95 ++++++++++++++++
.../ignite/internal/deployunit/DeploymentUnit.java | 52 +++++----
.../internal/deployunit/DeploymentUnitImpl.java | 70 ++++++++++++
.../deployunit/DeploymentUnitProcessor.java | 71 ++++++++++++
.../internal/deployunit/FileDeployerService.java | 25 ++---
.../ignite/internal/deployunit/UnitContent.java | 23 +---
.../internal/deployunit/ZipDeploymentUnit.java | 84 ++++++++++++++
.../ignite/deployment/FileDeployerServiceTest.java | 31 +++++-
.../internal/compute/ItComputeTestStandalone.java | 3 +-
.../testframework/WorkDirectoryExtension.java | 15 ++-
.../rest/api/deployment/DeploymentCodeApi.java | 10 +-
.../deployment/CompletedFileUploadSubscriber.java | 51 ++++-----
.../deployment/DeploymentManagementController.java | 34 +++---
.../rest/deployment/InputStreamCollector.java | 72 ++++++++++++
.../rest/deployment/InputStreamCollectorImpl.java | 60 ++++++++++
.../rest/deployment/ZipInputStreamCollector.java | 102 +++++++++++++++++
19 files changed, 830 insertions(+), 129 deletions(-)
diff --git a/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFile.java b/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFile.java
index 071aa615a46..c76b9a84031 100644
--- a/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFile.java
+++ b/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFile.java
@@ -29,10 +29,13 @@ class DeployFile {
private final int replicaTimeout;
- DeployFile(Path file, long expectedSize, int replicaTimeout) throws IOException {
+ private final boolean zip;
+
+ DeployFile(Path file, boolean zip, long expectedSize, int replicaTimeout) throws IOException {
this.file = file;
this.expectedSize = expectedSize;
this.replicaTimeout = replicaTimeout;
+ this.zip = zip;
ensureExists();
}
@@ -46,6 +49,10 @@ class DeployFile {
return file;
}
+ boolean zip() {
+ return zip;
+ }
+
long expectedSize() {
return expectedSize;
}
diff --git a/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFiles.java b/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFiles.java
index de5a1ffb265..68d368bdf82 100644
--- a/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFiles.java
+++ b/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/DeployFiles.java
@@ -18,26 +18,34 @@
package org.apache.ignite.internal.deployment;
import static org.apache.ignite.internal.deployunit.DeploymentStatus.DEPLOYED;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.fillDummyFile;
+import static org.apache.ignite.internal.testframework.WorkDirectoryExtension.zipDirectory;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+import static org.apache.ignite.internal.util.IgniteUtils.deleteIfExists;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
+import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
-import java.util.stream.Collectors;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
import org.apache.ignite.deployment.version.Version;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.deployunit.DeploymentUnit;
+import org.apache.ignite.internal.deployunit.DeploymentUnitImpl;
import org.apache.ignite.internal.deployunit.NodesToDeploy;
import org.apache.ignite.internal.deployunit.UnitStatuses;
import org.apache.ignite.internal.deployunit.UnitStatuses.UnitStatusesBuilder;
+import org.apache.ignite.internal.deployunit.ZipDeploymentUnit;
class DeployFiles {
private static final int BASE_REPLICA_TIMEOUT = 30;
@@ -56,6 +64,10 @@ class DeployFiles {
private DeployFile bigFile;
+ private DeployFile flatZipFile;
+
+ private DeployFile treeZipFile;
+
// TODO https://issues.apache.org/jira/browse/IGNITE-20204
DeployFiles(Path workDir) {
this.workDir = workDir;
@@ -63,7 +75,30 @@ class DeployFiles {
private static DeployFile create(Path path, long size, int replicaTimeout) {
try {
- return new DeployFile(path, size, replicaTimeout);
+ return new DeployFile(path, false, size, replicaTimeout);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private DeployFile createZip(Path path, Map<String, Long> content, int replicaTimeout) {
+ try {
+ Path zipTempFolder = workDir.resolve("tempDir");
+ Files.createDirectories(zipTempFolder);
+ long size = 0;
+ for (Entry<String, Long> e : content.entrySet()) {
+ String zipEntryPath = e.getKey();
+ Long entrySize = e.getValue();
+ Path entry = zipTempFolder.resolve(zipEntryPath);
+ if (entrySize > 0) {
+ Files.createDirectories(entry.getParent());
+ fillDummyFile(entry, entrySize);
+ }
+ size += entrySize;
+ }
+ zipDirectory(zipTempFolder, path);
+ deleteIfExists(zipTempFolder);
+ return new DeployFile(path, true, size, replicaTimeout);
} catch (IOException e) {
throw new RuntimeException(e);
}
@@ -90,6 +125,39 @@ class DeployFiles {
return bigFile;
}
+ DeployFile flatZipFile() {
+ if (flatZipFile == null) {
+ flatZipFile = createZip(
+ workDir.resolve("flat.zip"),
+ Map.of(
+ "a1", SMALL_IN_BYTES,
+ "a2", SMALL_IN_BYTES,
+ "a3", SMALL_IN_BYTES,
+ "a4", SMALL_IN_BYTES
+ ),
+ BASE_REPLICA_TIMEOUT
+ );
+ }
+ return flatZipFile;
+ }
+
+ DeployFile treeZipFile() {
+ if (treeZipFile == null) {
+ treeZipFile = createZip(
+ workDir.resolve("tree.zip"),
+ Map.of(
+ "a1/a2", SMALL_IN_BYTES,
+ "b1", SMALL_IN_BYTES,
+ "c1/c2/c3/c4", MEDIUM_IN_BYTES,
+ "d1/d2", SMALL_IN_BYTES,
+ "d1/a2", SMALL_IN_BYTES
+ ),
+ BASE_REPLICA_TIMEOUT * 3
+ );
+ }
+ return treeZipFile;
+ }
+
private Unit deployAndVerify(String id, Version version, DeployFile file, IgniteImpl entryNode) {
return deployAndVerify(id, version, false, file, entryNode);
}
@@ -110,14 +178,8 @@ class DeployFiles {
NodesToDeploy nodesToDeploy,
IgniteImpl entryNode
) {
- List<Path> paths = files.stream()
- .map(DeployFile::file)
- .collect(Collectors.toList());
-
- CompletableFuture<Boolean> deploy;
-
- DeploymentUnit deploymentUnit = fromPaths(paths);
- deploy = entryNode.deployment()
+ DeploymentUnit deploymentUnit = fromFiles(files);
+ CompletableFuture<Boolean> deploy = entryNode.deployment()
.deployAsync(id, version, force, deploymentUnit, nodesToDeploy)
.whenComplete((res, err) -> {
try {
@@ -134,8 +196,19 @@ class DeployFiles {
Path nodeUnitDirectory = unit.getNodeUnitDirectory(entryNode);
for (DeployFile file : files) {
- Path filePath = nodeUnitDirectory.resolve(file.file().getFileName());
- assertTrue(Files.exists(filePath));
+ if (file.zip()) {
+ try (ZipInputStream zis = new ZipInputStream(Files.newInputStream(file.file()))) {
+ ZipEntry ze;
+ while ((ze = zis.getNextEntry()) != null) {
+ assertTrue(Files.exists(nodeUnitDirectory.resolve(ze.getName())));
+ }
+ } catch (IOException e) {
+ fail(e);
+ }
+ } else {
+ Path filePath = nodeUnitDirectory.resolve(file.file().getFileName());
+ assertTrue(Files.exists(filePath));
+ }
}
return unit;
@@ -153,6 +226,14 @@ class DeployFiles {
return deployAndVerify(id, version, bigFile(), entryNode);
}
+ public Unit deployAndVerifyFlatZip(String id, Version version, IgniteImpl entryNode) {
+ return deployAndVerify(id, version, flatZipFile(), entryNode);
+ }
+
+ public Unit deployAndVerifyTreeZip(String id, Version version, IgniteImpl entryNode) {
+ return deployAndVerify(id, version, treeZipFile(), entryNode);
+ }
+
public static UnitStatuses buildStatus(String id, Unit... units) {
UnitStatusesBuilder builder = UnitStatuses.builder(id);
for (Unit unit : units) {
@@ -169,16 +250,20 @@ class DeployFiles {
Files.copy(file.file(), unitFile);
}
- private static DeploymentUnit fromPaths(List<Path> paths) {
- Objects.requireNonNull(paths);
+ private static DeploymentUnit fromFiles(List<DeployFile> files) {
Map<String, InputStream> map = new HashMap<>();
+ List<ZipInputStream> zips = new ArrayList<>();
try {
- for (Path path : paths) {
- map.put(path.getFileName().toString(), Files.newInputStream(path));
+ for (DeployFile file : files) {
+ if (file.zip()) {
+ zips.add(new ZipInputStream(Files.newInputStream(file.file())));
+ } else {
+ map.put(file.file().getFileName().toString(), Files.newInputStream(file.file()));
+ }
}
} catch (IOException e) {
throw new RuntimeException(e);
}
- return new DeploymentUnit(map);
+ return new ZipDeploymentUnit(new DeploymentUnitImpl(map), zips);
}
}
diff --git a/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java b/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
index fc5554e8baa..520da9f3dfa 100644
--- a/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
+++ b/modules/code-deployment/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
@@ -281,4 +281,35 @@ public class ItDeploymentUnitTest extends ClusterPerTestIntegrationTest {
smallUnit.waitUnitClean(igniteImpl(2));
}
+
+ @Test
+ public void testZipDeploy() {
+ String id = "test";
+ Unit unit = files.deployAndVerifyFlatZip(id, Version.parseVersion("1.1.0"), igniteImpl(1));
+
+ Unit unit2 = files.deployAndVerifyTreeZip(id, Version.parseVersion("1.1.1"), igniteImpl(1));
+
+ UnitStatuses status = buildStatus(id, unit, unit2);
+
+ await().timeout(2, SECONDS)
+ .pollDelay(500, MILLISECONDS)
+ .until(() -> igniteImpl(2).deployment().clusterStatusesAsync(), willBe(List.of(status)));
+ }
+
+ @Test
+ public void testZipAndFileDeploy() {
+ String id = "test";
+
+ Unit unit = files.deployAndVerify(
+ id, Version.parseVersion("1.1.0"), false, List.of(files.smallFile(), files.flatZipFile()),
+ new NodesToDeploy(List.of(igniteImpl(1).name())),
+ igniteImpl(0)
+ );
+
+ UnitStatuses status = buildStatus(id, unit);
+
+ await().timeout(2, SECONDS)
+ .pollDelay(500, MILLISECONDS)
+ .until(() -> igniteImpl(2).deployment().clusterStatusesAsync(), willBe(List.of(status)));
+ }
}
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployDeploymentUnitProcessor.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployDeploymentUnitProcessor.java
new file mode 100644
index 00000000000..50e45d4a3f8
--- /dev/null
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployDeploymentUnitProcessor.java
@@ -0,0 +1,95 @@
+/*
+ * 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.ignite.internal.deployunit;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map.Entry;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipInputStream;
+
+/**
+ * Implementation of {@link DeploymentUnitProcessor} that deploys deployment unit content to the file system.
+ *
+ * <p>This processor extracts and deploys files from deployment units to a specified target directory.
+ * It handles both regular deployment units (containing individual files as input streams) and ZIP-based
+ * deployment units (containing compressed archives that need extraction).
+ *
+ * <p>The deployment process ensures atomic file operations by:
+ * <ul>
+ * <li>First copying files to temporary locations with a {@code .tmp} suffix</li>
+ * <li>Then atomically moving them to their final destinations</li>
+ * <li>Creating necessary parent directories as needed</li>
+ * </ul>
+ *
+ * <p>This approach prevents partial deployments and ensures that files are either fully deployed
+ * or not deployed at all, maintaining consistency during the deployment process.
+ *
+ * <p>Type parameters:
+ * <ul>
+ * <li>{@code Path} - the argument type representing the target deployment directory</li>
+ * <li>{@code Void} - the return type (no meaningful return value)</li>
+ * </ul>
+ */
+public class DeployDeploymentUnitProcessor implements DeploymentUnitProcessor<Path, Void> {
+ /** Suffix used for temporary files during the deployment process. */
+ private static final String TMP_SUFFIX = ".tmp";
+
+ /** {@inheritDoc} */
+ @Override
+ public Void processContent(DeploymentUnitImpl unit, Path unitFolder) throws IOException {
+ for (Entry<String, InputStream> e : unit.content().entrySet()) {
+ doDeploy(unitFolder, e.getKey(), e.getValue());
+ }
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Void processContentWithUnzip(ZipDeploymentUnit unit, Path unitFolder) throws IOException {
+ for (ZipInputStream zis : unit.zipContent()) {
+ ZipEntry ze;
+ while ((ze = zis.getNextEntry()) != null) {
+ Path entryPath = unitFolder.resolve(ze.getName());
+
+ if (ze.isDirectory()) {
+ Files.createDirectories(entryPath);
+ } else {
+ Path unitFileFolder = entryPath.getParent();
+ Files.createDirectories(unitFileFolder);
+ doDeploy(unitFileFolder, entryPath.getFileName().toString(), zis);
+ }
+ }
+ }
+ return null;
+ }
+
+ private static void doDeploy(Path unitFolder, String entryName, InputStream is) throws IOException {
+ Path unitPath = unitFolder.resolve(entryName);
+ Files.createDirectories(unitPath.getParent());
+ Path unitPathTmp = unitFolder.resolve(entryName + TMP_SUFFIX);
+ Files.copy(is, unitPathTmp, REPLACE_EXISTING);
+ Files.move(unitPathTmp, unitPath, ATOMIC_MOVE, REPLACE_EXISTING);
+ }
+
+}
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
index bb36c8dccc0..0d701a994e2 100644
--- a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
@@ -17,32 +17,40 @@
package org.apache.ignite.internal.deployunit;
-import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
-
-import java.io.InputStream;
-import java.util.Map;
+import java.io.IOException;
/**
- * Deployment unit interface.
+ * Interface representing a deployment unit in the Apache Ignite code deployment system.
+ *
+ * <p>A deployment unit is a container for code and resources that can be deployed to and managed
+ * within an Ignite cluster. This interface provides a contract for processing deployment unit
+ * content through a processor pattern, allowing for flexible handling of different types of
+ * deployment units such as regular file-based units and ZIP-compressed units.
*/
-public class DeploymentUnit implements AutoCloseable {
- private final Map<String, InputStream> content;
-
- public DeploymentUnit(Map<String, InputStream> content) {
- this.content = content;
- }
-
+public interface DeploymentUnit extends AutoCloseable {
/**
- * Deployment unit content - a map from file name to input stream.
+ * Processes the deployment unit content using the provided processor.
+ *
+ * <p>This method delegates the processing of the deployment unit to the specified processor,
+ * following a strategy pattern. The processor determines how the unit content should be
+ * handled based on its implementation. Different processors can perform different operations
+ * on the same deployment unit.
+ *
+ * <p>The method supports generic type parameters to allow processors to work with different
+ * argument types and return different result types, providing flexibility for various
+ * processing scenarios.
+ *
+ * <p>For ZIP-based deployment units, the processor may need to handle both regular content
+ * and compressed content that requires extraction.
*
- * @return Deployment unit content.
+ *
+ * @param <T> the type of argument passed to the processor
+ * @param <R> the type of result returned by the processor
+ * @param processor the processor that will handle the deployment unit content
+ * @param unitFolder the argument to be passed to the processor during processing
+ * @return the result of the processing operation as determined by the processor
+ * @throws IOException if an I/O error occurs during processing, such as issues reading
+ * deployment unit content or writing processed results.
*/
- public Map<String, InputStream> content() {
- return content;
- }
-
- @Override
- public void close() throws Exception {
- closeAll(content.values());
- }
+ <T, R> R process(DeploymentUnitProcessor<T, R> processor, T unitFolder) throws IOException;
}
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitImpl.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitImpl.java
new file mode 100644
index 00000000000..1b4ec14e3ee
--- /dev/null
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitImpl.java
@@ -0,0 +1,70 @@
+/*
+ * 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.ignite.internal.deployunit;
+
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+/**
+ * Standard implementation of {@link DeploymentUnit} that handles regular (non-compressed) deployment content.
+ *
+ * <p>This class represents a deployment unit containing a collection of files, where each file is
+ * represented as a mapping from file name to its corresponding {@link InputStream}. This implementation is designed for straightforward
+ * deployment scenarios where the content does not require compression or special extraction handling.
+ *
+ * <p>The {@code DeploymentUnitImpl} is the primary implementation used for most deployment operations
+ * in the Apache Ignite code deployment system. It provides a simple and efficient way to package and deploy code artifacts, configuration
+ * files, resources, and other deployment assets to Ignite cluster nodes.
+ */
+public class DeploymentUnitImpl implements DeploymentUnit {
+
+ /**
+ * The deployment unit content represented as a mapping from file names to their input streams. Each entry represents a file within the
+ * deployment unit.
+ */
+ private final Map<String, InputStream> content;
+
+ /**
+ * Constructor.
+ */
+ public DeploymentUnitImpl(Map<String, InputStream> content) {
+ this.content = content;
+ }
+
+ /**
+ * Returns the deployment unit content as a map of file names to input streams.
+ */
+ public Map<String, InputStream> content() {
+ return content;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void close() throws Exception {
+ closeAll(content.values());
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <T, R> R process(DeploymentUnitProcessor<T, R> processor, T arg) throws IOException {
+ return processor.processContent(this, arg);
+ }
+}
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitProcessor.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitProcessor.java
new file mode 100644
index 00000000000..eb9351bd92e
--- /dev/null
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnitProcessor.java
@@ -0,0 +1,71 @@
+/*
+ * 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.ignite.internal.deployunit;
+
+import java.io.IOException;
+
+/**
+ * Processor interface for handling deployment unit content operations.
+ *
+ * <p>This interface defines a contract for processing deployment units, providing methods to handle
+ * both regular deployment units and ZIP-based deployment units. Implementations of this interface
+ * can perform various operations on deployment unit content such as deployment, validation,
+ * transformation, or extraction.
+ *
+ * <p>The processor uses a generic approach with two type parameters:
+ * <ul>
+ * <li>{@code T} - the type of argument passed to processing methods</li>
+ * <li>{@code R} - the type of result returned by processing methods</li>
+ * </ul>
+ *
+ * <p>This design allows for flexible implementations that can process deployment units with
+ * different argument types and return different result types based on the specific use case.
+ *
+ * @param <T> the type of argument passed to the processing methods
+ * @param <R> the type of result returned by the processing methods
+ */
+public interface DeploymentUnitProcessor<T, R> {
+ /**
+ * Processes the content of a regular deployment unit.
+ *
+ * <p>This method handles deployment units that contain a collection of files represented
+ * as input streams. The implementation should process each file in the deployment unit
+ * according to the specific processor's logic.
+ *
+ * @param unit the deployment unit containing the content to be processed
+ * @param arg the argument to be used during processing
+ * @return the result of the processing operation
+ * @throws IOException if an I/O error occurs during processing
+ */
+ R processContent(DeploymentUnitImpl unit, T arg) throws IOException;
+
+ /**
+ * Processes the content of a ZIP-based deployment unit with automatic extraction.
+ *
+ * <p>This method handles deployment units that contain ZIP archives. The implementation
+ * should process the ZIP content by extracting and handling each entry according to the
+ * specific processor's logic. This method is typically used when the deployment unit
+ * contains compressed content that needs to be extracted during processing.
+ *
+ * @param unit the ZIP deployment unit containing the compressed content to be processed
+ * @param arg the argument to be used during processing
+ * @return the result of the processing operation
+ * @throws IOException if an I/O error occurs during processing or extraction
+ */
+ R processContentWithUnzip(ZipDeploymentUnit unit, T arg) throws IOException;
+}
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
index 73d13708c7d..605cc82eb98 100644
--- a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
@@ -17,11 +17,7 @@
package org.apache.ignite.internal.deployunit;
-import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
-import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
-
import java.io.IOException;
-import java.io.InputStream;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -29,7 +25,6 @@ import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -46,8 +41,6 @@ import org.apache.ignite.internal.util.IgniteUtils;
public class FileDeployerService {
private static final IgniteLogger LOG = Loggers.forClass(FileDeployerService.class);
- private static final String TMP_SUFFIX = ".tmp";
-
private static final int DEPLOYMENT_EXECUTOR_SIZE = 4;
/**
@@ -57,6 +50,8 @@ public class FileDeployerService {
private final ExecutorService executor;
+ private final DeploymentUnitProcessor<Path, Void> deployProcessor = new DeployDeploymentUnitProcessor();
+
/** Constructor. */
public FileDeployerService(String nodeName) {
executor = Executors.newFixedThreadPool(
@@ -81,16 +76,10 @@ public class FileDeployerService {
return CompletableFuture.supplyAsync(() -> {
try {
Path unitFolder = unitPath(id, version);
-
Files.createDirectories(unitFolder);
- for (Entry<String, InputStream> entry : deploymentUnit.content().entrySet()) {
- String fileName = entry.getKey();
- Path unitPath = unitFolder.resolve(fileName);
- Path unitPathTmp = unitFolder.resolve(fileName + TMP_SUFFIX);
- Files.copy(entry.getValue(), unitPathTmp, REPLACE_EXISTING);
- Files.move(unitPathTmp, unitPath, ATOMIC_MOVE, REPLACE_EXISTING);
- }
+ deploymentUnit.process(deployProcessor, unitFolder);
+
return true;
} catch (IOException e) {
LOG.error("Failed to deploy unit " + id + ":" + version, e);
@@ -129,10 +118,12 @@ public class FileDeployerService {
return CompletableFuture.supplyAsync(() -> {
Map<String, byte[]> result = new HashMap<>();
try {
- Files.walkFileTree(unitPath(id, version), new SimpleFileVisitor<>() {
+ Path unitFolder = unitPath(id, version);
+ Files.walkFileTree(unitFolder, new SimpleFileVisitor<>() {
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
- result.put(file.getFileName().toString(), Files.readAllBytes(file));
+ Path unitStructure = unitFolder.relativize(file);
+ result.put(unitStructure.toString(), Files.readAllBytes(file));
return FileVisitResult.CONTINUE;
}
});
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitContent.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitContent.java
index 6ea4cf60b02..dcaa29a5246 100644
--- a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitContent.java
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitContent.java
@@ -18,15 +18,12 @@
package org.apache.ignite.internal.deployunit;
import java.io.ByteArrayInputStream;
-import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.stream.Collectors;
-import org.apache.ignite.internal.deployunit.exception.DeploymentUnitReadException;
/**
* Unit content representation.
@@ -78,24 +75,6 @@ public class UnitContent implements Iterable<Entry<String, byte[]>> {
return files.entrySet().iterator();
}
- /**
- * Read unit content from unit {@link DeploymentUnit}.
- *
- * @param deploymentUnit Deployment unit instance.
- * @return Unit content from provided deployment unit.
- */
- public static UnitContent readContent(DeploymentUnit deploymentUnit) {
- Map<String, byte[]> map = deploymentUnit.content().entrySet().stream()
- .collect(Collectors.toMap(Entry::getKey, entry -> {
- try {
- return entry.getValue().readAllBytes();
- } catch (IOException e) {
- throw new DeploymentUnitReadException(e);
- }
- }));
- return new UnitContent(map);
- }
-
/**
* Convert unit content to {@link DeploymentUnit}.
*
@@ -107,6 +86,6 @@ public class UnitContent implements Iterable<Entry<String, byte[]>> {
content.iterator().forEachRemaining(it -> {
files.put(it.getKey(), new ByteArrayInputStream(it.getValue()));
});
- return new DeploymentUnit(files);
+ return new DeploymentUnitImpl(files);
}
}
diff --git a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/ZipDeploymentUnit.java b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/ZipDeploymentUnit.java
new file mode 100644
index 00000000000..0758379d89a
--- /dev/null
+++ b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/ZipDeploymentUnit.java
@@ -0,0 +1,84 @@
+/*
+ * 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.ignite.internal.deployunit;
+
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.zip.ZipInputStream;
+
+/**
+ * A specialized implementation of {@link DeploymentUnit} that handles ZIP-compressed deployment content.
+ *
+ * <p>This class represents a deployment unit that contains both regular (uncompressed) content and
+ * ZIP-compressed archives that require extraction during processing. It extends the basic deployment
+ * unit functionality to handle mixed content types, providing automatic extraction capabilities for
+ * ZIP archives while maintaining support for regular file content.
+ */
+public class ZipDeploymentUnit implements DeploymentUnit {
+ /** The deployment unit containing regular (non-ZIP) content. */
+ private final DeploymentUnit notZippedContent;
+
+ /** Collection of ZIP input streams that require extraction during processing. */
+ private final Collection<ZipInputStream> zipContent;
+
+ /**
+ * Constructor.
+ */
+ public ZipDeploymentUnit(DeploymentUnit notZippedContent, Collection<ZipInputStream> zipContent) {
+ this.notZippedContent = notZippedContent;
+ this.zipContent = zipContent;
+ }
+
+ /**
+ * Processes the deployment unit content using a two-phase approach for mixed content types.
+ *
+ * <p>This method implements the {@link DeploymentUnit} processing contract with specialized
+ * handling for ZIP content.
+ *
+ * @param <T> the type of argument passed to the processor.
+ * @param <R> the type of result returned by the processor.
+ * @param processor the processor that will handle both regular and ZIP content;
+ * must implement both {@code processContent} and {@code processContentWithUnzip} methods.
+ * @param arg the argument to be passed to the processor during both processing phases.
+ * @return the result of the ZIP content processing phase.
+ * @throws IOException if an I/O error occurs during either processing phase.
+ */
+ @Override
+ public <T, R> R process(DeploymentUnitProcessor<T, R> processor, T arg) throws IOException {
+ notZippedContent.process(processor, arg);
+ return processor.processContentWithUnzip(this, arg);
+ }
+
+ /**
+ * Returns the collection of ZIP input streams that require extraction during processing.
+ */
+ public Collection<ZipInputStream> zipContent() {
+ return zipContent;
+ }
+
+ /**
+ * Closes this deployment unit and releases all associated resources.
+ */
+ @Override
+ public void close() throws Exception {
+ notZippedContent.close();
+ closeAll(zipContent);
+ }
+}
diff --git a/modules/code-deployment/src/test/java/org/apache/ignite/deployment/FileDeployerServiceTest.java b/modules/code-deployment/src/test/java/org/apache/ignite/deployment/FileDeployerServiceTest.java
index 0cb57643735..33e06ace784 100644
--- a/modules/code-deployment/src/test/java/org/apache/ignite/deployment/FileDeployerServiceTest.java
+++ b/modules/code-deployment/src/test/java/org/apache/ignite/deployment/FileDeployerServiceTest.java
@@ -29,12 +29,15 @@ import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.ignite.internal.deployunit.DeploymentUnit;
+import org.apache.ignite.internal.deployunit.DeploymentUnitImpl;
import org.apache.ignite.internal.deployunit.FileDeployerService;
import org.apache.ignite.internal.deployunit.UnitContent;
+import org.apache.ignite.internal.deployunit.exception.DeploymentUnitReadException;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
@@ -70,18 +73,18 @@ public class FileDeployerServiceTest {
@Test
public void test() throws Exception {
- try (DeploymentUnit unit = content()) {
+ try (DeploymentUnitImpl unit = content()) {
CompletableFuture<Boolean> deployed = service.deploy("id", parseVersion("1.0.0"), unit);
assertThat(deployed, willBe(true));
}
- try (DeploymentUnit unit = content()) {
+ try (DeploymentUnitImpl unit = content()) {
CompletableFuture<UnitContent> unitContent = service.getUnitContent("id", parseVersion("1.0.0"));
- assertThat(unitContent, willBe(equalTo(UnitContent.readContent(unit))));
+ assertThat(unitContent, willBe(equalTo(readContent(unit))));
}
}
- private DeploymentUnit content() {
+ private DeploymentUnitImpl content() {
Map<String, InputStream> map = Stream.of(file1, file2, file3)
.collect(Collectors.toMap(it -> it.getFileName().toString(), it -> {
try {
@@ -95,6 +98,24 @@ public class FileDeployerServiceTest {
}
}));
- return new DeploymentUnit(map);
+ return new DeploymentUnitImpl(map);
+ }
+
+ /**
+ * Read unit content from unit {@link DeploymentUnit}.
+ *
+ * @param deploymentUnit Deployment unit instance.
+ * @return Unit content from provided deployment unit.
+ */
+ private static UnitContent readContent(DeploymentUnitImpl deploymentUnit) {
+ Map<String, byte[]> map = deploymentUnit.content().entrySet().stream()
+ .collect(Collectors.toMap(Entry::getKey, entry -> {
+ try {
+ return entry.getValue().readAllBytes();
+ } catch (IOException e) {
+ throw new DeploymentUnitReadException(e);
+ }
+ }));
+ return new UnitContent(map);
}
}
diff --git a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestStandalone.java b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestStandalone.java
index 6c7964a8ff2..af74af30a03 100644
--- a/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestStandalone.java
+++ b/modules/compute/src/integrationTest/java/org/apache/ignite/internal/compute/ItComputeTestStandalone.java
@@ -48,6 +48,7 @@ import org.apache.ignite.compute.JobTarget;
import org.apache.ignite.compute.TaskDescriptor;
import org.apache.ignite.deployment.DeploymentUnit;
import org.apache.ignite.deployment.version.Version;
+import org.apache.ignite.internal.deployunit.DeploymentUnitImpl;
import org.apache.ignite.internal.deployunit.IgniteDeployment;
import org.apache.ignite.internal.deployunit.NodesToDeploy;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
@@ -229,7 +230,7 @@ class ItComputeTestStandalone extends ItComputeBaseTest {
CompletableFuture<Boolean> deployed = deployment.deployAsync(
unitId,
unitVersion,
- new org.apache.ignite.internal.deployunit.DeploymentUnit(Map.of(jarName, jarStream)),
+ new DeploymentUnitImpl(Map.of(jarName, jarStream)),
new NodesToDeploy(MAJORITY)
);
diff --git a/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/WorkDirectoryExtension.java b/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/WorkDirectoryExtension.java
index 76327903740..21845b4be36 100644
--- a/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/WorkDirectoryExtension.java
+++ b/modules/core/src/testFixtures/java/org/apache/ignite/internal/testframework/WorkDirectoryExtension.java
@@ -307,7 +307,20 @@ public class WorkDirectoryExtension
return PATTERN.matcher(property).matches();
}
- private static void zipDirectory(Path source, Path target) {
+ /**
+ * Creates a ZIP archive from the contents of a source directory.
+ *
+ * <p>This method recursively walks through the source directory and compresses all files (excluding directories)
+ * into a ZIP archive at the target location. The directory structure is preserved within the ZIP file using
+ * relative paths from the source directory.
+ *
+ * <p>The parent directories of the target ZIP file are created if they don't exist. If the target file already
+ * exists, it will be overwritten.
+ *
+ * @param source the path to the source directory to be zipped; must be an existing directory
+ * @param target the path where the ZIP archive will be created; parent directories will be created if necessary
+ */
+ public static void zipDirectory(Path source, Path target) {
try {
Files.createDirectories(target.getParent());
diff --git a/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java b/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
index a903ba6b912..2baf0d11b1f 100644
--- a/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
+++ b/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
@@ -30,6 +30,8 @@ import io.micronaut.http.annotation.Post;
import io.micronaut.http.annotation.QueryValue;
import io.micronaut.http.multipart.CompletedFileUpload;
import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.enums.ParameterIn;
import io.swagger.v3.oas.annotations.media.ArraySchema;
import io.swagger.v3.oas.annotations.media.Content;
import io.swagger.v3.oas.annotations.media.Schema;
@@ -77,7 +79,13 @@ public interface DeploymentCodeApi {
Optional<InitialDeployMode> deployMode,
@QueryValue
@Schema(name = "initialNodes", requiredMode = REQUIRED, description = "List of node identifiers to deploy to.")
- Optional<List<String>> initialNodes
+ Optional<List<String>> initialNodes,
+ @Parameter(
+ name = "X-Unzip-Units",
+ in = ParameterIn.HEADER,
+ description = "Unzip all uploaded archives with saving directory structure."
+ )
+ boolean unzip
);
/**
diff --git a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java
index e9c67064715..1d3dd4be065 100644
--- a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java
+++ b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java
@@ -19,9 +19,6 @@ package org.apache.ignite.internal.rest.deployment;
import io.micronaut.http.multipart.CompletedFileUpload;
import java.io.IOException;
-import java.io.InputStream;
-import java.util.HashMap;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.deployunit.DeploymentUnit;
import org.apache.ignite.internal.logger.IgniteLogger;
@@ -33,14 +30,19 @@ import org.reactivestreams.Subscription;
* Implementation of {@link Subscriber} based on {@link CompletedFileUpload} which will collect uploaded files to the
* {@link DeploymentUnit}.
*/
-class CompletedFileUploadSubscriber implements Subscriber<CompletedFileUpload>, AutoCloseable {
+class CompletedFileUploadSubscriber implements Subscriber<CompletedFileUpload> {
private static final IgniteLogger LOG = Loggers.forClass(CompletedFileUploadSubscriber.class);
private final CompletableFuture<DeploymentUnit> result = new CompletableFuture<>();
- private final Map<String, InputStream> content = new HashMap<>();
+ private final InputStreamCollector collector;
- private IOException ex;
+ private Throwable ex;
+
+ public CompletedFileUploadSubscriber(boolean unzip) {
+ InputStreamCollector isCollector = new InputStreamCollectorImpl();
+ this.collector = unzip ? new ZipInputStreamCollector(isCollector) : isCollector;
+ }
@Override
public void onSubscribe(Subscription subscription) {
@@ -50,20 +52,22 @@ class CompletedFileUploadSubscriber implements Subscriber<CompletedFileUpload>,
@Override
public void onNext(CompletedFileUpload item) {
try {
- content.put(item.getFilename(), item.getInputStream());
+ collector.addInputStream(item.getFilename(), item.getInputStream());
} catch (IOException e) {
LOG.error("Failed to read file: " + item.getFilename(), e);
- if (ex != null) {
- ex.addSuppressed(e);
- } else {
- ex = e;
- }
+ suppressException(e);
}
}
@Override
public void onError(Throwable throwable) {
- result.completeExceptionally(throwable);
+ try {
+ collector.clear();
+ } catch (Exception e) {
+ suppressException(e);
+ }
+ suppressException(throwable);
+ result.completeExceptionally(ex);
}
@Override
@@ -71,22 +75,19 @@ class CompletedFileUploadSubscriber implements Subscriber<CompletedFileUpload>,
if (ex != null) {
result.completeExceptionally(ex);
} else {
- result.complete(new DeploymentUnit(content));
+ result.complete(collector.toDeploymentUnit());
}
}
- public CompletableFuture<DeploymentUnit> result() {
- return result;
+ private void suppressException(Throwable t) {
+ if (ex == null) {
+ ex = t;
+ } else {
+ ex.addSuppressed(t);
+ }
}
- @Override
- public void close() throws Exception {
- result.thenAccept(it -> {
- try {
- it.close();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
+ public CompletableFuture<DeploymentUnit> result() {
+ return result;
}
}
diff --git a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
index 58996ee418b..48837ee9756 100644
--- a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
+++ b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.rest.deployment;
+import static org.apache.ignite.deployment.version.Version.parseVersion;
+
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.multipart.CompletedFileUpload;
import java.util.ArrayList;
@@ -63,30 +65,30 @@ public class DeploymentManagementController implements DeploymentCodeApi, Resour
String unitVersion,
Publisher<CompletedFileUpload> unitContent,
Optional<InitialDeployMode> deployMode,
- Optional<List<String>> initialNodes
+ Optional<List<String>> initialNodes,
+ boolean unzip
) {
-
- CompletedFileUploadSubscriber subscriber = new CompletedFileUploadSubscriber();
+ CompletedFileUploadSubscriber subscriber = new CompletedFileUploadSubscriber(unzip);
unitContent.subscribe(subscriber);
NodesToDeploy nodesToDeploy = initialNodes.map(NodesToDeploy::new)
.orElseGet(() -> new NodesToDeploy(fromInitialDeployMode(deployMode)));
- return subscriber.result().thenCompose(content -> {
- return deployment.deployAsync(unitId, Version.parseVersion(unitVersion), content, nodesToDeploy);
- }).whenComplete((res, throwable) -> {
- try {
- subscriber.close();
- } catch (Exception e) {
- LOG.error("Failed to close subscriber", e);
- }
- });
-
+ return subscriber.result().thenCompose(deploymentUnit ->
+ deployment.deployAsync(unitId, parseVersion(unitVersion), deploymentUnit, nodesToDeploy)
+ .whenComplete((unitStatus, throwable) -> {
+ try {
+ deploymentUnit.close();
+ } catch (Exception e) {
+ LOG.error("Failed to close subscriber", e);
+ }
+ })
+ );
}
@Override
public CompletableFuture<Boolean> undeploy(String unitId, String unitVersion) {
- return deployment.undeployAsync(unitId, Version.parseVersion(unitVersion));
+ return deployment.undeployAsync(unitId, parseVersion(unitVersion));
}
@Override
@@ -107,7 +109,7 @@ public class DeploymentManagementController implements DeploymentCodeApi, Resour
private CompletableFuture<List<UnitStatuses>> clusterStatuses(String unitId, Optional<String> version) {
if (version.isPresent()) {
- Version parsedVersion = Version.parseVersion(version.get());
+ Version parsedVersion = parseVersion(version.get());
return deployment.clusterStatusAsync(unitId, parsedVersion)
.thenApply(deploymentStatus -> {
if (deploymentStatus != null) {
@@ -140,7 +142,7 @@ public class DeploymentManagementController implements DeploymentCodeApi, Resour
private CompletableFuture<List<UnitStatuses>> nodeStatuses(String unitId, Optional<String> version) {
if (version.isPresent()) {
- Version parsedVersion = Version.parseVersion(version.get());
+ Version parsedVersion = parseVersion(version.get());
return deployment.nodeStatusAsync(unitId, parsedVersion)
.thenApply(deploymentStatus -> {
if (deploymentStatus != null) {
diff --git a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/InputStreamCollector.java b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/InputStreamCollector.java
new file mode 100644
index 00000000000..f7d77a39b86
--- /dev/null
+++ b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/InputStreamCollector.java
@@ -0,0 +1,72 @@
+/*
+ * 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.ignite.internal.rest.deployment;
+
+import java.io.InputStream;
+import org.apache.ignite.internal.deployunit.DeploymentUnit;
+
+/**
+ * Interface for collecting input streams and converting them into deployment units.
+ *
+ * <p>This interface provides a contract for accumulating input streams from various sources
+ * (such as file uploads, network transfers, or other data sources) and organizing them into
+ * a structured deployment unit that can be processed by the Apache Ignite deployment system.
+ *
+ * <p>Implementations of this interface are typically used in REST endpoints, file upload
+ * handlers, and other scenarios where deployment content is received in stream format and
+ * needs to be converted into deployment units for processing.
+ */
+public interface InputStreamCollector {
+ /**
+ * Adds an input stream with the specified filename to the collection.
+ *
+ * <p>This method accepts an input stream representing a file or resource that should be
+ * included in the deployment unit. The filename parameter provides the logical name or
+ * path for the content within the deployment unit structure.
+ *
+ * <p>Once added, the input stream becomes managed by this collector and should not be
+ * closed directly by the caller. The collector is responsible for proper cleanup of
+ * all managed streams.
+ *
+ *
+ * @param filename the logical name or path for the content within the deployment unit;
+ * must not be {@code null} or empty.
+ * @param is the input stream containing the content to be added; must not be {@code null}.
+ */
+ void addInputStream(String filename, InputStream is);
+
+ /**
+ * Clears all collected input streams and releases associated resources.
+ *
+ * <p>This method closes all input streams that have been added to this collector and
+ * clears the internal collection. After calling this method, the collector returns to
+ * its initial empty state and can be reused for collecting new streams.
+ */
+ void clear() throws Exception;
+
+ /**
+ * Converts the collected input streams into a deployment unit.
+ *
+ * <p>This method creates a {@link DeploymentUnit} instance containing all the input streams
+ * that have been added to this collector. The specific type of deployment unit returned
+ * depends on the implementation and the characteristics of the collected content.
+ *
+ * @return a deployment unit containing all collected input streams; never {@code null}.
+ */
+ DeploymentUnit toDeploymentUnit();
+}
diff --git a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/InputStreamCollectorImpl.java b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/InputStreamCollectorImpl.java
new file mode 100644
index 00000000000..e9520f6fc57
--- /dev/null
+++ b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/InputStreamCollectorImpl.java
@@ -0,0 +1,60 @@
+/*
+ * 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.ignite.internal.rest.deployment;
+
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.ignite.internal.deployunit.DeploymentUnit;
+import org.apache.ignite.internal.deployunit.DeploymentUnitImpl;
+
+/**
+ * Standard implementation of {@link InputStreamCollector} for collecting regular file content.
+ *
+ * <p>This implementation provides a straightforward approach to collecting input streams and
+ * converting them into a standard {@link DeploymentUnitImpl}. It maintains an internal map
+ * of filename-to-stream associations and creates deployment units containing regular
+ * (non-compressed) file content.
+ */
+public class InputStreamCollectorImpl implements InputStreamCollector {
+ /**
+ * Internal storage for collected input streams mapped by their filenames.
+ * The map maintains the association between logical file paths and their content streams.
+ */
+ private final Map<String, InputStream> content = new HashMap<>();
+
+ /** {@inheritDoc} */
+ @Override
+ public void addInputStream(String filename, InputStream is) {
+ content.put(filename, is);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public DeploymentUnit toDeploymentUnit() {
+ return new DeploymentUnitImpl(content);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void clear() throws Exception {
+ closeAll(content.values());
+ }
+}
diff --git a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/ZipInputStreamCollector.java b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/ZipInputStreamCollector.java
new file mode 100644
index 00000000000..b4c41f290c9
--- /dev/null
+++ b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/ZipInputStreamCollector.java
@@ -0,0 +1,102 @@
+/*
+ * 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.ignite.internal.rest.deployment;
+
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.zip.ZipInputStream;
+import org.apache.ignite.internal.deployunit.DeploymentUnit;
+import org.apache.ignite.internal.deployunit.ZipDeploymentUnit;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Advanced implementation of {@link InputStreamCollector} that automatically detects and handles ZIP content.
+ *
+ * <p>This decorator implementation wraps another {@link InputStreamCollector} and provides intelligent
+ * handling of mixed content types. It automatically detects ZIP archives among the added input streams
+ * and separates them from regular file content, creating a {@link ZipDeploymentUnit} that can handle
+ * both types of content appropriately.
+ */
+public class ZipInputStreamCollector implements InputStreamCollector {
+ /**
+ * The delegate collector that handles regular (non-ZIP) content.
+ * All streams that are not identified as ZIP archives are forwarded to this collector.
+ */
+ private final InputStreamCollector delegate;
+
+ /**
+ * Collection of detected ZIP input streams that require special processing.
+ * These streams will be handled by the ZipDeploymentUnit for automatic extraction.
+ */
+ private final List<ZipInputStream> zipContent = new ArrayList<>();
+
+ /**
+ * Constructor.
+ */
+ public ZipInputStreamCollector(InputStreamCollector delegate) {
+ this.delegate = delegate;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void addInputStream(String filename, InputStream is) {
+ InputStream result = is.markSupported() ? is : new BufferedInputStream(is);
+
+ ZipInputStream zis = tryZip(result);
+ if (zis != null) {
+ zipContent.add(zis);
+ } else {
+ delegate.addInputStream(filename, result);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Nullable
+ private static ZipInputStream tryZip(InputStream is) {
+ try {
+ ZipInputStream zis = new ZipInputStream(is);
+ if (zis.getNextEntry() != null) {
+ is.reset();
+ return zis;
+ } else {
+ is.reset();
+ return null;
+ }
+ } catch (IOException e) {
+ return null;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void clear() throws Exception {
+ delegate.clear();
+ closeAll(zipContent);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public DeploymentUnit toDeploymentUnit() {
+ return new ZipDeploymentUnit(delegate.toDeploymentUnit(), zipContent);
+ }
+}