NIFI-15673 - Add support for Author+Committer with git-based Flow Registry Clients

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

@@ -438,6 +438,13 @@ private String createContentCloud(final GitCreateContentRequest request, final S
             multipartBuilder.addPart(FIELD_PARENTS, StandardHttpContentType.TEXT_PLAIN, expectedCommitSha.getBytes(StandardCharsets.UTF_8));
         }
 
+        final String authorName = request.getAuthorName();
+        final String authorEmail = request.getAuthorEmail();
+        if (authorName != null && authorEmail != null) {
+            final String authorValue = "%s <%s>".formatted(authorName, authorEmail);
+            multipartBuilder.addPart(FIELD_AUTHOR, StandardHttpContentType.TEXT_PLAIN, authorValue.getBytes(StandardCharsets.UTF_8));
+        }
+
         final URI uri = getRepositoryUriBuilder().addPathSegment("src").build();
         final String errorMessage = "Error while committing content for repository [%s] on branch %s at path %s"
                 .formatted(repoName, branch, resolvedPath);

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

@@ -390,10 +390,13 @@ public String createContent(final GitCreateContentRequest request) throws FlowRe
 
         final String encoded = Base64.getEncoder().encodeToString(request.getContent().getBytes(StandardCharsets.UTF_8));
 
+        final String authorName = request.getAuthorName();
+        final String authorEmail = request.getAuthorEmail();
+        final Author author = (authorName != null && authorEmail != null) ? new Author(authorName, authorEmail) : null;
+
         final PushRequest pushRequest = new PushRequest(
                 List.of(new RefUpdate(REFS_HEADS_PREFIX + branch, oldObjectId)),
-                List.of(new Commit(message,
-                        List.of(new Change(changeType, new Item(path), new NewContent(encoded, CONTENT_TYPE_BASE64)))))
+                List.of(new Commit(message, List.of(new Change(changeType, new Item(path), new NewContent(encoded, CONTENT_TYPE_BASE64))), author))
         );
 
         final String json;
@@ -440,8 +443,7 @@ public InputStream deleteContent(final String filePath, final String commitMessa
 
         final PushRequest pushRequest = new PushRequest(
                 List.of(new RefUpdate(REFS_HEADS_PREFIX + branch, oldObjectId)),
-                List.of(new Commit(commitMessage,
-                        List.of(new Change(CHANGE_TYPE_DELETE, new Item(path), null))))
+                List.of(new Commit(commitMessage, List.of(new Change(CHANGE_TYPE_DELETE, new Item(path), null)), null))
         );
 
         final String json = MAPPER.writeValueAsString(pushRequest);
@@ -491,7 +493,10 @@ private record PushRequest(List<RefUpdate> refUpdates, List<Commit> commits) { }
 
     private record RefUpdate(String name, String oldObjectId) { }
 
-    private record Commit(String comment, List<Change> changes) { }
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    private record Commit(String comment, List<Change> changes, Author author) { }
+
+    private record Author(String name, String email) { }
 
     private record Item(String path) { }
 

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

@@ -238,6 +238,7 @@ public RegisteredFlow registerFlow(final FlowRegistryClientConfigurationContext
         final FlowLocation flowLocation = new FlowLocation(branch, flow.getBucketIdentifier(), flow.getIdentifier());
         final String filePath = getSnapshotFilePath(flowLocation);
         final String commitMessage = REGISTER_FLOW_MESSAGE_FORMAT.formatted(flow.getIdentifier());
+        final String userIdentity = context.getNiFiUserIdentity().orElse(null);
 
         final Optional<String> existingFileSha = repositoryClient.getContentSha(filePath, branch);
         if (existingFileSha.isPresent()) {
@@ -258,6 +259,8 @@ public RegisteredFlow registerFlow(final FlowRegistryClientConfigurationContext
                 .path(filePath)
                 .content(flowSnapshotSerializer.serialize(flowSnapshot))
                 .message(commitMessage)
+                .authorName(userIdentity)
+                .authorEmail(userIdentity)
                 .build();
 
         repositoryClient.createContent(request);
@@ -278,7 +281,8 @@ public RegisteredFlow deregisterFlow(final FlowRegistryClientConfigurationContex
         final String branch = flowLocation.getBranch();
         final String filePath = getSnapshotFilePath(flowLocation);
         final String commitMessage = DEREGISTER_FLOW_MESSAGE_FORMAT.formatted(flowLocation.getFlowId());
-        try (final InputStream deletedSnapshotContent = repositoryClient.deleteContent(filePath, commitMessage, branch)) {
+        final String userIdentity = context.getNiFiUserIdentity().orElse(null);
+        try (final InputStream deletedSnapshotContent = repositoryClient.deleteContent(filePath, commitMessage, branch, userIdentity, userIdentity)) {
             final RegisteredFlowSnapshot deletedSnapshot = getSnapshot(deletedSnapshotContent);
             populateFlowAndSnapshotMetadata(deletedSnapshot, flowLocation);
             updateBucketReferences(repositoryClient, deletedSnapshot, flowLocation.getBucketId());
@@ -437,13 +441,17 @@ public RegisteredFlowSnapshot registerFlowSnapshot(final FlowRegistryClientConfi
         final String originalFlowContentsGroupId = replaceGroupId(flowSnapshot.getFlowContents(), FLOW_CONTENTS_GROUP_ID);
         final Position originalFlowContentsPosition = replacePosition(flowSnapshot.getFlowContents(), new Position(0, 0));
 
+        final String userIdentity = context.getNiFiUserIdentity().orElse(null);
+
         final GitCreateContentRequest createContentRequest = GitCreateContentRequest.builder()
                 .branch(branch)
                 .path(filePath)
                 .content(flowSnapshotSerializer.serialize(flowSnapshot))
                 .message(commitMessage)
                 .existingContentSha(existingBlobSha)
                 .expectedCommitSha(expectedVersion)
+                .authorName(userIdentity)
+                .authorEmail(userIdentity)
                 .build();
 
         final String createContentCommitSha = repositoryClient.createContent(createContentRequest);

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

@@ -27,6 +27,8 @@ public class GitCreateContentRequest {
     private final String message;
     private final String existingContentSha;
     private final String expectedCommitSha;
+    private final String authorName;
+    private final String authorEmail;
 
     private GitCreateContentRequest(final Builder builder) {
         this.branch = Objects.requireNonNull(builder.branch);
@@ -37,6 +39,8 @@ private GitCreateContentRequest(final Builder builder) {
         this.existingContentSha = builder.existingContentSha;
         // Commit SHA for providers that support atomic commits via commit SHA
         this.expectedCommitSha = builder.expectedCommitSha;
+        this.authorName = builder.authorName;
+        this.authorEmail = builder.authorEmail;
     }
 
     public String getBranch() {
@@ -63,6 +67,14 @@ public String getExpectedCommitSha() {
         return expectedCommitSha;
     }
 
+    public String getAuthorName() {
+        return authorName;
+    }
+
+    public String getAuthorEmail() {
+        return authorEmail;
+    }
+
     public static Builder builder() {
         return new Builder();
     }
@@ -74,6 +86,8 @@ public static final class Builder {
         private String message;
         private String existingContentSha;
         private String expectedCommitSha;
+        private String authorName;
+        private String authorEmail;
 
         public Builder branch(final String branch) {
             this.branch = branch;
@@ -105,6 +119,16 @@ public Builder expectedCommitSha(final String expectedCommitSha) {
             return this;
         }
 
+        public Builder authorName(final String authorName) {
+            this.authorName = authorName;
+            return this;
+        }
+
+        public Builder authorEmail(final String authorEmail) {
+            this.authorEmail = authorEmail;
+            return this;
+        }
+
         public GitCreateContentRequest build() {
             return new GitCreateContentRequest(this);
         }

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,25 @@ default Optional<String> getContentShaAtCommit(String path, String commitSha) th
      */
     InputStream deleteContent(String filePath, String commitMessage, String branch) throws FlowRegistryException, IOException;
 
+    /**
+     * Deletes the file at the given path on the given branch, attributing the commit to the specified author.
+     *
+     * The caller of this method is responsible for closing the returned InputStream.
+     *
+     * @param filePath the path of the file
+     * @param commitMessage the commit message
+     * @param branch the branch
+     * @param authorName the name of the commit author, or null to use the authenticated user
+     * @param authorEmail the email of the commit author, or null to use the authenticated user
+     * @return the input stream to the deleted content
+     * @throws IOException if an I/O error occurs
+     * @throws FlowRegistryException if a non-I/O error occurs
+     */
+    default InputStream deleteContent(final String filePath, final String commitMessage, final String branch,
+                                      final String authorName, final String authorEmail) throws FlowRegistryException, IOException {
+        return deleteContent(filePath, commitMessage, branch);
+    }
+
     /**
      * Closes any resources held by the client.
      *

nifi-extension-bundles/nifi-github-bundle/nifi-github-extensions/pom.xml

@@ -38,6 +38,12 @@
             <groupId>org.kohsuke</groupId>
             <artifactId>github-api</artifactId>
             <version>${github-api.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.infradna.tool</groupId>
+                    <artifactId>bridge-method-annotation</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>io.jsonwebtoken</groupId>

nifi-extension-bundles/nifi-github-bundle/nifi-github-extensions/src/main/java/org/apache/nifi/github/GitHubFlowRegistryClient.java

@@ -115,7 +115,7 @@ protected List<PropertyDescriptor> createPropertyDescriptors() {
     }
 
     @Override
-    protected GitHubRepositoryClient createRepositoryClient(final FlowRegistryClientConfigurationContext context) throws IOException, FlowRegistryException {
+    protected GitRepositoryClient createRepositoryClient(final FlowRegistryClientConfigurationContext context) throws IOException, FlowRegistryException {
         return GitHubRepositoryClient.builder()
                 .logger(getLogger())
                 .apiUrl(context.getProperty(GITHUB_API_URL).getValue())

nifi-extension-bundles/nifi-github-bundle/nifi-github-extensions/src/main/java/org/apache/nifi/github/GitHubRepositoryClient.java

@@ -29,11 +29,13 @@
 import org.apache.nifi.ssl.SSLContextProvider;
 import org.kohsuke.github.GHCommit;
 import org.kohsuke.github.GHContent;
+import org.kohsuke.github.GHContentBuilder;
 import org.kohsuke.github.GHContentUpdateResponse;
 import org.kohsuke.github.GHMyself;
 import org.kohsuke.github.GHPermissionType;
 import org.kohsuke.github.GHRef;
 import org.kohsuke.github.GHRepository;
+import org.kohsuke.github.GHUser;
 import org.kohsuke.github.GitHub;
 import org.kohsuke.github.GitHubAbuseLimitHandler;
 import org.kohsuke.github.GitHubBuilder;
@@ -48,9 +50,10 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.io.InputStream;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.net.http.HttpClient;
 import java.security.PrivateKey;
-import java.time.Instant;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -201,13 +204,20 @@ public String createContent(final GitCreateContentRequest request) throws IOExce
         logger.debug("Creating content at path [{}] on branch [{}] in repo [{}] ", resolvedPath, branch, repository.getName());
         return execute(() -> {
             try {
-                final GHContentUpdateResponse response = repository.createContent()
+                final GHContentBuilder contentBuilder = repository.createContent()
                         .branch(branch)
                         .path(resolvedPath)
                         .content(request.getContent())
                         .message(request.getMessage())
-                        .sha(request.getExistingContentSha())
-                        .commit();
+                        .sha(request.getExistingContentSha());
+
+                final String authorName = request.getAuthorName();
+                final String authorEmail = request.getAuthorEmail();
+                if (authorName != null && authorEmail != null) {
+                    setContentBuilderAuthor(contentBuilder, authorName, authorEmail);
+                }
+
+                final GHContentUpdateResponse response = contentBuilder.commit();
                 return response.getCommit().getSha();
             } catch (final FileNotFoundException fnf) {
                 throwPathOrBranchNotFound(fnf, resolvedPath, branch);
@@ -472,17 +482,37 @@ private void throwPathOrBranchNotFound(final FileNotFoundException fileNotFoundE
         throw new FlowRegistryException("Path [" + path + "] or Branch [" + branch + "] not found", fileNotFoundException);
     }
 
+    /**
+     * Sets the author on a GHContentBuilder using reflection because Hub4j does not yet expose an author() method.
+     * The internal Requester supports with(String, Map) which maps to the GitHub API's author JSON object.
+     * This workaround can be replaced with contentBuilder.author(name, email) once Hub4j adds that method.
+     */
+    private void setContentBuilderAuthor(final GHContentBuilder contentBuilder, final String authorName, final String authorEmail) {
+        try {
+            final Field reqField = GHContentBuilder.class.getDeclaredField("req");
+            reqField.setAccessible(true);
+            final Object requester = reqField.get(contentBuilder);
+            final Method withMethod = requester.getClass().getMethod("with", String.class, Map.class);
+            withMethod.setAccessible(true);
+            withMethod.invoke(requester, "author", Map.of("name", authorName, "email", authorEmail));
+        } catch (final ReflectiveOperationException e) {
+            logger.warn("Unable to set commit author via GHContentBuilder reflection; commit will use the authenticated service account as author", e);
+        }
+    }
+
     private GitCommit toGitCommit(final GHCommit ghCommit) throws IOException {
         GitCommit commit = commitCache.getIfPresent(ghCommit.getSHA1());
         if (commit != null) {
             return commit;
         } else {
             final GHCommit.ShortInfo shortInfo = ghCommit.getCommitShortInfo();
+            final GHUser ghUser = ghCommit.getAuthor();
+            final String author = ghUser != null ? ghUser.getLogin() : shortInfo.getAuthor().getName();
             commit = new GitCommit(
                     ghCommit.getSHA1(),
-                    ghCommit.getAuthor().getLogin(),
+                    author,
                     shortInfo.getMessage(),
-                    Instant.ofEpochMilli(shortInfo.getCommitDate().getTime()));
+                    shortInfo.getCommitDate());
             commitCache.put(ghCommit.getSHA1(), commit);
             return commit;
         }