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 3cbf8d7dad8 NIFI-16191 Added OAuth 2 support for
ConfluentSchemaRegistry (#11534)
3cbf8d7dad8 is described below
commit 3cbf8d7dad82c72d4cd4533316e10af1151be8dd
Author: Rajmund Takács <[email protected]>
AuthorDate: Thu Aug 27 20:33:54 2026 +0200
NIFI-16191 Added OAuth 2 support for ConfluentSchemaRegistry (#11534)
Signed-off-by: David Handermann <[email protected]>
---
.../nifi-confluent-schema-registry-service/pom.xml | 4 ++
.../schemaregistry/ConfluentSchemaRegistry.java | 29 ++++++++++---
.../schemaregistry/client/AuthenticationType.java | 4 +-
.../client/RestSchemaRegistryClient.java | 48 ++++++++++++++++++----
.../ConfluentSchemaRegistryTest.java | 42 +++++++++++++++++++
.../client/RestSchemaRegistryClientTest.java | 47 +++++++++++++++++++++
6 files changed, 159 insertions(+), 15 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/pom.xml
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/pom.xml
index 2b3cfe64834..cc44ac3a1ca 100644
---
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/pom.xml
+++
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/pom.xml
@@ -67,6 +67,10 @@
<artifactId>nifi-web-client</artifactId>
<version>2.12.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-oauth2-provider-api</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-confluent-protobuf-message-name-resolver</artifactId>
diff --git
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
index 63b51824d0e..7bbc072e22a 100644
---
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
+++
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistry.java
@@ -36,6 +36,7 @@ import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.migration.PropertyConfiguration;
+import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.schema.access.SchemaField;
import org.apache.nifi.schema.access.SchemaNotFoundException;
@@ -140,6 +141,14 @@ public class ConfluentSchemaRegistry extends
AbstractControllerService implement
.sensitive(true)
.build();
+ static final PropertyDescriptor OAUTH2_ACCESS_TOKEN_PROVIDER = new
PropertyDescriptor.Builder()
+ .name("OAuth2 Access Token Provider")
+ .description("OAuth2 Access Token Provider used for Bearer
authentication to Confluent Schema Registry")
+ .identifiesControllerService(OAuth2AccessTokenProvider.class)
+ .required(true)
+ .dependsOn(AUTHENTICATION_TYPE,
AuthenticationType.OAUTH2.toString())
+ .build();
+
private static final List<PropertyDescriptor> PROPERTY_DESCRIPTORS =
List.of(
SCHEMA_REGISTRY_URLS,
SSL_CONTEXT,
@@ -148,7 +157,8 @@ public class ConfluentSchemaRegistry extends
AbstractControllerService implement
CACHE_EXPIRATION,
AUTHENTICATION_TYPE,
USERNAME,
- PASSWORD
+ PASSWORD,
+ OAUTH2_ACCESS_TOKEN_PROVIDER
);
private volatile SchemaRegistryClient client;
@@ -182,8 +192,7 @@ public class ConfluentSchemaRegistry extends
AbstractControllerService implement
final SSLContextProvider sslContextProvider =
context.getProperty(SSL_CONTEXT).asControllerService(SSLContextProvider.class);
- final String username = context.getProperty(USERNAME).getValue();
- final String password = context.getProperty(PASSWORD).getValue();
+ final AuthenticationType authenticationType =
AuthenticationType.valueOf(context.getProperty(AUTHENTICATION_TYPE).getValue());
// generate a map of http headers where the key is the remainder of
the property name after
// the request header prefix
@@ -196,8 +205,18 @@ public class ConfluentSchemaRegistry extends
AbstractControllerService implement
Map.Entry::getValue)
);
- final SchemaRegistryClient restClient = new
RestSchemaRegistryClient(baseUrls, timeoutMillis,
- sslContextProvider, username, password, getLogger(),
httpHeaders);
+ final SchemaRegistryClient restClient;
+ if (AuthenticationType.OAUTH2.equals(authenticationType)) {
+ OAuth2AccessTokenProvider oauth2AccessTokenProvider =
context.getProperty(OAUTH2_ACCESS_TOKEN_PROVIDER).asControllerService(OAuth2AccessTokenProvider.class);
+ oauth2AccessTokenProvider.getAccessDetails();
+ restClient = new RestSchemaRegistryClient(baseUrls, timeoutMillis,
+ sslContextProvider, oauth2AccessTokenProvider,
getLogger(), httpHeaders);
+ } else {
+ final String username = context.getProperty(USERNAME).getValue();
+ final String password = context.getProperty(PASSWORD).getValue();
+ restClient = new RestSchemaRegistryClient(baseUrls, timeoutMillis,
+ sslContextProvider, username, password, getLogger(),
httpHeaders);
+ }
final int cacheSize = context.getProperty(CACHE_SIZE).asInteger();
final long cacheExpiration =
context.getProperty(CACHE_EXPIRATION).asTimePeriod(TimeUnit.NANOSECONDS);
diff --git
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/AuthenticationType.java
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/AuthenticationType.java
index 1200575d5c0..460808b552b 100644
---
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/AuthenticationType.java
+++
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/AuthenticationType.java
@@ -22,5 +22,7 @@ package org.apache.nifi.confluent.schemaregistry.client;
public enum AuthenticationType {
BASIC,
- NONE
+ NONE,
+
+ OAUTH2
}
diff --git
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClient.java
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClient.java
index 7e80a05d76c..8cfc02a2c5b 100644
---
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClient.java
+++
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/main/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClient.java
@@ -25,6 +25,7 @@ import org.apache.avro.SchemaParseException;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.avro.AvroTypeUtil;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.schemaregistry.services.SchemaDefinition;
import org.apache.nifi.schemaregistry.services.StandardSchemaDefinition;
@@ -75,6 +76,8 @@ public class RestSchemaRegistryClient implements
SchemaRegistryClient {
private final Map<String, String> httpHeaders;
private final WebClientService webClientService;
+ private OAuth2AccessTokenProvider oauth2AccessTokenProvider;
+
private static final ObjectMapper objectMapper = new ObjectMapper();
private static final String SUBJECT_FIELD_NAME = "subject";
@@ -90,6 +93,7 @@ public class RestSchemaRegistryClient implements
SchemaRegistryClient {
private static final String APPLICATION_JSON_CONTENT_TYPE =
"application/json";
private static final String BASIC_CREDENTIALS_FORMAT = "%s:%s";
private static final String BASIC_AUTHORIZATION_FORMAT = "Basic %s";
+ private static final String BEARER_AUTHORIZATION_FORMAT = "Bearer %s";
public RestSchemaRegistryClient(final List<String> baseUrls,
final int timeoutMillis,
@@ -98,16 +102,34 @@ public class RestSchemaRegistryClient implements
SchemaRegistryClient {
final String password,
final ComponentLog logger,
final Map<String, String> httpHeaders) {
- this.baseUrls = new ArrayList<>(baseUrls);
- this.httpHeaders = new HashMap<>(httpHeaders);
+ this(baseUrls, timeoutMillis, sslContextProvider, logger, httpHeaders);
if (StringUtils.isNoneBlank(username, password)) {
final String credentials =
BASIC_CREDENTIALS_FORMAT.formatted(username, password);
final byte[] credentialsEncoded =
credentials.getBytes(StandardCharsets.UTF_8);
final String authorization =
Base64.getEncoder().encodeToString(credentialsEncoded);
final String basicAuthorization =
BASIC_AUTHORIZATION_FORMAT.formatted(authorization);
- this.httpHeaders.put(HttpHeaderName.AUTHORIZATION.getHeaderName(),
basicAuthorization);
+
this.httpHeaders.putIfAbsent(HttpHeaderName.AUTHORIZATION.getHeaderName(),
basicAuthorization);
}
+ }
+
+ public RestSchemaRegistryClient(final List<String> baseUrls,
+ final int timeoutMillis,
+ final SSLContextProvider
sslContextProvider,
+ final OAuth2AccessTokenProvider
oauth2AccessTokenProvider,
+ final ComponentLog logger,
+ final Map<String, String> httpHeaders) {
+ this(baseUrls, timeoutMillis, sslContextProvider, logger, httpHeaders);
+ this.oauth2AccessTokenProvider = oauth2AccessTokenProvider;
+ }
+
+ public RestSchemaRegistryClient(final List<String> baseUrls,
+ final int timeoutMillis,
+ final SSLContextProvider
sslContextProvider,
+ final ComponentLog logger,
+ final Map<String, String> httpHeaders) {
+ this.baseUrls = new ArrayList<>(baseUrls);
+ this.httpHeaders = new HashMap<>(httpHeaders);
final StandardWebClientService standardWebClientService = new
StandardWebClientService();
final Duration timeout = Duration.ofMillis(timeoutMillis);
@@ -393,9 +415,7 @@ public class RestSchemaRegistryClient implements
SchemaRegistryClient {
.header(HttpHeaderName.ACCEPT.getHeaderName(),
APPLICATION_JSON_CONTENT_TYPE)
.header(HttpHeaderName.CONTENT_TYPE.getHeaderName(),
SCHEMA_REGISTRY_CONTENT_TYPE);
- for (final Map.Entry<String, String> header :
httpHeaders.entrySet()) {
- requestBodySpec = requestBodySpec.header(header.getKey(),
header.getValue());
- }
+ requestBodySpec = applyRequestHeaders(requestBodySpec);
final String requestBody = schema.toString();
try (HttpResponseEntity responseEntity =
requestBodySpec.body(requestBody).retrieve()) {
@@ -445,9 +465,7 @@ public class RestSchemaRegistryClient implements
SchemaRegistryClient {
.uri(uri)
.header(HttpHeaderName.ACCEPT.getHeaderName(),
APPLICATION_JSON_CONTENT_TYPE);
- for (final Map.Entry<String, String> header :
httpHeaders.entrySet()) {
- requestBodySpec = requestBodySpec.header(header.getKey(),
header.getValue());
- }
+ requestBodySpec = applyRequestHeaders(requestBodySpec);
try (HttpResponseEntity responseEntity =
requestBodySpec.retrieve()) {
final int responseCode = responseEntity.statusCode();
@@ -479,6 +497,18 @@ public class RestSchemaRegistryClient implements
SchemaRegistryClient {
+ " from any of the Confluent Schema Registry URL's provided;
failure response message: " + errorMessage);
}
+ private HttpRequestBodySpec applyRequestHeaders(final HttpRequestBodySpec
requestBodySpec) {
+ HttpRequestBodySpec updatedRequest = requestBodySpec;
+ if (oauth2AccessTokenProvider != null) {
+ final String accessToken =
oauth2AccessTokenProvider.getAccessDetails().getAccessToken();
+ updatedRequest =
updatedRequest.header(HttpHeaderName.AUTHORIZATION.getHeaderName(),
BEARER_AUTHORIZATION_FORMAT.formatted(accessToken));
+ }
+ for (final Map.Entry<String, String> header : httpHeaders.entrySet()) {
+ updatedRequest = updatedRequest.header(header.getKey(),
header.getValue());
+ }
+ return updatedRequest;
+ }
+
private String getTrimmedBase(String baseUrl) {
return baseUrl.endsWith("/") ? baseUrl.substring(0, baseUrl.length() -
1) : baseUrl;
}
diff --git
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistryTest.java
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistryTest.java
index f4aacc6d7d8..2d90634c085 100644
---
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistryTest.java
+++
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/ConfluentSchemaRegistryTest.java
@@ -17,6 +17,8 @@
package org.apache.nifi.confluent.schemaregistry;
import org.apache.nifi.confluent.schemaregistry.client.AuthenticationType;
+import org.apache.nifi.oauth2.AccessToken;
+import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockPropertyConfiguration;
import org.apache.nifi.util.NoOpProcessor;
@@ -29,11 +31,16 @@ import org.junit.jupiter.api.Test;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
class ConfluentSchemaRegistryTest {
private static final String SERVICE_ID =
ConfluentSchemaRegistry.class.getSimpleName();
+ private static final String OAUTH2_ACCESS_TOKEN_PROVIDER_ID =
"oauth2AccessTokenProvider";
+
private TestRunner runner;
private ConfluentSchemaRegistry registry;
@@ -45,6 +52,41 @@ class ConfluentSchemaRegistryTest {
runner.addControllerService(SERVICE_ID, registry);
}
+ @Test
+ public void testValidateAuthenticationTypeOAuth2MissingProvider() {
+ runner.setProperty(registry,
ConfluentSchemaRegistry.AUTHENTICATION_TYPE,
AuthenticationType.OAUTH2.toString());
+ runner.assertNotValid(registry);
+ }
+
+ @Test
+ public void testValidateAndEnableAuthenticationTypeOAuth2() throws
InitializationException {
+ final OAuth2AccessTokenProvider oauth2AccessTokenProvider =
mock(OAuth2AccessTokenProvider.class);
+
when(oauth2AccessTokenProvider.getIdentifier()).thenReturn(OAUTH2_ACCESS_TOKEN_PROVIDER_ID);
+ when(oauth2AccessTokenProvider.getAccessDetails()).thenReturn(new
AccessToken("access-token", null, "Bearer", 3600L, null));
+
+ runner.addControllerService(OAUTH2_ACCESS_TOKEN_PROVIDER_ID,
oauth2AccessTokenProvider);
+ runner.enableControllerService(oauth2AccessTokenProvider);
+ runner.setProperty(registry,
ConfluentSchemaRegistry.AUTHENTICATION_TYPE,
AuthenticationType.OAUTH2.toString());
+ runner.setProperty(registry,
ConfluentSchemaRegistry.OAUTH2_ACCESS_TOKEN_PROVIDER,
OAUTH2_ACCESS_TOKEN_PROVIDER_ID);
+ runner.assertValid(registry);
+ runner.enableControllerService(registry);
+ }
+
+ @Test
+ public void
testEnableAuthenticationTypeOAuth2FailsWhenAccessTokenUnavailable() throws
InitializationException {
+ final OAuth2AccessTokenProvider oauth2AccessTokenProvider =
mock(OAuth2AccessTokenProvider.class);
+
when(oauth2AccessTokenProvider.getIdentifier()).thenReturn(OAUTH2_ACCESS_TOKEN_PROVIDER_ID);
+ when(oauth2AccessTokenProvider.getAccessDetails()).thenThrow(new
RuntimeException("token unavailable"));
+
+ runner.addControllerService(OAUTH2_ACCESS_TOKEN_PROVIDER_ID,
oauth2AccessTokenProvider);
+ runner.enableControllerService(oauth2AccessTokenProvider);
+ runner.setProperty(registry,
ConfluentSchemaRegistry.AUTHENTICATION_TYPE,
AuthenticationType.OAUTH2.toString());
+ runner.setProperty(registry,
ConfluentSchemaRegistry.OAUTH2_ACCESS_TOKEN_PROVIDER,
OAUTH2_ACCESS_TOKEN_PROVIDER_ID);
+ runner.assertValid(registry);
+
+ assertThrows(Throwable.class, () ->
runner.enableControllerService(registry));
+ }
+
@Test
public void
testValidateAuthenticationTypeBasicMissingUsernameAndPassword() {
runner.setProperty(registry,
ConfluentSchemaRegistry.AUTHENTICATION_TYPE,
AuthenticationType.BASIC.toString());
diff --git
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClientTest.java
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClientTest.java
index b261a5b608f..52703ffeb7c 100644
---
a/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClientTest.java
+++
b/nifi-extension-bundles/nifi-confluent-platform-bundle/nifi-confluent-schema-registry-service/src/test/java/org/apache/nifi/confluent/schemaregistry/client/RestSchemaRegistryClientTest.java
@@ -23,6 +23,8 @@ import mockwebserver3.MockWebServer;
import mockwebserver3.RecordedRequest;
import okhttp3.Headers;
import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.oauth2.AccessToken;
+import org.apache.nifi.oauth2.OAuth2AccessTokenProvider;
import org.apache.nifi.schema.access.SchemaNotFoundException;
import org.apache.nifi.schemaregistry.services.SchemaDefinition;
import org.apache.nifi.schemaregistry.services.SchemaDefinition.SchemaType;
@@ -44,6 +46,8 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
@SuppressWarnings({"OptionalGetWithoutIsPresent", "SameParameterValue"})
@ExtendWith(MockitoExtension.class)
@@ -59,6 +63,7 @@ class RestSchemaRegistryClientTest {
private static final String REFERENCED_SCHEMA_NAME = "common.proto";
private static final String PROTOBUF = "PROTOBUF";
private static final String AUTHORIZATION_HEADER = "Authorization";
+ private static final String OAUTH_ACCESS_TOKEN = "oauth-access-token";
private static final String AVRO_SCHEMA_TEXT = """
{
"type": "record",
@@ -146,6 +151,38 @@ class RestSchemaRegistryClientTest {
}
}
+ @Test
+ void testGetSchemaByNameWithOAuth2BearerToken() throws IOException,
SchemaNotFoundException, InterruptedException {
+ final OAuth2AccessTokenProvider oauth2AccessTokenProvider =
mockOAuth2AccessTokenProvider(OAUTH_ACCESS_TOKEN);
+ client = new RestSchemaRegistryClient(List.of(baseUrl), 30000, null,
oauth2AccessTokenProvider, logger, Map.of());
+
+ enqueueCompleteSchemaResponse(SUBJECT_NAME, SCHEMA_ID, SCHEMA_VERSION,
AVRO_SCHEMA_TEXT);
+
+ RecordSchema schema = client.getSchema(SUBJECT_NAME);
+
+ assertNotNull(schema);
+ assertEquals(SUBJECT_NAME, schema.getIdentifier().getName().get());
+
+ final RecordedRequest request = verifyRequest("GET", "/subjects/" +
SUBJECT_NAME + "/versions/latest");
+ assertEquals("Bearer " + OAUTH_ACCESS_TOKEN,
request.getHeaders().get(AUTHORIZATION_HEADER));
+ }
+
+ @Test
+ void
testGetSchemaByNameUsesConfiguredAuthorizationHeaderInsteadOfOAuth2Provider()
throws IOException, SchemaNotFoundException, InterruptedException {
+ final OAuth2AccessTokenProvider oauth2AccessTokenProvider =
mockOAuth2AccessTokenProvider(OAUTH_ACCESS_TOKEN);
+
+ client = new RestSchemaRegistryClient(List.of(baseUrl), 30000, null,
oauth2AccessTokenProvider, logger,
+ Map.of(AUTHORIZATION_HEADER, "Bearer static-token"));
+
+ enqueueCompleteSchemaResponse(SUBJECT_NAME, SCHEMA_ID, SCHEMA_VERSION,
AVRO_SCHEMA_TEXT);
+
+ RecordSchema schema = client.getSchema(SUBJECT_NAME);
+
+ assertNotNull(schema);
+ final RecordedRequest request = verifyRequest("GET", "/subjects/" +
SUBJECT_NAME + "/versions/latest");
+ assertEquals("Bearer static-token",
request.getHeaders().get(AUTHORIZATION_HEADER));
+ }
+
@Test
void testGetSchemaByIdWithoutSubjectAndVersionInfo() throws IOException,
SchemaNotFoundException, InterruptedException {
/*
@@ -470,4 +507,14 @@ class RestSchemaRegistryClientTest {
assertEquals(expectedPath, request.getTarget());
return request;
}
+
+ private OAuth2AccessTokenProvider mockOAuth2AccessTokenProvider(final
String accessToken) {
+ final OAuth2AccessTokenProvider oauth2AccessTokenProvider =
mock(OAuth2AccessTokenProvider.class);
+
when(oauth2AccessTokenProvider.getAccessDetails()).thenReturn(accessToken(accessToken));
+ return oauth2AccessTokenProvider;
+ }
+
+ private AccessToken accessToken(final String token) {
+ return new AccessToken(token, null, "Bearer", 3600L, null);
+ }
}
