NIFI-15148 - Registry Clients - Support for branch creation (backend only)

nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/main/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClient.java

@@ -50,6 +50,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalLong;
@@ -538,6 +539,85 @@ private String createContentDataCenter(final GitCreateContentRequest request, fi
         return getRequiredLatestCommit(branch, resolvedPath);
     }
 
+    @Override
+    public void createBranch(final String newBranchName, final String sourceBranch, final Optional<String> sourceCommitSha)
+            throws IOException, FlowRegistryException {
+        if (newBranchName == null || newBranchName.isBlank()) {
+            throw new IllegalArgumentException("Branch name must be specified");
+        }
+        if (sourceBranch == null || sourceBranch.isBlank()) {
+            throw new IllegalArgumentException("Source branch must be specified");
+        }
+
+        final String trimmedNewBranch = newBranchName.trim();
+        final String trimmedSourceBranch = sourceBranch.trim();
+
+        if (getBranches().contains(trimmedNewBranch)) {
+            throw new FlowRegistryException("Branch [%s] already exists".formatted(trimmedNewBranch));
+        }
+
+        logger.info("Creating branch [{}] from [{}] in repository [{}]", trimmedNewBranch, trimmedSourceBranch, repoName);
+
+        if (formFactor == BitbucketFormFactor.DATA_CENTER) {
+            createBranchDataCenter(trimmedNewBranch, trimmedSourceBranch, sourceCommitSha);
+        } else {
+            createBranchCloud(trimmedNewBranch, trimmedSourceBranch, sourceCommitSha);
+        }
+    }
+
+    private void createBranchCloud(final String newBranchName, final String sourceBranch, final Optional<String> sourceCommitSha) throws FlowRegistryException {
+        final String targetHash;
+        if (sourceCommitSha.isPresent() && !sourceCommitSha.get().isBlank()) {
+            targetHash = sourceCommitSha.get();
+        } else {
+            targetHash = getBranchHeadCloud(sourceBranch);
+        }
+
+        final URI uri = getRepositoryUriBuilder().addPathSegment("refs").addPathSegment("branches").build();
+        final String json;
+        try {
+            json = objectMapper.writeValueAsString(Map.of(FIELD_NAME, newBranchName, FIELD_TARGET, Map.of(FIELD_HASH, targetHash)));
+        } catch (final Exception e) {
+            throw new FlowRegistryException("Failed to serialize branch creation request", e);
+        }
+
+        try (final HttpResponseEntity response = this.webClient.getWebClientService()
+                .post()
+                .uri(uri)
+                .header(AUTHORIZATION_HEADER, authToken.getAuthzHeaderValue())
+                .header(CONTENT_TYPE_HEADER, "application/json")
+                .body(json)
+                .retrieve()) {
+            verifyStatusCode(response, "Error creating branch [%s] in repository [%s]".formatted(newBranchName, repoName), HttpURLConnection.HTTP_CREATED);
+        } catch (final IOException e) {
+            throw new FlowRegistryException("Failed closing Bitbucket create branch response", e);
+        }
+    }
+
+    private void createBranchDataCenter(final String newBranchName, final String sourceBranch, final Optional<String> sourceCommitSha) throws FlowRegistryException {
+        final String startPoint = sourceCommitSha.filter(sha -> !sha.isBlank()).orElse("refs/heads/" + sourceBranch);
+        final URI uri = getRepositoryUriBuilder().addPathSegment("branches").build();
+
+        final String json;
+        try {
+            json = objectMapper.writeValueAsString(Map.of(FIELD_NAME, newBranchName, "startPoint", startPoint));
+        } catch (final Exception e) {
+            throw new FlowRegistryException("Failed to serialize branch creation request", e);
+        }
+
+        try (final HttpResponseEntity response = this.webClient.getWebClientService()
+                .post()
+                .uri(uri)
+                .header(AUTHORIZATION_HEADER, authToken.getAuthzHeaderValue())
+                .header(CONTENT_TYPE_HEADER, "application/json")
+                .body(json)
+                .retrieve()) {
+            verifyStatusCode(response, "Error creating branch [%s] in repository [%s]".formatted(newBranchName, repoName), HttpURLConnection.HTTP_OK);
+        } catch (final IOException e) {
+            throw new FlowRegistryException("Failed closing Bitbucket create branch response", e);
+        }
+    }
+
     @Override
     public InputStream deleteContent(final String filePath, final String commitMessage, final String branch) throws FlowRegistryException {
         final String resolvedPath = getResolvedPath(filePath);

nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/test/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClientTest.java

@@ -35,10 +35,12 @@
 import org.mockito.stubbing.OngoingStubbing;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 import java.io.InputStream;
 import java.net.HttpURLConnection;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.util.Optional;
 import java.util.OptionalLong;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -126,14 +128,21 @@ private void stubPostChain(final HttpResponseEntity... responses) {
         postAfterHeaders = mock(HttpRequestBodySpec.class);
         lenient().when(webClientService.post()).thenReturn(postSpec);
         lenient().when(postSpec.uri(any(URI.class))).thenReturn(postBodySpec);
+        lenient().when(postBodySpec.header(anyString(), anyString())).thenReturn(postBodySpec);
         lenient().when(postBodySpec.body(any(InputStream.class), any(OptionalLong.class))).thenReturn(afterBody);
+        lenient().when(postBodySpec.body(anyString())).thenReturn(afterBody);
         lenient().when(afterBody.header(anyString(), anyString())).thenReturn(postAfterHeaders);
         lenient().when(postAfterHeaders.header(anyString(), anyString())).thenReturn(postAfterHeaders);
 
-        OngoingStubbing<HttpResponseEntity> stubbing = when(postAfterHeaders.retrieve());
+        OngoingStubbing<HttpResponseEntity> stubbing = lenient().when(postAfterHeaders.retrieve());
         for (final HttpResponseEntity response : responses) {
             stubbing = stubbing.thenReturn(response);
         }
+
+        OngoingStubbing<HttpResponseEntity> directStubbing = lenient().when(afterBody.retrieve());
+        for (final HttpResponseEntity response : responses) {
+            directStubbing = directStubbing.thenReturn(response);
+        }
     }
 
     private HttpResponseEntity mockResponse(final int statusCode, final String body) {
@@ -275,6 +284,52 @@ void testCreateContentCloudNullExpectedCommitSha() throws FlowRegistryException
         assertEquals(RESULT_COMMIT_SHA, commitSha);
     }
 
+    @Test
+    void testCreateBranchCloudSuccess() throws FlowRegistryException, IOException {
+        stubGetChain(
+                branchListResponse(),
+                branchListResponse(),
+                branchHeadResponse(BRANCH_HEAD_SHA)
+        );
+        stubPostChain(createdResponse());
+
+        final BitbucketRepositoryClient client = buildCloudClient();
+        client.createBranch("feature", "main", Optional.empty());
+    }
+
+    @Test
+    void testCreateBranchCloudWithCommitSha() throws FlowRegistryException, IOException {
+        stubGetChain(branchListResponse(), branchListResponse());
+        stubPostChain(createdResponse());
+
+        final BitbucketRepositoryClient client = buildCloudClient();
+        client.createBranch("feature", "main", Optional.of("abc123"));
+    }
+
+    @Test
+    void testCreateBranchCloudAlreadyExists() throws FlowRegistryException {
+        stubGetChain(branchListResponse(), branchListResponse());
+
+        final BitbucketRepositoryClient client = buildCloudClient();
+        final FlowRegistryException exception = assertThrows(FlowRegistryException.class,
+                () -> client.createBranch("main", "main", Optional.empty()));
+        assertTrue(exception.getMessage().contains("already exists"));
+    }
+
+    @Test
+    void testCreateBranchBlankNameRejected() throws FlowRegistryException {
+        stubGetChain(branchListResponse());
+        final BitbucketRepositoryClient client = buildCloudClient();
+        assertThrows(IllegalArgumentException.class, () -> client.createBranch("  ", "main", Optional.empty()));
+    }
+
+    @Test
+    void testCreateBranchBlankSourceRejected() throws FlowRegistryException {
+        stubGetChain(branchListResponse());
+        final BitbucketRepositoryClient client = buildCloudClient();
+        assertThrows(IllegalArgumentException.class, () -> client.createBranch("feature", "  ", Optional.empty()));
+    }
+
     @Test
     void testCreateContentDataCenterUnchanged() throws FlowRegistryException {
         final HttpRequestUriSpec getSpec = mock(HttpRequestUriSpec.class);

nifi-extension-bundles/nifi-azure-bundle/nifi-azure-registry-clients/src/main/java/org/apache/nifi/azure/devops/AzureDevOpsRepositoryClient.java

@@ -121,6 +121,7 @@ public class AzureDevOpsRepositoryClient implements GitRepositoryClient {
     private static final String CHANGE_TYPE_DELETE = "delete";
     private static final String CONTENT_TYPE_BASE64 = "base64encoded";
     private static final int MAX_PUSH_ATTEMPTS = 3;
+    private static final String ZERO_OBJECT_ID = "0000000000000000000000000000000000000000";
 
     // Common query parameter names and values
     private static final String VERSION_DESCRIPTOR_VERSION = "versionDescriptor.version";
@@ -443,6 +444,75 @@ private HttpResponseEntity executePush(final URI pushUri, final String branch, f
                 .retrieve();
     }
 
+    @Override
+    public void createBranch(final String newBranchName, final String sourceBranch, final Optional<String> sourceCommitSha)
+            throws IOException, FlowRegistryException {
+        if (newBranchName == null || newBranchName.isBlank()) {
+            throw new IllegalArgumentException("Branch name must be specified");
+        }
+        if (sourceBranch == null || sourceBranch.isBlank()) {
+            throw new IllegalArgumentException("Source branch must be specified");
+        }
+
+        final String trimmedNewBranch = newBranchName.trim();
+        final String trimmedSourceBranch = sourceBranch.trim();
+
+        if (branchExists(trimmedNewBranch)) {
+            throw new FlowRegistryException("Branch [%s] already exists".formatted(trimmedNewBranch));
+        }
+
+        final String baseCommitSha;
+        if (sourceCommitSha.isPresent() && !sourceCommitSha.get().isBlank()) {
+            baseCommitSha = sourceCommitSha.get();
+        } else {
+            baseCommitSha = fetchBranchHead(trimmedSourceBranch);
+        }
+
+        logger.info("Creating branch [{}] from [{}] at commit [{}] in repo [{}]", trimmedNewBranch, trimmedSourceBranch, baseCommitSha, repoName);
+
+        final URI refsUri = getUriBuilder().addPathSegment(SEGMENT_REFS)
+                .addQueryParameter(API, API_VERSION)
+                .build();
+
+        final String json;
+        try {
+            json = MAPPER.writeValueAsString(List.of(new CreateRefRequest(REFS_HEADS_PREFIX + trimmedNewBranch, ZERO_OBJECT_ID, baseCommitSha)));
+        } catch (final Exception e) {
+            throw new FlowRegistryException("Failed to serialize branch creation request", e);
+        }
+
+        final HttpResponseEntity response = this.webClient.getWebClientService()
+                .post()
+                .uri(refsUri)
+                .header(AUTHORIZATION_HEADER, bearerToken())
+                .header(CONTENT_TYPE_HEADER, MediaType.APPLICATION_JSON.getMediaType())
+                .body(json)
+                .retrieve();
+
+        if (response.statusCode() != HttpURLConnection.HTTP_OK) {
+            throw new FlowRegistryException("Failed to create branch [%s] in repo [%s] - %s".formatted(trimmedNewBranch, repoName, getErrorMessage(response)));
+        }
+    }
+
+    private boolean branchExists(final String branchName) throws FlowRegistryException {
+        final URI refUri = getUriBuilder().addPathSegment(SEGMENT_REFS)
+                .addQueryParameter(PARAM_FILTER, FILTER_HEADS_PREFIX + branchName)
+                .addQueryParameter(API, API_VERSION)
+                .build();
+        final JsonNode refResponse = executeGet(refUri);
+        final JsonNode values = refResponse.get(JSON_FIELD_VALUE);
+        if (values == null || !values.isArray()) {
+            return false;
+        }
+        for (final JsonNode ref : values) {
+            final String refName = ref.get(JSON_FIELD_NAME).asText();
+            if (refName.equals(REFS_HEADS_PREFIX + branchName)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     @Override
     public InputStream deleteContent(final String filePath, final String commitMessage, final String branch) throws FlowRegistryException, IOException {
         final String path = getResolvedPath(filePath);
@@ -515,6 +585,8 @@ private record Change(String changeType, Item item, NewContent newContent) { }
 
     private record NewContent(String content, String contentType) { }
 
+    private record CreateRefRequest(String name, String oldObjectId, String newObjectId) { }
+
     /**
      * Create URI builder for accessing the repository.
      *

nifi-extension-bundles/nifi-azure-bundle/nifi-azure-registry-clients/src/test/java/org/apache/nifi/azure/devops/AzureDevOpsRepositoryClientTest.java

@@ -42,6 +42,7 @@
 import java.net.HttpURLConnection;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.util.Optional;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -303,4 +304,50 @@ void testDeleteContentUsesFetchBranchHead() throws FlowRegistryException, IOExce
             assertTrue(body.contains("delete-commit"));
         }
     }
+
+    @Test
+    void testCreateBranchSuccess() throws FlowRegistryException, IOException {
+        final HttpResponseEntity emptyRefsResponse = mockResponse(HttpURLConnection.HTTP_OK, "{\"value\":[]}");
+        stubGetChain(repoInfoResponse(), permissionsResponse(), emptyRefsResponse, branchHeadResponse(BRANCH_HEAD_SHA));
+        stubPostChain(mockResponse(HttpURLConnection.HTTP_OK, "{\"value\":[{\"name\":\"refs/heads/feature\"}]}"));
+
+        final AzureDevOpsRepositoryClient client = buildClient();
+        client.createBranch("feature", "main", Optional.empty());
+    }
+
+    @Test
+    void testCreateBranchWithSourceCommitSha() throws FlowRegistryException, IOException {
+        final HttpResponseEntity emptyRefsResponse = mockResponse(HttpURLConnection.HTTP_OK, "{\"value\":[]}");
+        stubGetChain(repoInfoResponse(), permissionsResponse(), emptyRefsResponse);
+        stubPostChain(mockResponse(HttpURLConnection.HTTP_OK, "{\"value\":[{\"name\":\"refs/heads/feature\"}]}"));
+
+        final AzureDevOpsRepositoryClient client = buildClient();
+        client.createBranch("feature", "main", Optional.of("abc123"));
+    }
+
+    @Test
+    void testCreateBranchAlreadyExists() throws FlowRegistryException {
+        final HttpResponseEntity existingBranchResponse = mockResponse(HttpURLConnection.HTTP_OK,
+                "{\"value\":[{\"name\":\"refs/heads/feature\",\"objectId\":\"abc123\"}]}");
+        stubGetChain(repoInfoResponse(), permissionsResponse(), existingBranchResponse);
+
+        final AzureDevOpsRepositoryClient client = buildClient();
+        final FlowRegistryException exception = assertThrows(FlowRegistryException.class,
+                () -> client.createBranch("feature", "main", Optional.empty()));
+        assertTrue(exception.getMessage().contains("already exists"));
+    }
+
+    @Test
+    void testCreateBranchBlankNameRejected() throws FlowRegistryException {
+        stubGetChain(repoInfoResponse(), permissionsResponse());
+        final AzureDevOpsRepositoryClient client = buildClient();
+        assertThrows(IllegalArgumentException.class, () -> client.createBranch("  ", "main", Optional.empty()));
+    }
+
+    @Test
+    void testCreateBranchBlankSourceRejected() throws FlowRegistryException {
+        stubGetChain(repoInfoResponse(), permissionsResponse());
+        final AzureDevOpsRepositoryClient client = buildClient();
+        assertThrows(IllegalArgumentException.class, () -> client.createBranch("feature", "  ", Optional.empty()));
+    }
 }

nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java

@@ -208,6 +208,42 @@ public FlowRegistryBranch getDefaultBranch(final FlowRegistryClientConfiguration
         return defaultBranch;
     }
 
+    @Override
+    public void createBranch(final FlowRegistryClientConfigurationContext context, final FlowVersionLocation sourceLocation, final String newBranchName)
+            throws FlowRegistryException, IOException {
+        if (StringUtils.isBlank(newBranchName)) {
+            throw new IllegalArgumentException("Branch name must be specified when creating a new branch");
+        }
+
+        final GitRepositoryClient repositoryClient = getRepositoryClient(context);
+        verifyWritePermissions(repositoryClient);
+
+        final String sourceBranch = resolveSourceBranch(context, sourceLocation);
+        if (StringUtils.isBlank(sourceBranch)) {
+            throw new FlowRegistryException("Unable to determine source branch for new branch creation");
+        }
+
+        final Optional<String> sourceCommitSha = sourceLocation == null ? Optional.empty() : Optional.ofNullable(sourceLocation.getVersion());
+        final String trimmedBranchName = newBranchName.trim();
+        final String trimmedSourceBranch = sourceBranch.trim();
+
+        getLogger().info("Creating branch [{}] from branch [{}]", trimmedBranchName, trimmedSourceBranch);
+
+        try {
+            repositoryClient.createBranch(trimmedBranchName, trimmedSourceBranch, sourceCommitSha);
+        } catch (final UnsupportedOperationException e) {
+            throw new FlowRegistryException("Configured repository client does not support branch creation", e);
+        }
+    }
+
+    private String resolveSourceBranch(final FlowRegistryClientConfigurationContext context, final FlowVersionLocation sourceLocation) {
+        if (sourceLocation != null && StringUtils.isNotBlank(sourceLocation.getBranch())) {
+            return sourceLocation.getBranch();
+        }
+        final String defaultBranch = context.getProperty(REPOSITORY_BRANCH).getValue();
+        return StringUtils.isBlank(defaultBranch) ? null : defaultBranch;
+    }
+
     @Override
     public Set<FlowRegistryBucket> getBuckets(final FlowRegistryClientConfigurationContext context, final String branch) throws IOException, FlowRegistryException {
         final GitRepositoryClient repositoryClient = getRepositoryClient(context);

nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/client/GitRepositoryClient.java

@@ -155,6 +155,21 @@ default Optional<String> getContentShaAtCommit(String path, String commitSha) th
      */
     InputStream deleteContent(String filePath, String commitMessage, String branch) throws FlowRegistryException, IOException;
 
+    /**
+     * Creates a new branch in the repository.
+     *
+     * @param newBranchName the name of the branch to create
+     * @param sourceBranch the name of the source branch
+     * @param sourceCommitSha optional commit SHA to use as the starting point for the new branch. If empty, the head commit of the source branch should be used.
+     * @throws IOException if an I/O error occurs
+     * @throws FlowRegistryException if a non-I/O error occurs
+     * @throws UnsupportedOperationException if the repository implementation does not support branch creation
+     */
+    default void createBranch(final String newBranchName, final String sourceBranch, final Optional<String> sourceCommitSha)
+            throws IOException, FlowRegistryException {
+        throw new UnsupportedOperationException("Branch creation is not supported");
+    }
+
     /**
      * Closes any resources held by the client.
      *