From 08a92fb13049711c918339c7502d319d3b0e4ebd Mon Sep 17 00:00:00 2001 From: Jannik Rebmann Date: Wed, 23 Sep 2026 13:45:35 +0200 Subject: [PATCH] NIFI-16359 Cache commits per (path, branch) to remove per-process-group API load --- .../nifi-git-flow-registry/pom.xml | 4 + .../git/AbstractGitFlowRegistryClient.java | 76 ++++++++++++- .../AbstractGitFlowRegistryClientTest.java | 105 +++++++++++++++++- .../nifi/gitlab/GitLabFlowRegistryClient.java | 3 +- .../nifi/gitlab/GitLabRepositoryClient.java | 3 +- 5 files changed, 180 insertions(+), 11 deletions(-) diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/pom.xml b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/pom.xml index a38528a49c0a..4a3da1e8a031 100644 --- a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/pom.xml +++ b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/pom.xml @@ -24,6 +24,10 @@ jar + + com.github.ben-manes.caffeine + caffeine + com.fasterxml.jackson.module jackson-module-jakarta-xmlbind-annotations diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java index 318e1597ca2e..50bc449d30e0 100644 --- a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java +++ b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/main/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClient.java @@ -17,10 +17,13 @@ package org.apache.nifi.registry.flow.git; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import org.apache.nifi.components.ConfigVerificationResult; import org.apache.nifi.components.ConfigVerificationResult.Outcome; import org.apache.nifi.components.DescribedValue; import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; import org.apache.nifi.components.ValidationContext; import org.apache.nifi.components.ValidationResult; import org.apache.nifi.flow.ConnectableComponent; @@ -68,6 +71,8 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.CompletionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import java.util.regex.Pattern; @@ -130,6 +135,13 @@ public abstract class AbstractGitFlowRegistryClient extends AbstractFlowRegistry .required(true) .build(); + public static final PropertyDescriptor COMMIT_CACHE_TTL = new PropertyDescriptor.Builder() + .name("Commit Cache TTL") + .description("Specifies the maximum staleness window for detecting a new remote version. Leave blank to disable commit caching.") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .required(false) + .build(); + static final String DEFAULT_BUCKET_NAME = "default"; static final String DEFAULT_BUCKET_KEEP_FILE_PATH = DEFAULT_BUCKET_NAME + "/.keep"; static final String DEFAULT_BUCKET_KEEP_FILE_CONTENT = "Do Not Delete"; @@ -143,6 +155,10 @@ public abstract class AbstractGitFlowRegistryClient extends AbstractFlowRegistry static final String SNAPSHOT_FILE_PATH_FORMAT = "%s/%s" + SNAPSHOT_FILE_EXTENSION; static final String FLOW_CONTENTS_GROUP_ID = "flow-contents-group"; + static final long COMMIT_CACHE_MAX_ENTRIES = 1000; + + private volatile Cache> commitCache; + private volatile FlowSnapshotSerializer flowSnapshotSerializer; private volatile GitRepositoryClient repositoryClient; private volatile Pattern directoryExclusionPattern; @@ -160,6 +176,7 @@ public void initialize(final FlowRegistryClientInitializationContext context) { combinedPropertyDescriptors.add(DIRECTORY_FILTER_EXCLUDE); combinedPropertyDescriptors.add(PARAMETER_CONTEXT_VALUES); combinedPropertyDescriptors.add(COMMIT_AUTHOR_SOURCE); + combinedPropertyDescriptors.add(COMMIT_CACHE_TTL); combinedPropertyDescriptors.add(SSL_CONTEXT_SERVICE); combinedPropertyDescriptors.add(SYNCHRONIZATION_INTERVAL); propertyDescriptors = Collections.unmodifiableList(combinedPropertyDescriptors); @@ -313,6 +330,7 @@ public RegisteredFlow registerFlow(final FlowRegistryClientConfigurationContext .build(); repositoryClient.createContent(request); + invalidateCommitCache(filePath, branch); // Re-populate fields before returning flow.setBucketName(originalBucketId); @@ -332,6 +350,7 @@ public RegisteredFlow deregisterFlow(final FlowRegistryClientConfigurationContex final String commitMessage = DEREGISTER_FLOW_MESSAGE_FORMAT.formatted(flowLocation.getFlowId()); final String userIdentity = resolveAuthorIdentity(context); try (final InputStream deletedSnapshotContent = repositoryClient.deleteContent(filePath, commitMessage, branch, userIdentity, userIdentity)) { + invalidateCommitCache(filePath, branch); final RegisteredFlowSnapshot deletedSnapshot = getSnapshot(deletedSnapshotContent); populateFlowAndSnapshotMetadata(deletedSnapshot, flowLocation); updateBucketReferences(repositoryClient, deletedSnapshot, flowLocation.getBucketId()); @@ -412,9 +431,9 @@ public RegisteredFlowSnapshot registerFlowSnapshot(final FlowRegistryClientConfi // Capture the expected version before any modifications - this is the commit SHA the user believes they are committing on top of final String expectedVersion = snapshotMetadata.getVersion(); - // Get the current version (latest commit SHA) from the repository - final List commits = repositoryClient.getCommits(filePath, branch); - final String currentVersion = commits.isEmpty() ? null : commits.getFirst().id(); + // Get the current version (latest commit SHA) directly from the repository so a stale cache cannot incorrectly + // permit a write when another user has already committed a newer version. + final String currentVersion = getCommitsWithoutCache(filePath, branch).stream().findFirst().map(GitCommit::id).orElse(null); // Check for version conflict: if the user expects a specific version but it doesn't match the current version in the repository, // another user may have committed changes in the meantime. Reject the commit unless FORCE_COMMIT is specified. @@ -509,6 +528,7 @@ public RegisteredFlowSnapshot registerFlowSnapshot(final FlowRegistryClientConfi } else if (createContentCommitSha.isEmpty() || createContentCommitSha.isBlank()) { throw new FlowRegistryException("Created Content Commit SHA is empty"); } + invalidateCommitCache(filePath, branch); final VersionedFlowCoordinates versionedFlowCoordinates = new VersionedFlowCoordinates(); versionedFlowCoordinates.setRegistryId(getIdentifier()); @@ -544,7 +564,7 @@ public Set getFlowVersions(final FlowRegistryCli final String filePath = getSnapshotFilePath(flowLocation); final Set snapshotMetadataSet = new LinkedHashSet<>(); - for (final GitCommit commit : repositoryClient.getCommits(filePath, branch)) { + for (final GitCommit commit : getCommitsCached(filePath, branch)) { final RegisteredFlowSnapshotMetadata snapshotMetadata = createSnapshotMetadata(commit, flowLocation); if (snapshotMetadata.getComments() != null && snapshotMetadata.getComments().startsWith(REGISTER_FLOW_MESSAGE_PREFIX)) { continue; @@ -562,7 +582,7 @@ public Optional getLatestVersion(final FlowRegistryClientConfigurationCo final String branch = flowLocation.getBranch(); final String filePath = getSnapshotFilePath(flowLocation); - final List commits = repositoryClient.getCommits(filePath, branch); + final List commits = getCommitsCached(filePath, branch); final String latestVersion = commits.isEmpty() ? null : commits.getFirst().id(); return Optional.ofNullable(latestVersion); } @@ -600,6 +620,43 @@ private RegisteredFlowSnapshotMetadata createSnapshotMetadata(final GitCommit co return snapshotMetadata; } + private List getCommitsWithoutCache(final String filePath, final String branch) throws FlowRegistryException, IOException { + return repositoryClient.getCommits(filePath, branch); + } + + private List getCommitsCached(final String filePath, final String branch) throws FlowRegistryException, IOException { + final String key = filePath + "\n" + branch; + final Cache> cache = commitCache; + if (cache == null) { + return repositoryClient.getCommits(filePath, branch); + } + try { + return cache.get(key, k -> { + try { + return List.copyOf(repositoryClient.getCommits(filePath, branch)); + } catch (final FlowRegistryException | IOException e) { + throw new CompletionException(e); + } + }); + } catch (final CompletionException e) { + final Throwable cause = e.getCause(); + if (cause instanceof FlowRegistryException fre) { + throw fre; + } + if (cause instanceof IOException ioe) { + throw ioe; + } + throw new FlowRegistryException("Failed to list commits for " + filePath + " on " + branch, cause); + } + } + + private void invalidateCommitCache(final String filePath, final String branch) { + final Cache> cache = commitCache; + if (cache != null) { + cache.invalidate(filePath + "\n" + branch); + } + } + private RegisteredFlow mapToRegisteredFlow(final BucketLocation bucketLocation, final String filename) { final String branch = bucketLocation.getBranch(); final String bucketId = bucketLocation.getBucketId(); @@ -732,6 +789,15 @@ private void verifyReadPermissions(final GitRepositoryClient repositoryClient) t protected synchronized GitRepositoryClient getRepositoryClient(final FlowRegistryClientConfigurationContext context) throws IOException, FlowRegistryException { if (!clientInitialized.get()) { getLogger().info("Initializing repository client"); + if (commitCache != null) { + commitCache.invalidateAll(); + } + final PropertyValue commitCacheTtl = context.getProperty(COMMIT_CACHE_TTL); + final String cacheTtl = commitCacheTtl == null ? null : commitCacheTtl.getValue(); + commitCache = StringUtils.isBlank(cacheTtl) ? null : Caffeine.newBuilder() + .expireAfterWrite(commitCacheTtl.asTimePeriod(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS) + .maximumSize(COMMIT_CACHE_MAX_ENTRIES) + .build(); repositoryClient = createRepositoryClient(context); initializeDefaultBucket(context); directoryExclusionPattern = Pattern.compile(context.getProperty(DIRECTORY_FILTER_EXCLUDE).getValue()); diff --git a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/test/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClientTest.java b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/test/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClientTest.java index ec98d490beeb..6400a941f707 100644 --- a/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/test/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClientTest.java +++ b/nifi-extension-bundles/nifi-extension-utils/nifi-git-flow-registry/src/test/java/org/apache/nifi/registry/flow/git/AbstractGitFlowRegistryClientTest.java @@ -20,11 +20,17 @@ import org.apache.nifi.components.ConfigVerificationResult.Outcome; import org.apache.nifi.components.PropertyDescriptor; import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.flow.VersionedProcessGroup; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.registry.flow.FlowLocation; import org.apache.nifi.registry.flow.FlowRegistryClientConfigurationContext; import org.apache.nifi.registry.flow.FlowRegistryClientInitializationContext; import org.apache.nifi.registry.flow.FlowRegistryException; import org.apache.nifi.registry.flow.FlowVersionLocation; +import org.apache.nifi.registry.flow.RegisterAction; +import org.apache.nifi.registry.flow.RegisteredFlow; +import org.apache.nifi.registry.flow.RegisteredFlowSnapshot; +import org.apache.nifi.registry.flow.RegisteredFlowSnapshotMetadata; import org.apache.nifi.registry.flow.git.client.GitCommit; import org.apache.nifi.registry.flow.git.client.GitCreateContentRequest; import org.apache.nifi.registry.flow.git.client.GitRepositoryClient; @@ -34,6 +40,7 @@ import java.io.IOException; import java.io.InputStream; +import java.time.Instant; import java.util.List; import java.util.Map; import java.util.Objects; @@ -161,10 +168,88 @@ void createBranchUnsupportedThrowsFlowRegistryException() throws Exception { assertTrue(repositoryClient.getCreatedBranchCommit().isEmpty()); } + @Test + void commitsAreCachedWithinTtl() throws Exception { + final TestGitRepositoryClient repositoryClient = new TestGitRepositoryClient(true, true, Set.of("bucket-a")); + final TestGitFlowRegistryClient flowRegistryClient = new TestGitFlowRegistryClient(() -> repositoryClient, "git@example.git"); + flowRegistryClient.initialize(createInitializationContext()); + final FlowRegistryClientConfigurationContext context = createContext("main", "[.].*", "30 sec"); + final FlowLocation flowLocation = new FlowLocation("main", "bucket-a", "flow-x"); + + assertEquals(Optional.of("commit-1"), flowRegistryClient.getLatestVersion(context, flowLocation)); + assertEquals(Optional.of("commit-1"), flowRegistryClient.getLatestVersion(context, flowLocation)); + + assertEquals(1, repositoryClient.getCommitsInvocationCount()); + } + + @Test + void conflictCheckReadsFreshRepositoryStateAfterCachePopulation() throws Exception { + final TestGitRepositoryClient repositoryClient = new TestGitRepositoryClient(true, true, Set.of("bucket-a")); + repositoryClient.setCurrentCommit("commit-1"); + final TestGitFlowRegistryClient flowRegistryClient = new TestGitFlowRegistryClient(() -> repositoryClient, "git@example.git"); + flowRegistryClient.initialize(createInitializationContext()); + final FlowRegistryClientConfigurationContext context = createContext("main", "[.].*", "30 sec"); + + final RegisteredFlowSnapshot snapshot = new RegisteredFlowSnapshot(); + final RegisteredFlowSnapshotMetadata metadata = new RegisteredFlowSnapshotMetadata(); + metadata.setBranch("main"); + metadata.setBucketIdentifier("bucket-a"); + metadata.setFlowIdentifier("flow-x"); + metadata.setVersion("commit-1"); + snapshot.setSnapshotMetadata(metadata); + snapshot.setFlow(new RegisteredFlow()); + snapshot.setFlowContents(new VersionedProcessGroup()); + + final FlowLocation flowLocation = new FlowLocation("main", "bucket-a", "flow-x"); + assertEquals(Optional.of("commit-1"), flowRegistryClient.getLatestVersion(context, flowLocation)); + + repositoryClient.setCurrentCommit("commit-2"); + final FlowRegistryException conflict = assertThrows(FlowRegistryException.class, + () -> flowRegistryClient.registerFlowSnapshot(context, snapshot, RegisterAction.COMMIT)); + + assertTrue(conflict.getMessage().contains("Version conflict detected")); + assertTrue(conflict.getMessage().contains("commit-1")); + assertTrue(conflict.getMessage().contains("commit-2")); + + repositoryClient.setCurrentCommit("commit-3"); + final RegisteredFlow flow = new RegisteredFlow(); + flow.setIdentifier("flow-x"); + flow.setBucketIdentifier("bucket-a"); + flow.setBranch("main"); + flowRegistryClient.registerFlow(context, flow); + + assertEquals(Optional.of("commit-3"), flowRegistryClient.getLatestVersion(context, flowLocation)); + } + + @Test + void successfulCreateContentInvalidatesCommits() throws Exception { + final TestGitRepositoryClient repositoryClient = new TestGitRepositoryClient(true, true, Set.of("bucket-a")); + final TestGitFlowRegistryClient flowRegistryClient = new TestGitFlowRegistryClient(() -> repositoryClient, "git@example.git"); + flowRegistryClient.initialize(createInitializationContext()); + final FlowRegistryClientConfigurationContext context = createContext("main", "[.].*", "30 sec"); + final FlowLocation flowLocation = new FlowLocation("main", "bucket-a", "flow-x"); + + flowRegistryClient.getLatestVersion(context, flowLocation); + final RegisteredFlow flow = new RegisteredFlow(); + flow.setIdentifier("flow-x"); + flow.setBucketIdentifier("bucket-a"); + flow.setBranch("main"); + flowRegistryClient.registerFlow(context, flow); + flowRegistryClient.getLatestVersion(context, flowLocation); + + assertEquals(2, repositoryClient.getCommitsInvocationCount()); + } + private FlowRegistryClientConfigurationContext createContext(final String branch, final String exclusionPattern) { + return createContext(branch, exclusionPattern, null); + } + + private FlowRegistryClientConfigurationContext createContext(final String branch, final String exclusionPattern, final String commitCacheTtl) { final Map properties = Map.of( AbstractGitFlowRegistryClient.REPOSITORY_BRANCH, new MockPropertyValue(branch), - AbstractGitFlowRegistryClient.DIRECTORY_FILTER_EXCLUDE, new MockPropertyValue(exclusionPattern) + AbstractGitFlowRegistryClient.DIRECTORY_FILTER_EXCLUDE, new MockPropertyValue(exclusionPattern), + AbstractGitFlowRegistryClient.COMMIT_AUTHOR_SOURCE, new MockPropertyValue("SERVICE_USER"), + AbstractGitFlowRegistryClient.COMMIT_CACHE_TTL, new MockPropertyValue(commitCacheTtl) ); return new FlowRegistryClientConfigurationContext() { @@ -253,6 +338,8 @@ private static class TestGitRepositoryClient implements GitRepositoryClient { private String createdBranch; private String createdBranchSource; private Optional createdBranchCommit = Optional.empty(); + private String currentCommit = "commit-1"; + private int commitsInvocationCount; TestGitRepositoryClient(final boolean canRead, final boolean canWrite, final Set bucketNames) { this.canRead = canRead; @@ -290,6 +377,14 @@ Optional getCreatedBranchCommit() { return createdBranchCommit; } + int getCommitsInvocationCount() { + return commitsInvocationCount; + } + + void setCurrentCommit(final String currentCommit) { + this.currentCommit = currentCommit; + } + boolean isClosed() { return closed; } @@ -347,7 +442,8 @@ public Set getFileNames(final String directory, final String branch) { @Override public List getCommits(final String path, final String branch) { - throw new UnsupportedOperationException("Not required for test"); + commitsInvocationCount++; + return List.of(new GitCommit(currentCommit, "author", "message", Instant.EPOCH)); } @Override @@ -362,7 +458,7 @@ public InputStream getContentFromCommit(final String path, final String commitSh @Override public Optional getContentSha(final String path, final String branch) { - throw new UnsupportedOperationException("Not required for test"); + return Optional.empty(); } @Override @@ -372,7 +468,8 @@ public Optional getContentShaAtCommit(final String path, final String co @Override public String createContent(final GitCreateContentRequest request) { - return "test-commit"; + currentCommit = "commit-3"; + return currentCommit; } @Override diff --git a/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabFlowRegistryClient.java b/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabFlowRegistryClient.java index 20af7b1a38e7..632f54c80979 100644 --- a/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabFlowRegistryClient.java +++ b/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabFlowRegistryClient.java @@ -32,7 +32,8 @@ import java.util.concurrent.TimeUnit; @Tags({"git", "gitlab", "registry", "flow"}) -@CapabilityDescription("Flow Registry Client that uses the GitLab REST API to version control flows in a GitLab Project.") +@CapabilityDescription("Flow Registry Client that uses the GitLab REST API to version control flows in a GitLab Project. " + + "Note that for a given flow, the registry client will retrieve at most the last 10 commits to limit API calls.") public class GitLabFlowRegistryClient extends AbstractGitFlowRegistryClient { static final PropertyDescriptor GITLAB_API_URL = new PropertyDescriptor.Builder() diff --git a/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabRepositoryClient.java b/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabRepositoryClient.java index 7e5db4d4d19e..aea6064589f3 100644 --- a/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabRepositoryClient.java +++ b/nifi-extension-bundles/nifi-gitlab-bundle/nifi-gitlab-extensions/src/main/java/org/apache/nifi/gitlab/GitLabRepositoryClient.java @@ -81,6 +81,7 @@ public class GitLabRepositoryClient implements GitRepositoryClient { private static final String DIRECTORY_MODE = "040000"; private static final int DEFAULT_ITEMS_PER_PAGE = 100; + private static final int COMMIT_PAGE_SIZE = 10; private static final TokenInfo UNKNOWN_TOKEN = new TokenInfo("unknown", false, false); @@ -212,7 +213,7 @@ public List getCommits(final String path, final String branch) throws logger.debug("Getting commits for path [{}] on branch [{}] in repository [{}]", resolvedPath, branch, projectPath); final CommitsApi commitsApi = gitLab.getCommitsApi(); - return execute(() -> commitsApi.getCommits(projectPath, branch, resolvedPath).stream() + return execute(() -> commitsApi.getCommits(projectPath, branch, null, null, resolvedPath, COMMIT_PAGE_SIZE).next().stream() .map(this::toGitCommit) .toList() );