This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new eee7824b3e3 NIFI-16537 Fixed BitbucketFlowRegistryClient failing to
commit large flow to Bitbucket Cloud (#11876)
eee7824b3e3 is described below
commit eee7824b3e3b81b9db6cd6c0e3480356c46727f5
Author: Pierre Villard <[email protected]>
AuthorDate: Mon Nov 2 22:25:34 2026 +0100
NIFI-16537 Fixed BitbucketFlowRegistryClient failing to commit large flow
to Bitbucket Cloud (#11876)
Signed-off-by: David Handermann <[email protected]>
---
.../bitbucket/BitbucketRepositoryClient.java | 41 ++++++++++++----
.../bitbucket/BitbucketRepositoryClientTest.java | 55 ++++++++++++++++++----
2 files changed, 77 insertions(+), 19 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/main/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClient.java
b/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/main/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClient.java
index f5e2a9b0308..23f071b3ebf 100644
---
a/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/main/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClient.java
+++
b/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/main/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClient.java
@@ -26,6 +26,7 @@ import org.apache.nifi.registry.flow.FlowRegistryException;
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;
+import org.apache.nifi.stream.io.LimitingInputStream;
import org.apache.nifi.web.client.api.HttpResponseEntity;
import org.apache.nifi.web.client.api.HttpUriBuilder;
import org.apache.nifi.web.client.api.StandardHttpContentType;
@@ -103,6 +104,8 @@ public class BitbucketRepositoryClient implements
GitRepositoryClient {
private static final String ENTRY_FILE_DATA_CENTER = "FILE";
private static final String ENTRY_FILE_CLOUD = "commit_file";
private static final int MAX_PUSH_ATTEMPTS = 3;
+ private static final int MAX_ERROR_RESPONSE_LENGTH = 512;
+ private static final String UNKNOWN_CONTENT_TYPE = "unknown";
private final ObjectMapper objectMapper = JsonMapper.builder().build();
@@ -433,7 +436,7 @@ public class BitbucketRepositoryClient implements
GitRepositoryClient {
final String branchHead = getBranchHeadCloud(branch);
final StandardMultipartFormDataStreamBuilder multipartBuilder =
new StandardMultipartFormDataStreamBuilder();
- multipartBuilder.addPart(resolvedPath,
StandardHttpContentType.APPLICATION_JSON,
request.getContent().getBytes(StandardCharsets.UTF_8));
+ multipartBuilder.addPart(resolvedPath, getFileName(resolvedPath),
StandardHttpContentType.APPLICATION_JSON,
request.getContent().getBytes(StandardCharsets.UTF_8));
multipartBuilder.addPart(FIELD_MESSAGE,
StandardHttpContentType.TEXT_PLAIN,
request.getMessage().getBytes(StandardCharsets.UTF_8));
multipartBuilder.addPart(FIELD_BRANCH,
StandardHttpContentType.TEXT_PLAIN, branch.getBytes(StandardCharsets.UTF_8));
multipartBuilder.addPart(FIELD_PARENTS,
StandardHttpContentType.TEXT_PLAIN,
branchHead.getBytes(StandardCharsets.UTF_8));
@@ -445,10 +448,11 @@ public class BitbucketRepositoryClient implements
GitRepositoryClient {
multipartBuilder.addPart(FIELD_AUTHOR,
StandardHttpContentType.TEXT_PLAIN,
authorValue.getBytes(StandardCharsets.UTF_8));
}
+ final byte[] requestBody = toByteArray(multipartBuilder.build());
final HttpResponseEntity response =
this.webClient.getWebClientService()
.post()
.uri(uri)
- .body(multipartBuilder.build(), OptionalLong.empty())
+ .body(new ByteArrayInputStream(requestBody),
OptionalLong.of(requestBody.length))
.header(AUTHORIZATION_HEADER,
authToken.getAuthzHeaderValue())
.header(CONTENT_TYPE_HEADER,
multipartBuilder.getHttpContentType().getContentType())
.retrieve();
@@ -639,10 +643,11 @@ public class BitbucketRepositoryClient implements
GitRepositoryClient {
final URI uri =
getRepositoryUriBuilder().addPathSegment("src").build();
final String errorMessage = "Error while deleting content for
repository [%s] on branch %s at path %s"
.formatted(repoName, branch, resolvedPath);
+ final byte[] requestBody = toByteArray(multipartBuilder.build());
try (final HttpResponseEntity response =
this.webClient.getWebClientService()
.post()
.uri(uri)
- .body(multipartBuilder.build(), OptionalLong.empty())
+ .body(new ByteArrayInputStream(requestBody),
OptionalLong.of(requestBody.length))
.header(AUTHORIZATION_HEADER, authToken.getAuthzHeaderValue())
.header(CONTENT_TYPE_HEADER,
multipartBuilder.getHttpContentType().getContentType())
.retrieve()) {
@@ -1066,15 +1071,26 @@ public class BitbucketRepositoryClient implements
GitRepositoryClient {
}
}
- private String getErrorMessage(HttpResponseEntity response) throws
FlowRegistryException {
+ private String getErrorMessage(final HttpResponseEntity response) throws
FlowRegistryException {
+ final byte[] responseBody;
+ try (InputStream body = response.body()) {
+ responseBody = body == null ? new byte[0] : new
LimitingInputStream(body, MAX_ERROR_RESPONSE_LENGTH).readAllBytes();
+ } catch (final IOException e) {
+ throw new FlowRegistryException("Failed to read Bitbucket API
response with HTTP status %d".formatted(response.statusCode()), e);
+ }
+
+ if (responseBody.length == 0) {
+ return "Empty response from Bitbucket API [%s] with HTTP status
%d".formatted(response.uri(), response.statusCode());
+ }
+
final JsonNode jsonResponse;
try {
- jsonResponse = this.objectMapper.readTree(response.body());
- } catch (IOException e) {
- throw new FlowRegistryException("Could not parse response from
Bitbucket API", e);
+ jsonResponse = this.objectMapper.readTree(responseBody);
+ } catch (final IOException e) {
+ return getNonJsonErrorMessage(response, responseBody);
}
- if (jsonResponse == null) {
- return "Unknown error";
+ if (jsonResponse == null || jsonResponse.isMissingNode()) {
+ return getNonJsonErrorMessage(response, responseBody);
}
final JsonNode errorNode = jsonResponse.get(FIELD_ERROR_MESSAGE);
@@ -1098,6 +1114,13 @@ public class BitbucketRepositoryClient implements
GitRepositoryClient {
return jsonResponse.toString();
}
+ private String getNonJsonErrorMessage(final HttpResponseEntity response,
final byte[] responseBody) {
+ final String contentType =
response.headers().getFirstHeader(CONTENT_TYPE_HEADER).orElse(UNKNOWN_CONTENT_TYPE);
+ final String body = new String(responseBody,
StandardCharsets.UTF_8).strip().replaceAll("\\s+", " ");
+ return "Non-JSON response from Bitbucket API [%s] with HTTP status %d,
Content-Type [%s], body [%s]"
+ .formatted(response.uri(), response.statusCode(), contentType,
body);
+ }
+
private JsonNode parseResponseBody(final HttpResponseEntity response,
final URI uri) throws FlowRegistryException {
try {
return objectMapper.readTree(response.body());
diff --git
a/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/test/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClientTest.java
b/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/test/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClientTest.java
index 8129d237ca1..d6d748b5854 100644
---
a/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/test/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClientTest.java
+++
b/nifi-extension-bundles/nifi-atlassian-bundle/nifi-atlassian-extensions/src/test/java/org/apache/nifi/atlassian/bitbucket/BitbucketRepositoryClientTest.java
@@ -20,6 +20,7 @@ package org.apache.nifi.atlassian.bitbucket;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.registry.flow.FlowRegistryException;
import org.apache.nifi.registry.flow.git.client.GitCreateContentRequest;
+import org.apache.nifi.web.client.api.HttpEntityHeaders;
import org.apache.nifi.web.client.api.HttpRequestBodySpec;
import org.apache.nifi.web.client.api.HttpRequestHeadersSpec;
import org.apache.nifi.web.client.api.HttpRequestUriSpec;
@@ -30,6 +31,10 @@ import
org.apache.nifi.web.client.provider.api.WebClientServiceProvider;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import org.mockito.stubbing.OngoingStubbing;
@@ -43,14 +48,17 @@ import java.nio.charset.StandardCharsets;
import java.util.Optional;
import java.util.OptionalLong;
import java.util.Set;
+import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -78,6 +86,8 @@ class BitbucketRepositoryClientTest {
private HttpRequestBodySpec getBodySpec;
private HttpRequestBodySpec postAfterHeaders;
+ private HttpRequestBodySpec postBodySpec;
+
@BeforeEach
void setUp() {
lenient().when(webClientProvider.getWebClientService()).thenReturn(webClientService);
@@ -124,7 +134,7 @@ class BitbucketRepositoryClientTest {
private void stubPostChain(final HttpResponseEntity... responses) {
final HttpRequestUriSpec postSpec = mock(HttpRequestUriSpec.class);
- final HttpRequestBodySpec postBodySpec =
mock(HttpRequestBodySpec.class);
+ postBodySpec = mock(HttpRequestBodySpec.class);
final HttpRequestHeadersSpec afterBody =
mock(HttpRequestHeadersSpec.class);
postAfterHeaders = mock(HttpRequestBodySpec.class);
lenient().when(webClientService.post()).thenReturn(postSpec);
@@ -147,7 +157,14 @@ class BitbucketRepositoryClientTest {
}
private HttpResponseEntity mockResponse(final int statusCode, final String
body) {
+ return mockResponse(statusCode, body, "application/json");
+ }
+
+ private HttpResponseEntity mockResponse(final int statusCode, final String
body, final String contentType) {
final HttpResponseEntity response = mock(HttpResponseEntity.class);
+ final HttpEntityHeaders headers = mock(HttpEntityHeaders.class);
+
lenient().when(headers.getFirstHeader("Content-Type")).thenReturn(Optional.of(contentType));
+ lenient().when(response.headers()).thenReturn(headers);
lenient().when(response.statusCode()).thenReturn(statusCode);
lenient().when(response.body()).thenReturn(new
ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8)));
return response;
@@ -173,10 +190,6 @@ class BitbucketRepositoryClientTest {
return mockResponse(HttpURLConnection.HTTP_CONFLICT,
"{\"error\":{\"message\":\"conflict\"}}");
}
- private HttpResponseEntity serverErrorResponse() {
- return mockResponse(HttpURLConnection.HTTP_INTERNAL_ERROR,
"{\"error\":{\"message\":\"internal error\"}}");
- }
-
private HttpResponseEntity createdResponse() {
return mockResponse(HttpURLConnection.HTTP_CREATED, "{}");
}
@@ -197,7 +210,7 @@ class BitbucketRepositoryClientTest {
}
@Test
- void testCreateContentCloudSuccess() throws FlowRegistryException {
+ void testCreateContentCloudSuccess() throws FlowRegistryException,
IOException {
stubGetChain(
branchListResponse(),
branchHeadResponse(BRANCH_HEAD_SHA),
@@ -209,6 +222,14 @@ class BitbucketRepositoryClientTest {
final BitbucketRepositoryClient client = buildCloudClient();
final String commitSha =
client.createContent(createRequest("existing-sha", null));
assertEquals(RESULT_COMMIT_SHA, commitSha);
+
+ final ArgumentCaptor<InputStream> requestBody =
ArgumentCaptor.forClass(InputStream.class);
+ final ArgumentCaptor<OptionalLong> contentLength =
ArgumentCaptor.forClass(OptionalLong.class);
+ verify(postBodySpec).body(requestBody.capture(),
contentLength.capture());
+ assertTrue(contentLength.getValue().isPresent());
+ assertTrue(contentLength.getValue().getAsLong() > CONTENT.length());
+ final String multipartBody = new
String(requestBody.getValue().readAllBytes(), StandardCharsets.UTF_8);
+ assertTrue(multipartBody.contains("name=\"%s\";
filename=\"test.json\"".formatted(FILE_PATH)));
}
@Test
@@ -257,17 +278,31 @@ class BitbucketRepositoryClientTest {
assertTrue(exception.getMessage().contains("Push failed after 3
attempts"));
}
- @Test
- void testCreateContentCloudNon409ErrorNotRetried() throws
FlowRegistryException {
+ @ParameterizedTest
+ @MethodSource("nonJsonErrorResponses")
+ void testCreateContentCloudNon409ErrorNotRetried(final String
responseBody, final String expectedMessage, final String unexpectedMessage)
+ throws FlowRegistryException {
stubGetChain(
branchListResponse(),
branchHeadResponse(BRANCH_HEAD_SHA)
);
- stubPostChain(serverErrorResponse());
+ stubPostChain(mockResponse(HttpURLConnection.HTTP_BAD_REQUEST,
responseBody, "text/html"));
final BitbucketRepositoryClient client = buildCloudClient();
- assertThrows(FlowRegistryException.class,
+ final FlowRegistryException exception =
assertThrows(FlowRegistryException.class,
() -> client.createContent(createRequest("existing-sha",
null)));
+ assertTrue(exception.getMessage().contains("HTTP status 400"));
+ assertTrue(exception.getMessage().contains(expectedMessage));
+ assertFalse(exception.getMessage().contains(unexpectedMessage));
+ }
+
+ private static Stream<Arguments> nonJsonErrorResponses() {
+ final String truncatedBody = "<html>" + "a".repeat(506);
+ return Stream.of(
+ Arguments.of("\n<html><body>Bad Request (400)</body></html>",
"Content-Type [text/html], body [<html><body>Bad Request (400)</body></html>]",
"Empty response"),
+ Arguments.of("", "Empty response from Bitbucket API",
"Non-JSON response"),
+ Arguments.of(truncatedBody + "TRUNCATED", "body
[%s]".formatted(truncatedBody), "TRUNCATED")
+ );
}
@Test
