Skip to content

NIFI-15145: Add RecordLookup, KeyValueLookup, and MapCacheClient Services for Couchbase bundle #10467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
exceptionfactory merged 2 commits into from
Mar 4, 2026
Merged

NIFI-15145: Add RecordLookup, KeyValueLookup, and MapCacheClient Services for Couchbase bundle #10467

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
628c4b9
NIFI-15145: Add RecordLookup, KeyValueLookup, and MapCacheClient Serv…
mark-bathori Oct 27, 2025
c6e6be6
Review items
mark-bathori Feb 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions nifi-extension-bundles/nifi-couchbase-bundle/nifi-couchbase-nar/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,11 @@
<artifactId>nifi-couchbase-processors</artifactId>
<version>2.9.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-couchbase-services</artifactId>
<version>2.9.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-couchbase-services-api-nar</artifactId>
Expand Down
22 changes: 11 additions & 11 deletions ...essors/src/main/java/org/apache/nifi/processors/couchbase/AbstractCouchbaseProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,26 +50,24 @@

public abstract class AbstractCouchbaseProcessor extends AbstractProcessor {

protected CouchbaseConnectionService connectionService;

public static final PropertyDescriptor DOCUMENT_ID = new PropertyDescriptor.Builder()
.name("Document ID")
.description("Couchbase document identifier, or an expression to construct the Couchbase document identifier.")
.description("Couchbase document identifier, or an expression to construct the Couchbase document identifier")
.required(true)
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.build();

public static final PropertyDescriptor COUCHBASE_CONNECTION_SERVICE = new PropertyDescriptor.Builder()
.name("Couchbase Connection Service")
.description("A Couchbase Connection Service which manages connections to a Couchbase cluster.")
.description("Service responsible for managing connections to Couchbase cluster")
.required(true)
.identifiesControllerService(CouchbaseConnectionService.class)
.build();

public static final PropertyDescriptor BUCKET_NAME = new PropertyDescriptor.Builder()
.name("Bucket Name")
.description("The name of the bucket where documents will be stored. Each bucket contains a hierarchy of scopes and collections to group keys and values logically.")
.description("The name of the bucket where documents will be stored. Each bucket contains a hierarchy of scopes and collections to group keys and values logically")
.required(true)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.defaultValue(DEFAULT_BUCKET)
Expand All @@ -78,7 +76,7 @@ public abstract class AbstractCouchbaseProcessor extends AbstractProcessor {

public static final PropertyDescriptor SCOPE_NAME = new PropertyDescriptor.Builder()
.name("Scope Name")
.description("The name of the scope which is a logical namespace within a bucket, serving to categorize and organize related collections.")
.description("The name of the scope which is a logical namespace within a bucket, serving to categorize and organize related collections")
.required(true)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.defaultValue(DEFAULT_SCOPE)
Expand All @@ -87,7 +85,7 @@ public abstract class AbstractCouchbaseProcessor extends AbstractProcessor {

public static final PropertyDescriptor COLLECTION_NAME = new PropertyDescriptor.Builder()
.name("Collection Name")
.description("The name of collection which is a logical container within a scope, used to hold documents.")
.description("The name of collection which is a logical container within a scope, used to hold documents")
.required(true)
.addValidator(StandardValidators.NON_BLANK_VALIDATOR)
.defaultValue(DEFAULT_COLLECTION)
Expand All @@ -96,25 +94,25 @@ public abstract class AbstractCouchbaseProcessor extends AbstractProcessor {

public static final PropertyDescriptor DOCUMENT_TYPE = new PropertyDescriptor.Builder()
.name("Document Type")
.description("The content type for storing the document.")
.description("The content type for storing the document")
.required(true)
.allowableValues(DocumentType.values())
.defaultValue(DocumentType.JSON.toString())
.build();

public static final Relationship REL_SUCCESS = new Relationship.Builder()
.name("success")
.description("A FlowFile is routed to this relationship after the data ingestion was successful.")
.description("A FlowFile is routed to this relationship after the data ingestion was successful")
.build();

public static final Relationship REL_FAILURE = new Relationship.Builder()
.name("failure")
.description("A FlowFile is routed to this relationship if the operation failed and retrying the operation will also fail, such as an invalid data or schema.")
.description("A FlowFile is routed to this relationship if the operation failed and retrying the operation will also fail, such as an invalid data or schema")
.build();

public static final Relationship REL_RETRY = new Relationship.Builder()
.name("retry")
.description("All FlowFile that fail due to server/cluster availability go to this relationship.")
.description("All FlowFile that fail due to server/cluster availability go to this relationship")
.build();

private static final List<PropertyDescriptor> PROPERTIES = List.of(
Expand All @@ -128,6 +126,8 @@ public abstract class AbstractCouchbaseProcessor extends AbstractProcessor {

private static final Set<Relationship> RELATIONSHIPS = Set.of(REL_SUCCESS, REL_FAILURE, REL_RETRY);

protected volatile CouchbaseConnectionService connectionService;

@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
return PROPERTIES;
Expand Down
2 changes: 0 additions & 2 deletions ...couchbase-processors/src/main/java/org/apache/nifi/processors/couchbase/PutCouchbase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.services.couchbase.CouchbaseClient;
import org.apache.nifi.services.couchbase.CouchbaseConnectionService;
import org.apache.nifi.services.couchbase.exception.CouchbaseException;
import org.apache.nifi.services.couchbase.utils.CouchbaseContext;
import org.apache.nifi.services.couchbase.utils.CouchbaseUpsertResult;
Expand Down Expand Up @@ -69,7 +68,6 @@ public void onTrigger(ProcessContext context, ProcessSession session) throws Pro
}

final long startNanos = System.nanoTime();
final CouchbaseConnectionService connectionService = context.getProperty(COUCHBASE_CONNECTION_SERVICE).asControllerService(CouchbaseConnectionService.class);
final String documentId = context.getProperty(DOCUMENT_ID).evaluateAttributeExpressions(flowFile).getValue();

final CouchbaseContext couchbaseContext = getCouchbaseContext(context, flowFile);
Expand Down
41 changes: 41 additions & 0 deletions ...rs/src/test/java/org/apache/nifi/processors/couchbase/AbstractCouchbaseProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.nifi.processors.couchbase;

import org.apache.nifi.services.couchbase.CouchbaseClient;
import org.apache.nifi.services.couchbase.CouchbaseConnectionService;

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

public abstract class AbstractCouchbaseProcessorTest {

protected static final String SERVICE_ID = "couchbaseConnectionService";
protected static final String TEST_DOCUMENT_ID = "test-document-id";
protected static final String TEST_DOCUMENT_CONTENT = "{\"key\":\"value\"}";
protected static final String TEST_SERVICE_LOCATION = "couchbase://test-location";
protected static final long TEST_CAS = 1L;

protected static CouchbaseConnectionService mockConnectionService(CouchbaseClient client) {
final CouchbaseConnectionService connectionService = mock(CouchbaseConnectionService.class);
when(connectionService.getIdentifier()).thenReturn(SERVICE_ID);
when(connectionService.getClient(any())).thenReturn(client);
when(connectionService.getServiceLocation()).thenReturn(TEST_SERVICE_LOCATION);
return connectionService;
}
}
71 changes: 24 additions & 47 deletions ...hbase-processors/src/test/java/org/apache/nifi/processors/couchbase/GetCouchbaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,8 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -60,12 +60,7 @@
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

public class GetCouchbaseTest {

private static final String SERVICE_ID = "couchbaseConnectionService";
private static final String TEST_DOCUMENT_ID = "test-document-id";
private static final String TEST_SERVICE_LOCATION = "couchbase://test-location";
private static final long TEST_CAS = 1L;
public class GetCouchbaseTest extends AbstractCouchbaseProcessorTest {

private TestRunner runner;

Expand All @@ -76,20 +71,15 @@ public void init() {

@Test
public void testOnTriggerWithProvidedDocumentId() throws CouchbaseException, InitializationException {
final String content = "{\"key\":\"value\"}";

final CouchbaseGetResult result = new CouchbaseGetResult(content.getBytes(), TEST_CAS);
final CouchbaseGetResult result = new CouchbaseGetResult(TEST_DOCUMENT_CONTENT.getBytes(), TEST_CAS);

final CouchbaseClient client = mock(CouchbaseClient.class);
when(client.getDocument(anyString())).thenReturn(result);

final CouchbaseConnectionService service = mock(CouchbaseConnectionService.class);
when(service.getIdentifier()).thenReturn(SERVICE_ID);
when(service.getClient(any())).thenReturn(client);
when(service.getServiceLocation()).thenReturn(TEST_SERVICE_LOCATION);
final CouchbaseConnectionService connectionService = mockConnectionService(client);

runner.addControllerService(SERVICE_ID, service);
runner.enableControllerService(service);
runner.addControllerService(SERVICE_ID, connectionService);
runner.enableControllerService(connectionService);
runner.setProperty(DOCUMENT_ID, TEST_DOCUMENT_ID);
runner.setProperty(COUCHBASE_CONNECTION_SERVICE, SERVICE_ID);
runner.enqueue(new byte[0]);
Expand All @@ -101,7 +91,7 @@ public void testOnTriggerWithProvidedDocumentId() throws CouchbaseException, Ini
runner.assertTransferCount(REL_FAILURE, 0);

final MockFlowFile outFile = runner.getFlowFilesForRelationship(REL_SUCCESS).getFirst();
outFile.assertContentEquals(content);
outFile.assertContentEquals(TEST_DOCUMENT_CONTENT);
outFile.assertAttributeEquals(BUCKET_ATTRIBUTE, DEFAULT_BUCKET);
outFile.assertAttributeEquals(SCOPE_ATTRIBUTE, DEFAULT_SCOPE);
outFile.assertAttributeEquals(COLLECTION_ATTRIBUTE, DEFAULT_COLLECTION);
Expand All @@ -116,24 +106,18 @@ public void testOnTriggerWithProvidedDocumentId() throws CouchbaseException, Ini

@Test
public void testWithDocumentIdAsFlowFileAttribute() throws CouchbaseException, InitializationException {
final String content = "{\"key\":\"value\"}";

final CouchbaseGetResult result = new CouchbaseGetResult(content.getBytes(), TEST_CAS);
final CouchbaseGetResult result = new CouchbaseGetResult(TEST_DOCUMENT_CONTENT.getBytes(), TEST_CAS);

final CouchbaseClient client = mock(CouchbaseClient.class);
when(client.getDocument(anyString())).thenReturn(result);

final CouchbaseConnectionService service = mock(CouchbaseConnectionService.class);
when(service.getIdentifier()).thenReturn(SERVICE_ID);
when(service.getClient(any())).thenReturn(client);
final CouchbaseConnectionService connectionService = mockConnectionService(client);

final MockFlowFile flowFile = new MockFlowFile(0);
final Map<String, String> attributes = new HashMap<>();
attributes.put("flowfile_document_id", TEST_DOCUMENT_ID);
flowFile.putAttributes(attributes);
flowFile.putAttributes(Collections.singletonMap("flowfile_document_id", TEST_DOCUMENT_ID));

runner.addControllerService(SERVICE_ID, service);
runner.enableControllerService(service);
runner.addControllerService(SERVICE_ID, connectionService);
runner.enableControllerService(connectionService);
runner.setProperty(DOCUMENT_ID, "${flowfile_document_id}");
runner.setProperty(COUCHBASE_CONNECTION_SERVICE, SERVICE_ID);
runner.enqueue(flowFile);
Expand All @@ -145,7 +129,7 @@ public void testWithDocumentIdAsFlowFileAttribute() throws CouchbaseException, I
runner.assertTransferCount(REL_FAILURE, 0);

final MockFlowFile outFile = runner.getFlowFilesForRelationship(REL_SUCCESS).getFirst();
outFile.assertContentEquals(content);
outFile.assertContentEquals(TEST_DOCUMENT_CONTENT);
outFile.assertAttributeEquals(BUCKET_ATTRIBUTE, DEFAULT_BUCKET);
outFile.assertAttributeEquals(SCOPE_ATTRIBUTE, DEFAULT_SCOPE);
outFile.assertAttributeEquals(COLLECTION_ATTRIBUTE, DEFAULT_COLLECTION);
Expand All @@ -154,24 +138,20 @@ public void testWithDocumentIdAsFlowFileAttribute() throws CouchbaseException, I

@Test
public void testWithFlowFileAttributes() throws CouchbaseException, InitializationException {
final String documentId = "test-document-id";
final String content = "{\"key\":\"value\"}";
final String testBucket = "test-bucket";
final String testScope = "test-scope";
final String testCollection = "test-collection";

final CouchbaseGetResult result = new CouchbaseGetResult(content.getBytes(), TEST_CAS);
final CouchbaseGetResult result = new CouchbaseGetResult(TEST_DOCUMENT_CONTENT.getBytes(), TEST_CAS);

final CouchbaseClient client = mock(CouchbaseClient.class);
when(client.getDocument(anyString())).thenReturn(result);

final CouchbaseConnectionService service = mock(CouchbaseConnectionService.class);
when(service.getIdentifier()).thenReturn(SERVICE_ID);
when(service.getClient(any())).thenReturn(client);
final CouchbaseConnectionService connectionService = mockConnectionService(client);

runner.addControllerService(SERVICE_ID, service);
runner.enableControllerService(service);
runner.setProperty(DOCUMENT_ID, documentId);
runner.addControllerService(SERVICE_ID, connectionService);
runner.enableControllerService(connectionService);
runner.setProperty(DOCUMENT_ID, TEST_DOCUMENT_ID);
runner.setProperty(BUCKET_NAME, "${bucket.attribute}");
runner.setProperty(SCOPE_NAME, "${scope.attribute}");
runner.setProperty(COLLECTION_NAME, "${collection.attribute}");
Expand All @@ -181,17 +161,16 @@ public void testWithFlowFileAttributes() throws CouchbaseException, Initializati
attributes.put("bucket.attribute", testBucket);
attributes.put("scope.attribute", testScope);
attributes.put("collection.attribute", testCollection);
final byte[] input = documentId.getBytes(StandardCharsets.UTF_8);
runner.enqueue(input, attributes);
runner.enqueue(new byte[0], attributes);
runner.run();

verify(client, times(1)).getDocument(eq(documentId));
verify(client, times(1)).getDocument(eq(TEST_DOCUMENT_ID));

runner.assertTransferCount(REL_SUCCESS, 1);
runner.assertTransferCount(REL_FAILURE, 0);

final MockFlowFile outFile = runner.getFlowFilesForRelationship(REL_SUCCESS).getFirst();
outFile.assertContentEquals(content);
outFile.assertContentEquals(TEST_DOCUMENT_CONTENT);
outFile.assertAttributeEquals(BUCKET_ATTRIBUTE, testBucket);
outFile.assertAttributeEquals(SCOPE_ATTRIBUTE, testScope);
outFile.assertAttributeEquals(COLLECTION_ATTRIBUTE, testCollection);
Expand All @@ -204,13 +183,11 @@ public void testWithFailure() throws CouchbaseException, InitializationException
when(client.getExceptionCategory(any())).thenReturn(ExceptionCategory.FAILURE);
when(client.getDocument(anyString())).thenThrow(new CouchbaseException("", new TestCouchbaseException("Test exception")));

final CouchbaseConnectionService service = mock(CouchbaseConnectionService.class);
when(service.getIdentifier()).thenReturn(SERVICE_ID);
when(service.getClient(any())).thenReturn(client);
final CouchbaseConnectionService connectionService = mockConnectionService(client);

runner.setProperty(DOCUMENT_ID, TEST_DOCUMENT_ID);
runner.addControllerService(SERVICE_ID, service);
runner.enableControllerService(service);
runner.addControllerService(SERVICE_ID, connectionService);
runner.enableControllerService(connectionService);
runner.setProperty(COUCHBASE_CONNECTION_SERVICE, SERVICE_ID);
runner.enqueue(new byte[0]);
runner.run();
Expand Down
Loading
Loading