You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by lo...@apache.org on 2024/10/30 09:51:54 UTC
(nifi-minifi-cpp) branch MINIFICPP-2469 updated (d0203f306 -> 396e25c6f)
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a change to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
omit d0203f306 Add test logs
omit a88632f9f Add extenstion docs and refactor docker tests
omit 19663a403 MINIFICPP-2469 Create GetCouchbaseKey processor
omit d12fc1b76 Fix test on Windows
omit 9dd8d0146 Fix test after rebase
omit 7438e139d Update extensions in README.md
omit 5cf601df0 Refactor tests
omit 866f6daf9 Review update
omit b31fff387 Add Document Type property
omit ee8f5e4a9 Refactor
omit a22a7a8fa Remove connection state management
omit c205a504d Fix clang issue and revise temporary failures
omit 281e3eed6 Review update
omit fc809299a Handle temporary errors on connection initialization
omit c96806180 MINIFICPP-2467 MINIFICPP-2468 PutCouchbaseKey processor and controller
add 2daccca48 bump python version
add 624743dad MINIFICPP-2465 SplitContent
add 24e19a324 MINIFICPP-2471 SegmentContent
add 102a219e4 MINIFICPP-2474 Fix memory leaks in extensions/libarchive
add e62b5bb15 MINIFICPP-2480 workaround python version incompatibilities around 3.6 vs 3.11
add d0116706e MINIFICPP-2482 drop python3.6 support on mac
new 9bfd7df3a MINIFICPP-2467 MINIFICPP-2468 PutCouchbaseKey processor and controller
new 3593eeb92 Handle temporary errors on connection initialization
new 6c08d9615 Review update
new 3c4d115b7 Fix clang issue and revise temporary failures
new 6e74934c5 Remove connection state management
new 50f738d0c Refactor
new 7e8e56c35 Add Document Type property
new abc4c2d26 Review update
new 57d1c6403 Refactor tests
new e1086ceae Update extensions in README.md
new 3fcbc57af Fix test after rebase
new 660fde2ce Fix test on Windows
new 87e65111b MINIFICPP-2469 Create GetCouchbaseKey processor
new 839243724 Add extenstion docs and refactor docker tests
new 396e25c6f Add test logs
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (d0203f306)
\
N -- N -- N refs/heads/MINIFICPP-2469 (396e25c6f)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 15 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
PROCESSORS.md | 66 +++
README.md | 6 +-
extensions/libarchive/FocusArchiveEntry.cpp | 29 +-
extensions/libarchive/MergeContent.h | 57 +-
extensions/libarchive/ReadArchiveStream.cpp | 10 +-
extensions/libarchive/ReadArchiveStream.h | 20 +-
.../SmartArchivePtrs.h} | 24 +-
extensions/libarchive/UnfocusArchiveEntry.cpp | 39 +-
extensions/libarchive/WriteArchiveStream.cpp | 6 +-
extensions/libarchive/WriteArchiveStream.h | 37 +-
extensions/libarchive/tests/CMakeLists.txt | 1 +
.../libarchive/tests/CompressContentTests.cpp | 36 +-
extensions/libarchive/tests/FocusArchiveTests.cpp | 16 +-
.../libarchive/tests/ManipulateArchiveTests.cpp | 2 +-
extensions/libarchive/tests/MergeFileTests.cpp | 56 +-
extensions/libarchive/tests/util/ArchiveTests.cpp | 57 +-
.../libarchive/{ => tests/util}/ArchiveTests.h | 15 +-
extensions/python/CMakeLists.txt | 6 +-
extensions/python/PythonInterpreter.cpp | 83 ++-
.../processors/SegmentContent.cpp | 108 ++++
.../processors/SegmentContent.h | 78 +++
.../processors/SplitContent.cpp | 224 ++++++++
.../standard-processors/processors/SplitContent.h | 130 +++++
.../tests/unit/SegmentContentTests.cpp | 263 +++++++++
.../tests/unit/SplitContentTests.cpp | 597 +++++++++++++++++++++
libminifi/include/io/InputStream.h | 8 +
libminifi/src/core/ProcessSession.cpp | 1 +
libminifi/test/libtest/unit/TestBase.cpp | 18 +-
libminifi/test/libtest/unit/TestBase.h | 1 +
29 files changed, 1749 insertions(+), 245 deletions(-)
copy extensions/{test-processors/LogOnDestructionProcessor.cpp => libarchive/SmartArchivePtrs.h} (54%)
rename extensions/libarchive/{ => tests/util}/ArchiveTests.h (87%)
create mode 100644 extensions/standard-processors/processors/SegmentContent.cpp
create mode 100644 extensions/standard-processors/processors/SegmentContent.h
create mode 100644 extensions/standard-processors/processors/SplitContent.cpp
create mode 100644 extensions/standard-processors/processors/SplitContent.h
create mode 100644 extensions/standard-processors/tests/unit/SegmentContentTests.cpp
create mode 100644 extensions/standard-processors/tests/unit/SplitContentTests.cpp
(nifi-minifi-cpp) 14/15: Add extenstion docs and refactor docker tests
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 839243724204880bf079fe7334ca76c3a8be919c
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 21 14:24:31 2024 +0200
Add extenstion docs and refactor docker tests
---
README.md | 2 +-
docker/test/integration/features/couchbase.feature | 32 +++++++++++-----------
2 files changed, 17 insertions(+), 17 deletions(-)
diff --git a/README.md b/README.md
index 069d142ad..3c659f335 100644
--- a/README.md
+++ b/README.md
@@ -75,7 +75,7 @@ The next table outlines CMAKE flags that correspond with MiNiFi extensions. Exte
| AWS | [AWSCredentialsService](CONTROLLERS.md#awscredentialsservice)<br/>[PutS3Object](PROCESSORS.md#puts3object)<br/>[DeleteS3Object](PROCESSORS.md#deletes3object)<br/>[FetchS3Object](PROCESSORS.md#fetchs3object)<br/>[ListS3](PROCESSORS.md#lists3) [...]
| Azure | [AzureStorageCredentialsService](CONTROLLERS.md#azurestoragecredentialsservice)<br/>[PutAzureBlobStorage](PROCESSORS.md#putazureblobstorage)<br/>[DeleteAzureBlobStorage](PROCESSORS.md#deleteazureblobstorage)<br/>[FetchAzureBlobStorage](PROCESSORS.md#fetchazureblobstorage)<br/>[ListAzureBlobStorage](PROCESSORS.md#listazureblobstorage)<br/>[PutAzureDataLakeStorage](PROCESSORS.md#putazuredatalakestorage)<br/>[DeleteAzureDataLakeStorage](PROCESSORS.md#del [...]
| CivetWeb | [ListenHTTP](PROCESSORS.md#listenhttp) [...]
-| Couchbase | [CouchbaseClusterService](CONTROLLERS.md#couchbaseclusterservice)<br/>[PutCouchbaseKey](PROCESSORS.md#putcouchbasekey) [...]
+| Couchbase | [CouchbaseClusterService](CONTROLLERS.md#couchbaseclusterservice)<br/>[PutCouchbaseKey](PROCESSORS.md#putcouchbasekey)<br/>[GetCouchbaseKey](PROCESSORS.md#getcouchbasekey) [...]
| Elasticsearch | [ElasticsearchCredentialsControllerService](CONTROLLERS.md#elasticsearchcredentialscontrollerservice)<br/>[PostElasticsearch](PROCESSORS.md#postelasticsearch) [...]
| ExecuteProcess (Linux and macOS) | [ExecuteProcess](PROCESSORS.md#executeprocess) [...]
| Google Cloud Platform | [DeleteGCSObject](PROCESSORS.md#deletegcsobject)<br>[FetchGCSObject](PROCESSORS.md#fetchgcsobject)<br>[GCPCredentialsControllerService](CONTROLLERS.md#gcpcredentialscontrollerservice)<br>[ListGCSBucket](PROCESSORS.md#listgcsbucket)<br>[PutGCSObject](PROCESSORS.md#putgcsobject) [...]
diff --git a/docker/test/integration/features/couchbase.feature b/docker/test/integration/features/couchbase.feature
index d9a4bb664..646eaacd9 100644
--- a/docker/test/integration/features/couchbase.feature
+++ b/docker/test/integration/features/couchbase.feature
@@ -67,17 +67,17 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1"}' of type "Binary" in Couchbase
Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor
- Given a Couchbase server is set up
- And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And the "Couchbase Cluster Controller Service" property of the GetCouchbaseKey processor is set to "CouchbaseClusterService"
And a PutFile processor with the "Directory" property set to "/tmp/output"
And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
- And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
- And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up with the name "CouchbaseClusterService"
And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
@@ -94,19 +94,19 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor using binary storage
- Given a Couchbase server is set up
- And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the PutCouchbaseKey processor is set to "Binary"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the GetCouchbaseKey processor is set to "Binary"
+ And the "Couchbase Cluster Controller Service" property of the GetCouchbaseKey processor is set to "CouchbaseClusterService"
And a PutFile processor with the "Directory" property set to "/tmp/output"
And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
- And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
- And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up with the name "CouchbaseClusterService"
And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
@@ -123,20 +123,20 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor and put the result in an attribute
- Given a Couchbase server is set up
- And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the PutCouchbaseKey processor is set to "String"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the GetCouchbaseKey processor is set to "String"
+ And the "Couchbase Cluster Controller Service" property of the GetCouchbaseKey processor is set to "CouchbaseClusterService"
And the "Put Value to Attribute" property of the GetCouchbaseKey processor is set to "get_couchbase_result"
And a PutFile processor with the "Directory" property set to "/tmp/output"
And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
- And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
- And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up with the name "CouchbaseClusterService"
And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
@@ -154,17 +154,17 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And the Minifi logs contain the following message: 'key:get_couchbase_result value:{"field1": "value1", "field2": "value2"}' in less than 1 seconds
Scenario: GetCouchbaseKey transfers FlowFile to failure relationship on Couchbase value type mismatch
- Given a Couchbase server is set up
- And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the PutCouchbaseKey processor is set to "String"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the GetCouchbaseKey processor is set to "Binary"
- And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
- And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+ And the "Couchbase Cluster Controller Service" property of the GetCouchbaseKey processor is set to "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up with the name "CouchbaseClusterService"
And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
(nifi-minifi-cpp) 15/15: Add test logs
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 396e25c6f79e6e3a3981fc0b84b9473bc0afabd2
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Oct 22 14:11:18 2024 +0200
Add test logs
---
extensions/couchbase/tests/GetCouchbaseKeyTests.cpp | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp b/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp
index 69f84aa06..d518ace03 100644
--- a/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp
+++ b/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp
@@ -38,6 +38,11 @@ struct ExpectedCallOptions {
class GetCouchbaseKeyTestController : public TestController {
public:
GetCouchbaseKeyTestController() {
+ LogTestController::getInstance().setDebug<TestPlan>();
+ LogTestController::getInstance().setDebug<minifi::core::Processor>();
+ LogTestController::getInstance().setTrace<minifi::core::ProcessSession>();
+ LogTestController::getInstance().setDebug<controllers::CouchbaseClusterService>();
+ LogTestController::getInstance().setDebug<processors::GetCouchbaseKey>();
auto controller_service_node = controller_.plan->addController("MockCouchbaseClusterService", "MockCouchbaseClusterService");
mock_couchbase_cluster_service_ = std::static_pointer_cast<MockCouchbaseClusterService>(controller_service_node->getControllerServiceImplementation());
proc_->setProperty(processors::GetCouchbaseKey::CouchbaseClusterControllerService, "MockCouchbaseClusterService");
(nifi-minifi-cpp) 01/15: MINIFICPP-2467 MINIFICPP-2468 PutCouchbaseKey processor and controller
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 9bfd7df3a274e2293b529f79577dc4bdac530b5a
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Oct 1 11:09:18 2024 +0200
MINIFICPP-2467 MINIFICPP-2468 PutCouchbaseKey processor and controller
---
.github/workflows/ci.yml | 6 +-
.github/workflows/verify-package.yml | 2 +-
CONTROLLERS.md | 18 ++
LICENSE | 154 ++++++++++++-
NOTICE | 3 +
PROCESSORS.md | 41 ++++
Windows.md | 2 +-
bootstrap.sh | 2 +
bstrp_functions.sh | 2 +
cmake/Asio.cmake | 2 +-
cmake/Bustache.cmake | 7 +
cmake/Couchbase.cmake | 45 ++++
cmake/DockerConfig.cmake | 1 +
cmake/MiNiFiOptions.cmake | 1 +
cmake/Spdlog.cmake | 4 +-
cmake/fmt.cmake | 4 +-
docker/test/integration/cluster/ContainerStore.py | 13 ++
.../test/integration/cluster/DockerTestCluster.py | 7 +-
.../integration/cluster/containers/Container.py | 4 +
.../cluster/containers/CouchbaseServerContainer.py | 55 +++++
.../features/MiNiFi_integration_test_driver.py | 21 +-
docker/test/integration/features/couchbase.feature | 41 ++++
docker/test/integration/features/steps/steps.py | 23 ++
.../minifi/controllers/CouchbaseClusterService.py | 27 +++
.../minifi/processors/PutCouchbaseKey.py | 24 ++
.../couchbase/CMakeLists.txt | 29 ++-
extensions/couchbase/CouchbaseCollection.h | 40 ++++
.../controllerservices/CouchbaseClusterService.cpp | 122 +++++++++++
.../controllerservices/CouchbaseClusterService.h | 150 +++++++++++++
.../couchbase/processors/PutCouchbaseKey.cpp | 91 ++++++++
extensions/couchbase/processors/PutCouchbaseKey.h | 152 +++++++++++++
extensions/couchbase/tests/CMakeLists.txt | 39 ++++
.../couchbase/tests/MockCouchbaseClusterService.h | 111 ++++++++++
.../couchbase/tests/PutCouchbaseKeyTests.cpp | 169 +++++++++++++++
libminifi/include/Exception.h | 3 +-
libminifi/include/utils/SmallString.h | 2 +-
libminifi/test/unit/LoggerTests.cpp | 2 +-
thirdparty/bustache/add-append.patch | 39 ++++
thirdparty/couchbase/remove-thirdparty.patch | 241 +++++++++++++++++++++
win_build_vs.bat | 4 +-
40 files changed, 1668 insertions(+), 35 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7c3770f3c..6e3dccd53 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -3,7 +3,7 @@ on: [push, pull_request, workflow_dispatch]
env:
DOCKER_CMAKE_FLAGS: -DDOCKER_VERIFY_THREAD=3 -DUSE_SHARED_LIBS= -DSTRICT_GSL_CHECKS=AUDIT -DCI_BUILD=ON -DENABLE_AWS=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_AZURE=ON -DENABLE_SQL=ON \
-DENABLE_SPLUNK=ON -DENABLE_GCP=ON -DENABLE_OPC=ON -DENABLE_PYTHON_SCRIPTING=ON -DENABLE_LUA_SCRIPTING=ON -DENABLE_KUBERNETES=ON -DENABLE_TEST_PROCESSORS=ON -DENABLE_PROMETHEUS=ON \
- -DENABLE_ELASTICSEARCH=ON -DENABLE_GRAFANA_LOKI=ON -DDOCKER_BUILD_ONLY=ON
+ -DENABLE_ELASTICSEARCH=ON -DENABLE_GRAFANA_LOKI=ON -DENABLE_COUCHBASE=ON -DDOCKER_BUILD_ONLY=ON
SCCACHE_GHA_ENABLE: true
CCACHE_DIR: ${{ GITHUB.WORKSPACE }}/.ccache
jobs:
@@ -26,6 +26,7 @@ jobs:
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
-DENABLE_CONTROLLER=ON
+ -DENABLE_COUCHBASE=ON
-DENABLE_ELASTICSEARCH=ON
-DENABLE_ENCRYPT_CONFIG=ON
-DENABLE_EXPRESSION_LANGUAGE=ON
@@ -131,6 +132,7 @@ jobs:
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
-DENABLE_CONTROLLER=ON
+ -DENABLE_COUCHBASE=ON
-DENABLE_COVERAGE=
-DENABLE_ELASTICSEARCH=ON
-DENABLE_ENCRYPT_CONFIG=ON
@@ -245,6 +247,7 @@ jobs:
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
-DENABLE_CONTROLLER=ON
+ -DENABLE_COUCHBASE=OFF
-DENABLE_ELASTICSEARCH=OFF
-DENABLE_ENCRYPT_CONFIG=ON
-DENABLE_EXPRESSION_LANGUAGE=ON
@@ -342,6 +345,7 @@ jobs:
-DENABLE_BZIP2=ON
-DENABLE_CIVET=ON
-DENABLE_CONTROLLER=ON
+ -DENABLE_COUCHBASE=ON
-DENABLE_ELASTICSEARCH=ON
-DENABLE_ENCRYPT_CONFIG=ON
-DENABLE_EXPRESSION_LANGUAGE=ON
diff --git a/.github/workflows/verify-package.yml b/.github/workflows/verify-package.yml
index b82f61512..73a69eecb 100644
--- a/.github/workflows/verify-package.yml
+++ b/.github/workflows/verify-package.yml
@@ -3,7 +3,7 @@ on: [workflow_dispatch]
env:
DOCKER_CMAKE_FLAGS: -DDOCKER_VERIFY_THREAD=3 -DUSE_SHARED_LIBS= -DSTRICT_GSL_CHECKS=AUDIT -DCI_BUILD=ON -DENABLE_AWS=ON -DENABLE_LIBRDKAFKA=ON -DENABLE_MQTT=ON -DENABLE_AZURE=ON -DENABLE_SQL=ON \
-DENABLE_SPLUNK=ON -DENABLE_GCP=ON -DENABLE_OPC=ON -DENABLE_PYTHON_SCRIPTING=ON -DENABLE_LUA_SCRIPTING=ON -DENABLE_KUBERNETES=ON -DENABLE_TEST_PROCESSORS=ON -DENABLE_PROMETHEUS=ON \
- -DENABLE_ELASTICSEARCH=OFF -DENABLE_GRAFANA_LOKI=ON -DDOCKER_BUILD_ONLY=ON
+ -DENABLE_ELASTICSEARCH=OFF -DENABLE_GRAFANA_LOKI=ON -DENABLE_COUCHBASE=ON -DDOCKER_BUILD_ONLY=ON
CCACHE_DIR: ${{ GITHUB.WORKSPACE }}/.ccache
jobs:
rocky-build:
diff --git a/CONTROLLERS.md b/CONTROLLERS.md
index e45ccbfb5..5587a1969 100644
--- a/CONTROLLERS.md
+++ b/CONTROLLERS.md
@@ -17,6 +17,7 @@ limitations under the License.
- [AWSCredentialsService](#AWSCredentialsService)
- [AzureStorageCredentialsService](#AzureStorageCredentialsService)
+- [CouchbaseClusterService](#CouchbaseClusterService)
- [ElasticsearchCredentialsControllerService](#ElasticsearchCredentialsControllerService)
- [GCPCredentialsControllerService](#GCPCredentialsControllerService)
- [JsonRecordSetReader](#JsonRecordSetReader)
@@ -71,6 +72,23 @@ In the list below, the names of required properties appear in bold. Any other pr
| **Use Managed Identity Credentials** | false | true<br/>false | If true Managed Identity credentials will be used together with the Storage Account Name for authentication. |
+## CouchbaseClusterService
+
+### Description
+
+Provides a centralized Couchbase connection and bucket passwords management. Bucket passwords can be specified via dynamic properties.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|-----------------------|---------------|------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Connection String** | | | The hostnames or ip addresses of the bootstraping nodes and optional parameters. Syntax) couchbase://node1,node2,nodeN?param1=value1¶m2=value2¶mN=valueN |
+| User Name | | | The user name to authenticate MiNiFi as a Couchbase client. |
+| User Password | | | The user password to authenticate MiNiFi as a Couchbase client.<br/>**Sensitive Property: true** |
+
+
## ElasticsearchCredentialsControllerService
### Description
diff --git a/LICENSE b/LICENSE
index 087a3fb55..00f7db88b 100644
--- a/LICENSE
+++ b/LICENSE
@@ -214,6 +214,7 @@ This product bundles 'Kubernetes Client Library for C' (kubernetes-client/c), wh
This project bundles a configuration file from 'Kubernetes Metrics Server' (kubernetes-sigs/metrics-server), which is available under an ALv2 license
This project bundles 'OpenSSL' which is available under an ALv2 license
This project bundles 'gRPC' which is available under an ALv2 license
+This project bundles 'couchbase-cxx-clien' which is available under an ALv2 license
The Apache NiFi - MiNiFi C++ project contains subcomponents with separate copyright
notices and license terms. Your use of the source code for the these
@@ -1002,7 +1003,7 @@ These libraries are in the public domain (or the equivalent where that is not po
regexp.c and regexp.h from https://github.com/ccxvii/minilibs sha 875c33568b5a4aa4fb3dd0c52ea98f7f0e5ca684
-This product bundles snappy within librdkafka, under the license below.
+This product bundles snappy within librdkafka and couchbase-cxx-client, under the license below.
LICENSE.snappy
--------------------------------------------------------------
@@ -3307,3 +3308,154 @@ Permission is hereby granted, free of charge, to any person obtaining a copy of
The above copyright notice and this permission notice (including the next paragraph) shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+
+This product bundles HdrHistogram within couchbase-cxx-client, under the license below.
+
+Creative Commons Legal Code
+
+CC0 1.0 Universal
+
+ CREATIVE COMMONS CORPORATION IS NOT A LAW FIRM AND DOES NOT PROVIDE
+ LEGAL SERVICES. DISTRIBUTION OF THIS DOCUMENT DOES NOT CREATE AN
+ ATTORNEY-CLIENT RELATIONSHIP. CREATIVE COMMONS PROVIDES THIS
+ INFORMATION ON AN "AS-IS" BASIS. CREATIVE COMMONS MAKES NO WARRANTIES
+ REGARDING THE USE OF THIS DOCUMENT OR THE INFORMATION OR WORKS
+ PROVIDED HEREUNDER, AND DISCLAIMS LIABILITY FOR DAMAGES RESULTING FROM
+ THE USE OF THIS DOCUMENT OR THE INFORMATION OR WORKS PROVIDED
+ HEREUNDER.
+
+Statement of Purpose
+
+The laws of most jurisdictions throughout the world automatically confer
+exclusive Copyright and Related Rights (defined below) upon the creator
+and subsequent owner(s) (each and all, an "owner") of an original work of
+authorship and/or a database (each, a "Work").
+
+Certain owners wish to permanently relinquish those rights to a Work for
+the purpose of contributing to a commons of creative, cultural and
+scientific works ("Commons") that the public can reliably and without fear
+of later claims of infringement build upon, modify, incorporate in other
+works, reuse and redistribute as freely as possible in any form whatsoever
+and for any purposes, including without limitation commercial purposes.
+These owners may contribute to the Commons to promote the ideal of a free
+culture and the further production of creative, cultural and scientific
+works, or to gain reputation or greater distribution for their Work in
+part through the use and efforts of others.
+
+For these and/or other purposes and motivations, and without any
+expectation of additional consideration or compensation, the person
+associating CC0 with a Work (the "Affirmer"), to the extent that he or she
+is an owner of Copyright and Related Rights in the Work, voluntarily
+elects to apply CC0 to the Work and publicly distribute the Work under its
+terms, with knowledge of his or her Copyright and Related Rights in the
+Work and the meaning and intended legal effect of CC0 on those rights.
+
+1. Copyright and Related Rights. A Work made available under CC0 may be
+protected by copyright and related or neighboring rights ("Copyright and
+Related Rights"). Copyright and Related Rights include, but are not
+limited to, the following:
+
+ i. the right to reproduce, adapt, distribute, perform, display,
+ communicate, and translate a Work;
+ ii. moral rights retained by the original author(s) and/or performer(s);
+iii. publicity and privacy rights pertaining to a person's image or
+ likeness depicted in a Work;
+ iv. rights protecting against unfair competition in regards to a Work,
+ subject to the limitations in paragraph 4(a), below;
+ v. rights protecting the extraction, dissemination, use and reuse of data
+ in a Work;
+ vi. database rights (such as those arising under Directive 96/9/EC of the
+ European Parliament and of the Council of 11 March 1996 on the legal
+ protection of databases, and under any national implementation
+ thereof, including any amended or successor version of such
+ directive); and
+vii. other similar, equivalent or corresponding rights throughout the
+ world based on applicable law or treaty, and any national
+ implementations thereof.
+
+2. Waiver. To the greatest extent permitted by, but not in contravention
+of, applicable law, Affirmer hereby overtly, fully, permanently,
+irrevocably and unconditionally waives, abandons, and surrenders all of
+Affirmer's Copyright and Related Rights and associated claims and causes
+of action, whether now known or unknown (including existing as well as
+future claims and causes of action), in the Work (i) in all territories
+worldwide, (ii) for the maximum duration provided by applicable law or
+treaty (including future time extensions), (iii) in any current or future
+medium and for any number of copies, and (iv) for any purpose whatsoever,
+including without limitation commercial, advertising or promotional
+purposes (the "Waiver"). Affirmer makes the Waiver for the benefit of each
+member of the public at large and to the detriment of Affirmer's heirs and
+successors, fully intending that such Waiver shall not be subject to
+revocation, rescission, cancellation, termination, or any other legal or
+equitable action to disrupt the quiet enjoyment of the Work by the public
+as contemplated by Affirmer's express Statement of Purpose.
+
+3. Public License Fallback. Should any part of the Waiver for any reason
+be judged legally invalid or ineffective under applicable law, then the
+Waiver shall be preserved to the maximum extent permitted taking into
+account Affirmer's express Statement of Purpose. In addition, to the
+extent the Waiver is so judged Affirmer hereby grants to each affected
+person a royalty-free, non transferable, non sublicensable, non exclusive,
+irrevocable and unconditional license to exercise Affirmer's Copyright and
+Related Rights in the Work (i) in all territories worldwide, (ii) for the
+maximum duration provided by applicable law or treaty (including future
+time extensions), (iii) in any current or future medium and for any number
+of copies, and (iv) for any purpose whatsoever, including without
+limitation commercial, advertising or promotional purposes (the
+"License"). The License shall be deemed effective as of the date CC0 was
+applied by Affirmer to the Work. Should any part of the License for any
+reason be judged legally invalid or ineffective under applicable law, such
+partial invalidity or ineffectiveness shall not invalidate the remainder
+of the License, and in such case Affirmer hereby affirms that he or she
+will not (i) exercise any of his or her remaining Copyright and Related
+Rights in the Work or (ii) assert any associated claims and causes of
+action with respect to the Work, in either case contrary to Affirmer's
+express Statement of Purpose.
+
+4. Limitations and Disclaimers.
+
+ a. No trademark or patent rights held by Affirmer are waived, abandoned,
+ surrendered, licensed or otherwise affected by this document.
+ b. Affirmer offers the Work as-is and makes no representations or
+ warranties of any kind concerning the Work, express, implied,
+ statutory or otherwise, including without limitation warranties of
+ title, merchantability, fitness for a particular purpose, non
+ infringement, or the absence of latent or other defects, accuracy, or
+ the present or absence of errors, whether or not discoverable, all to
+ the greatest extent permissible under applicable law.
+ c. Affirmer disclaims responsibility for clearing rights of other persons
+ that may apply to the Work or any use thereof, including without
+ limitation any person's Copyright and Related Rights in the Work.
+ Further, Affirmer disclaims responsibility for obtaining any necessary
+ consents, permissions or other rights required for any use of the
+ Work.
+ d. Affirmer understands and acknowledges that Creative Commons is not a
+ party to this document and has no duty or obligation with respect to
+ this CC0 or use of the Work.
+
+
+This product bundles llhttp within couchbase-cxx-client, under the license below.
+
+This software is licensed under the MIT License.
+
+Copyright Fedor Indutny, 2018.
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to permit
+persons to whom the Software is furnished to do so, subject to the
+following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/NOTICE b/NOTICE
index 8a2390bb3..81a9644b2 100644
--- a/NOTICE
+++ b/NOTICE
@@ -73,6 +73,9 @@ This software includes third party software subject to the following copyrights:
- RE2 - Copyright (c) 2009 The RE2 Authors. All rights reserved.
- c-ares - Copyright (c) 1998 Massachusetts Institute of Technology Copyright (c) 2007 - 2023 Daniel Stenberg with many contributors, see AUTHORS file.
- lua - Copyright (c) 1994–2023 Lua.org, PUC-Rio.
+- couchbase-cxx-client - Copyright 2023-Present Couchbase, Inc.
+- snappy - Copyright 2011, Google Inc.
+- llhttp - Copyright Fedor Indutny, 2018.
The licenses for these third party components are included in LICENSE.txt
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 9a09eae09..df3148523 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -77,6 +77,7 @@ limitations under the License.
- [PushGrafanaLokiREST](#PushGrafanaLokiREST)
- [PutAzureBlobStorage](#PutAzureBlobStorage)
- [PutAzureDataLakeStorage](#PutAzureDataLakeStorage)
+- [PutCouchbaseKey](#PutCouchbaseKey)
- [PutFile](#PutFile)
- [PutGCSObject](#PutGCSObject)
- [PutOPCProcessor](#PutOPCProcessor)
@@ -2202,6 +2203,46 @@ In the list below, the names of required properties appear in bold. Any other pr
| failure | Files that could not be written to Azure storage for some reason are transferred to this relationship |
+## PutCouchbaseKey
+
+### Description
+
+Put a document to Couchbase Server via Key/Value access.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|------------------------------------------|---------------|----------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Couchbase Cluster Controller Service** | | | A Couchbase Cluster Controller Service which manages connections to a Couchbase cluster. |
+| **Bucket Name** | default | | The name of bucket to access.<br/>**Supports Expression Language: true** |
+| Scope Name | | | Scope to use inside the bucket. If not specified, the _default scope is used.<br/>**Supports Expression Language: true** |
+| Collection Name | | | Collection to use inside the bucket scope. If not specified, the _default collection is used.<br/>**Supports Expression Language: true** |
+| Document Id | | | A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.<br/>**Supports Expression Language: true** |
+| **Persist To** | NONE | NONE<br/>ACTIVE<br/>ONE<br/>TWO<br/>THREE<br/>FOUR | Durability constraint about disk persistence. |
+| **Replicate To** | NONE | NONE<br/>ONE<br/>TWO<br/>THREE | Durability constraint about replication. |
+
+### Relationships
+
+| Name | Description |
+|---------|------------------------------------------------------------------------------------------------------------|
+| success | All FlowFiles that are written to Couchbase Server are routed to this relationship. |
+| failure | All FlowFiles failed to be written to Couchbase Server and not retry-able are routed to this relationship. |
+| retry | All FlowFiles failed to be written to Couchbase Server but can be retried are routed to this relationship. |
+
+### Output Attributes
+
+| Attribute | Relationship | Description |
+|-------------------------------|--------------|-----------------------------------------------|
+| couchbase.bucket | success | Bucket where the document was stored. |
+| couchbase.doc.id | success | Id of the document. |
+| couchbase.doc.cas | success | CAS of the document. |
+| couchbase.doc.sequence.number | success | Sequence number associated with the document. |
+| couchbase.partition.uuid | success | UUID of partition. |
+| couchbase.partition.id | success | ID of partition (also known as vBucket). |
+
+
## PutFile
### Description
diff --git a/Windows.md b/Windows.md
index a87abd85f..9a2e5b916 100644
--- a/Windows.md
+++ b/Windows.md
@@ -122,7 +122,7 @@ A basic working CMake configuration can be inferred from the `win_build_vs.bat`.
```
mkdir build
cd build
-cmake -G "Visual Studio 17 2022" -A x64 -DMINIFI_INCLUDE_VC_REDIST_MERGE_MODULES=OFF -DTEST_CUSTOM_WEL_PROVIDER=OFF -DENABLE_SQL=OFF -DMINIFI_USE_REAL_ODBC_TEST_DRIVER=OFF -DCMAKE_BUILD_TYPE_INIT=Release -DCMAKE_BUILD_TYPE=Release -DWIN32=WIN32 -DENABLE_LIBRDKAFKA=OFF -DENABLE_AWS=OFF -DENABLE_PDH= -DENABLE_AZURE=OFF -DENABLE_SFTP=OFF -DENABLE_SPLUNK= -DENABLE_GCP= -DENABLE_OPENCV=OFF -DENABLE_PROMETHEUS=OFF -DENABLE_ELASTICSEARCH= -DUSE_SHARED_LIBS=OFF -DENABLE_CONTROLLER=ON -DENABLE_BU [...]
+cmake -G "Visual Studio 17 2022" -A x64 -DMINIFI_INCLUDE_VC_REDIST_MERGE_MODULES=OFF -DTEST_CUSTOM_WEL_PROVIDER=OFF -DENABLE_SQL=OFF -DMINIFI_USE_REAL_ODBC_TEST_DRIVER=OFF -DCMAKE_BUILD_TYPE_INIT=Release -DCMAKE_BUILD_TYPE=Release -DWIN32=WIN32 -DENABLE_LIBRDKAFKA=OFF -DENABLE_AWS=OFF -DENABLE_PDH= -DENABLE_AZURE=OFF -DENABLE_SFTP=OFF -DENABLE_SPLUNK= -DENABLE_GCP= -DENABLE_OPENCV=OFF -DENABLE_PROMETHEUS=OFF -DENABLE_ELASTICSEARCH= -DUSE_SHARED_LIBS=OFF -DENABLE_CONTROLLER=ON -DENABLE_BU [...]
msbuild /m nifi-minifi-cpp.sln /property:Configuration=Release /property:Platform=x64
copy minifi_main\Release\minifi.exe minifi_main\
cpack
diff --git a/bootstrap.sh b/bootstrap.sh
index 45e630ade..6b331de18 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -316,6 +316,8 @@ add_option PROCFS_ENABLED ${TRUE} "ENABLE_PROCFS"
add_option PROMETHEUS_ENABLED ${TRUE} "ENABLE_PROMETHEUS"
+add_option COUCHBASE_ENABLED ${FALSE} "ENABLE_COUCHBASE"
+
USE_SHARED_LIBS=${TRUE}
ASAN_ENABLED=${FALSE}
MINIFI_FAIL_ON_WARNINGS=${FALSE}
diff --git a/bstrp_functions.sh b/bstrp_functions.sh
index 471938d05..7311eb166 100755
--- a/bstrp_functions.sh
+++ b/bstrp_functions.sh
@@ -401,6 +401,7 @@ show_supported_features() {
echo "AE. Prometheus Support .........$(print_feature_status PROMETHEUS_ENABLED)"
echo "AF. Elasticsearch Support ......$(print_feature_status ELASTIC_ENABLED)"
echo "AG. Grafana Loki Support .......$(print_feature_status GRAFANA_LOKI_ENABLED)"
+ echo "AH. Couchbase Support ..........$(print_feature_status COUCHBASE_ENABLED)"
echo "****************************************"
echo " Build Options."
echo "****************************************"
@@ -448,6 +449,7 @@ read_feature_options(){
ae) ToggleFeature PROMETHEUS_ENABLED ;;
af) ToggleFeature ELASTIC_ENABLED ;;
ag) ToggleFeature GRAFANA_LOKI_ENABLED ;;
+ ah) ToggleFeature COUCHBASE_ENABLED ;;
1) ToggleFeature TESTS_ENABLED ;;
2) EnableAllFeatures ;;
4) ToggleFeature USE_SHARED_LIBS;;
diff --git a/cmake/Asio.cmake b/cmake/Asio.cmake
index 28f4bb928..6f94e6afe 100644
--- a/cmake/Asio.cmake
+++ b/cmake/Asio.cmake
@@ -27,5 +27,5 @@ if(NOT asio_POPULATED)
add_library(asio INTERFACE)
target_include_directories(asio SYSTEM INTERFACE ${asio_SOURCE_DIR}/asio/include)
find_package(Threads)
- target_link_libraries(asio INTERFACE Threads::Threads)
+ target_link_libraries(asio INTERFACE Threads::Threads OpenSSL::SSL OpenSSL::Crypto)
endif()
diff --git a/cmake/Bustache.cmake b/cmake/Bustache.cmake
index 6c037af6c..e3d104cda 100644
--- a/cmake/Bustache.cmake
+++ b/cmake/Bustache.cmake
@@ -21,8 +21,15 @@ include(GetFmt)
get_fmt()
set(BUSTACHE_USE_FMT ON CACHE STRING "" FORCE)
+
+set(PATCH_FILE "${CMAKE_SOURCE_DIR}/thirdparty/bustache/add-append.patch")
+
+set(PC ${Bash_EXECUTABLE} -c "set -x &&\
+ (\\\"${Patch_EXECUTABLE}\\\" -p1 -R -s -f --dry-run -i \\\"${PATCH_FILE}\\\" || \\\"${Patch_EXECUTABLE}\\\" -p1 -N -i \\\"${PATCH_FILE}\\\")")
+
FetchContent_Declare(Bustache
GIT_REPOSITORY https://github.com/jamboree/bustache.git
GIT_TAG 47096caa8e1f9f7ebe34e3a022dbb822c174011d
+ PATCH_COMMAND "${PC}"
)
FetchContent_MakeAvailable(Bustache)
diff --git a/cmake/Couchbase.cmake b/cmake/Couchbase.cmake
new file mode 100644
index 000000000..cf4ff7fb1
--- /dev/null
+++ b/cmake/Couchbase.cmake
@@ -0,0 +1,45 @@
+# 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.
+
+include(FetchContent)
+
+include(fmt)
+include(Spdlog)
+include(Asio)
+
+set(COUCHBASE_CXX_CLIENT_BUILD_STATIC ON CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_BUILD_SHARED OFF CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_BUILD_TESTS OFF CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_BUILD_DOCS OFF CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_BUILD_EXAMPLES OFF CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_BUILD_TOOLS OFF CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_POST_LINKED_OPENSSL ON CACHE BOOL "" FORCE)
+set(COUCHBASE_CXX_CLIENT_INSTALL OFF CACHE BOOL "" FORCE)
+
+set(PATCH_FILE "${CMAKE_SOURCE_DIR}/thirdparty/couchbase/remove-thirdparty.patch")
+
+set(PC ${Bash_EXECUTABLE} -c "set -x &&\
+ (\\\"${Patch_EXECUTABLE}\\\" -p1 -R -s -f --dry-run -i \\\"${PATCH_FILE}\\\" || \\\"${Patch_EXECUTABLE}\\\" -p1 -N -i \\\"${PATCH_FILE}\\\")")
+
+FetchContent_Declare(couchbase-cxx-client
+ URL https://github.com/couchbase/couchbase-cxx-client/releases/download/1.0.2/couchbase-cxx-client-1.0.2.tar.gz
+ URL_HASH SHA256=1954e6f5e063d94675428182bc8b1b82fd8e8532c10d1787f157aeb18bb37769
+ PATCH_COMMAND "${PC}"
+)
+FetchContent_MakeAvailable(couchbase-cxx-client)
+
+set(COUCHBASE_INCLUDE_DIR "${couchbase-cxx-client_SOURCE_DIR}" CACHE STRING "" FORCE)
diff --git a/cmake/DockerConfig.cmake b/cmake/DockerConfig.cmake
index 7346de263..860503387 100644
--- a/cmake/DockerConfig.cmake
+++ b/cmake/DockerConfig.cmake
@@ -67,6 +67,7 @@ add_custom_target(
-DENABLE_OPENCV=OFF
-DENABLE_BUSTACHE=OFF
-DENABLE_SFTP=OFF
+ -DENABLE_COUCHBASE=OFF
-DENABLE_TEST_PROCESSORS=OFF
-DENABLE_ROCKSDB=ON
-DENABLE_LIBARCHIVE=ON
diff --git a/cmake/MiNiFiOptions.cmake b/cmake/MiNiFiOptions.cmake
index 8a7ccc58c..e69548ba0 100644
--- a/cmake/MiNiFiOptions.cmake
+++ b/cmake/MiNiFiOptions.cmake
@@ -111,6 +111,7 @@ add_minifi_option(ENABLE_TEST_PROCESSORS "Enables test processors" OFF)
add_minifi_option(ENABLE_PROMETHEUS "Enables Prometheus support." ON)
add_minifi_option(ENABLE_GRAFANA_LOKI "Enable Grafana Loki support" OFF)
add_minifi_option(ENABLE_GRPC_FOR_LOKI "Enable gRPC for Grafana Loki extension" ON)
+add_minifi_option(ENABLE_COUCHBASE "Enable Couchbase support" OFF)
add_minifi_option(ENABLE_EXECUTE_PROCESS "Enable ExecuteProcess processor" OFF)
add_minifi_option(ENABLE_CONTROLLER "Enables the build of MiNiFi controller binary." ON)
diff --git a/cmake/Spdlog.cmake b/cmake/Spdlog.cmake
index 3b0729829..0631ca783 100644
--- a/cmake/Spdlog.cmake
+++ b/cmake/Spdlog.cmake
@@ -20,8 +20,8 @@ include(FetchContent)
set(SPDLOG_FMT_EXTERNAL ON CACHE STRING "" FORCE)
FetchContent_Declare(Spdlog
- URL https://github.com/gabime/spdlog/archive/refs/tags/v1.12.0.tar.gz
- URL_HASH SHA256=4dccf2d10f410c1e2feaff89966bfc49a1abb29ef6f08246335b110e001e09a9
+ URL https://github.com/gabime/spdlog/archive/refs/tags/v1.14.1.tar.gz
+ URL_HASH SHA256=1586508029a7d0670dfcb2d97575dcdc242d3868a259742b69f100801ab4e16b
OVERRIDE_FIND_PACKAGE
)
FetchContent_MakeAvailable(Spdlog)
diff --git a/cmake/fmt.cmake b/cmake/fmt.cmake
index fc48799ce..93fd56efb 100644
--- a/cmake/fmt.cmake
+++ b/cmake/fmt.cmake
@@ -18,8 +18,8 @@
#
include(FetchContent)
FetchContent_Declare(Fmt
- URL https://github.com/fmtlib/fmt/archive/refs/tags/10.1.0.tar.gz
- URL_HASH SHA256=deb0a3ad2f5126658f2eefac7bf56a042488292de3d7a313526d667f3240ca0a
+ URL https://github.com/fmtlib/fmt/archive/refs/tags/11.0.2.tar.gz
+ URL_HASH SHA256=6cb1e6d37bdcb756dbbe59be438790db409cdb4868c66e888d5df9f13f7c027f
OVERRIDE_FIND_PACKAGE
)
FetchContent_MakeAvailable(Fmt)
diff --git a/docker/test/integration/cluster/ContainerStore.py b/docker/test/integration/cluster/ContainerStore.py
index f1e13a894..9c8527458 100644
--- a/docker/test/integration/cluster/ContainerStore.py
+++ b/docker/test/integration/cluster/ContainerStore.py
@@ -40,6 +40,7 @@ from .containers.GrafanaLokiContainer import GrafanaLokiContainer
from .containers.GrafanaLokiContainer import GrafanaLokiOptions
from .containers.ReverseProxyContainer import ReverseProxyContainer
from .containers.DiagSlave import DiagSlave
+from .containers.CouchbaseServerContainer import CouchbaseServerContainer
from .FeatureContext import FeatureContext
@@ -302,6 +303,14 @@ class ContainerStore:
network=self.network,
image_store=self.image_store,
command=command))
+ elif engine == "couchbase-server":
+ return self.containers.setdefault(container_name,
+ CouchbaseServerContainer(feature_context=feature_context,
+ name=container_name,
+ vols=self.vols,
+ network=self.network,
+ image_store=self.image_store,
+ command=command))
else:
raise Exception('invalid flow engine: \'%s\'' % engine)
@@ -411,3 +420,7 @@ class ContainerStore:
def enable_ssl_in_nifi(self):
self.nifi_options.use_ssl = True
+
+ def run_post_startup_commands(self, container_name):
+ container_name = self.get_container_name_with_postfix(container_name)
+ return self.containers[container_name].run_post_startup_commands()
diff --git a/docker/test/integration/cluster/DockerTestCluster.py b/docker/test/integration/cluster/DockerTestCluster.py
index 153ffce26..f5fb913b7 100644
--- a/docker/test/integration/cluster/DockerTestCluster.py
+++ b/docker/test/integration/cluster/DockerTestCluster.py
@@ -305,8 +305,11 @@ class DockerTestCluster:
startup_success = self.wait_for_startup_log(container_name, 300)
if not startup_success:
logging.error("Cluster startup failed for %s", container_name)
- self.log_app_output()
- return startup_success
+ return False
+ if not self.container_store.run_post_startup_commands(container_name):
+ logging.error("Failed to run post startup commands for container %s", container_name)
+ return False
+ return True
def wait_for_all_containers_to_finish_startup(self):
for container_name in self.container_store.get_container_names():
diff --git a/docker/test/integration/cluster/containers/Container.py b/docker/test/integration/cluster/containers/Container.py
index b4a8efe85..d2e73fd4b 100644
--- a/docker/test/integration/cluster/containers/Container.py
+++ b/docker/test/integration/cluster/containers/Container.py
@@ -34,6 +34,7 @@ class Container:
# Get docker client
self.client = docker.from_env()
self.deployed = False
+ self.post_startup_commands_finished = False
def cleanup(self):
logging.info('Cleaning up container: %s', self.name)
@@ -84,3 +85,6 @@ class Container:
def get_app_log(self):
raise NotImplementedError()
+
+ def run_post_startup_commands(self):
+ return True
diff --git a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
new file mode 100644
index 000000000..a964690fa
--- /dev/null
+++ b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
@@ -0,0 +1,55 @@
+# 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.
+from .Container import Container
+from utils import retry_check
+
+
+class CouchbaseServerContainer(Container):
+ def __init__(self, feature_context, name, vols, network, image_store, command=None):
+ super().__init__(feature_context, name, 'couchbase-server', vols, network, image_store, command)
+
+ def get_startup_finished_log_entry(self):
+ return "logs available in"
+
+ @retry_check(15, 2)
+ def run_post_startup_commands(self):
+ if self.post_startup_commands_finished:
+ return True
+
+ commands = [
+ ["couchbase-cli", "cluster-init", "-c", "localhost", "--cluster-username", "Administrator", "--cluster-password", "password123", "--services", "data,index,query",
+ "--cluster-ramsize", "2048", "--cluster-index-ramsize", "256"],
+ ["couchbase-cli", "bucket-create", "-c", "localhost", "--username", "Administrator", "--password", "password123", "--bucket", "test_bucket", "--bucket-type", "couchbase",
+ "--bucket-ramsize", "1024"]
+ ]
+ for command in commands:
+ (code, _) = self.client.containers.get(self.name).exec_run(command)
+ if code != 0:
+ return False
+ self.post_startup_commands_finished = True
+ return True
+
+ def deploy(self):
+ if not self.set_deployed():
+ return
+
+ port_list = [*range(8091, 8098), 9123, 11207, 11210, 11280, *range(18091, 18097)]
+ self.docker_container = self.client.containers.run(
+ "couchbase:community-7.6.2",
+ detach=True,
+ name=self.name,
+ network=self.network.name,
+ ports={f'{port}/tcp': port for port in port_list},
+ entrypoint=self.command)
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index 2eef2f22d..40b2fe56c 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -63,38 +63,43 @@ class MiNiFi_integration_test:
self.cluster.acquire_container(context=context, name='kafka-broker', engine='kafka-broker')
self.cluster.deploy_container(name='zookeeper')
self.cluster.deploy_container(name='kafka-broker')
- assert self.cluster.wait_for_container_startup_to_finish('kafka-broker')
+ assert self.cluster.wait_for_container_startup_to_finish('kafka-broker') or self.cluster.log_app_output()
def start_splunk(self, context):
self.cluster.acquire_container(context=context, name='splunk', engine='splunk')
self.cluster.deploy_container(name='splunk')
- assert self.cluster.wait_for_container_startup_to_finish('splunk')
- assert self.cluster.enable_splunk_hec_indexer('splunk', 'splunk_hec_token')
+ assert self.cluster.wait_for_container_startup_to_finish('splunk') or self.cluster.log_app_output()
+ assert self.cluster.enable_splunk_hec_indexer('splunk', 'splunk_hec_token') or self.cluster.log_app_output()
def start_elasticsearch(self, context):
self.cluster.acquire_container(context=context, name='elasticsearch', engine='elasticsearch')
self.cluster.deploy_container('elasticsearch')
- assert self.cluster.wait_for_container_startup_to_finish('elasticsearch')
+ assert self.cluster.wait_for_container_startup_to_finish('elasticsearch') or self.cluster.log_app_output()
def start_opensearch(self, context):
self.cluster.acquire_container(context=context, name='opensearch', engine='opensearch')
self.cluster.deploy_container('opensearch')
- assert self.cluster.wait_for_container_startup_to_finish('opensearch')
+ assert self.cluster.wait_for_container_startup_to_finish('opensearch') or self.cluster.log_app_output()
def start_minifi_c2_server(self, context):
self.cluster.acquire_container(context=context, name="minifi-c2-server", engine="minifi-c2-server")
self.cluster.deploy_container('minifi-c2-server')
- assert self.cluster.wait_for_container_startup_to_finish('minifi-c2-server')
+ assert self.cluster.wait_for_container_startup_to_finish('minifi-c2-server') or self.cluster.log_app_output()
+
+ def start_couchbase_server(self, context):
+ self.cluster.acquire_container(context=context, name='couchbase-server', engine='couchbase-server')
+ self.cluster.deploy_container('couchbase-server')
+ assert self.cluster.wait_for_container_startup_to_finish('couchbase-server') or self.cluster.log_app_output()
def start(self, container_name=None):
if container_name is not None:
logging.info("Starting container %s", container_name)
self.cluster.deploy_container(container_name)
- assert self.cluster.wait_for_container_startup_to_finish(container_name)
+ assert self.cluster.wait_for_container_startup_to_finish(container_name) or self.cluster.log_app_output()
return
logging.info("MiNiFi_integration_test start")
self.cluster.deploy_all()
- assert self.cluster.wait_for_all_containers_to_finish_startup()
+ assert self.cluster.wait_for_all_containers_to_finish_startup() or self.cluster.log_app_output()
def stop(self, container_name):
logging.info("Stopping container %s", container_name)
diff --git a/docker/test/integration/features/couchbase.feature b/docker/test/integration/features/couchbase.feature
new file mode 100644
index 000000000..2e4fcd8e6
--- /dev/null
+++ b/docker/test/integration/features/couchbase.feature
@@ -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.
+
+@ENABLE_COUCHBASE
+Feature: Executing Couchbase operations from MiNiFi-C++
+ Background:
+ Given the content of "/tmp/output" is monitored
+
+ Scenario: A MiNiFi instance can insert data to test bucket with PutCouchbaseKey processor
+ Given a Couchbase server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 60 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.sequence.number value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.partition.uuid value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.partition.id value:[1-9][0-9]*" in less than 1 seconds
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 92dd64548..7f5857f67 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -25,6 +25,7 @@ from minifi.controllers.ElasticsearchCredentialsService import ElasticsearchCred
from minifi.controllers.ODBCService import ODBCService
from minifi.controllers.KubernetesControllerService import KubernetesControllerService
from minifi.controllers.JsonRecordSetWriter import JsonRecordSetWriter
+from minifi.controllers.CouchbaseClusterService import CouchbaseClusterService
from behave import given, then, when
from behave.model_describe import ModelDescriptor
@@ -271,6 +272,7 @@ def step_impl(context):
@given("a file with the content \"{content}\" is present in \"{path}\"")
+@given("a file with the content '{content}' is present in '{path}'")
@then("a file with the content \"{content}\" is placed in \"{path}\"")
def step_impl(context, content, path):
context.test.add_test_data(path, content)
@@ -1354,3 +1356,24 @@ def step_impl(context):
@given(u'PLC register has been set with {modbus_cmd} command')
def step_impl(context, modbus_cmd):
context.test.set_value_on_plc_with_modbus(context.test.get_container_name_with_postfix('diag-slave-tcp'), modbus_cmd)
+
+
+# Couchbase
+@given("a Couchbase server is set up")
+def step_impl(context):
+ context.test.acquire_container(context=context, name="couchbase-server", engine="couchbase-server")
+
+
+@when(u'a Couchbase server is started')
+def step_impl(context):
+ context.test.start_couchbase_server(context)
+
+
+@given("a CouchbaseClusterService is setup up for {processor_name} with the name \"{service_name}\"")
+def step_impl(context, processor_name, service_name):
+ couchbase_cluster_controller_service = CouchbaseClusterService(
+ name=service_name,
+ connection_string="couchbase://{server_hostname}".format(server_hostname=context.test.get_container_name_with_postfix("couchbase-server")))
+ processor = context.test.get_node_by_name(processor_name)
+ processor.controller_services.append(couchbase_cluster_controller_service)
+ processor.set_property("Couchbase Cluster Controller Service", couchbase_cluster_controller_service.name)
diff --git a/docker/test/integration/minifi/controllers/CouchbaseClusterService.py b/docker/test/integration/minifi/controllers/CouchbaseClusterService.py
new file mode 100644
index 000000000..e06c1e6ff
--- /dev/null
+++ b/docker/test/integration/minifi/controllers/CouchbaseClusterService.py
@@ -0,0 +1,27 @@
+# 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.
+
+
+from ..core.ControllerService import ControllerService
+
+
+class CouchbaseClusterService(ControllerService):
+ def __init__(self, name, connection_string):
+ super(CouchbaseClusterService, self).__init__(name=name)
+
+ self.service_class = 'CouchbaseClusterService'
+ self.properties['Connection String'] = connection_string
+ self.properties['User Name'] = "Administrator"
+ self.properties['User Password'] = "password123"
diff --git a/docker/test/integration/minifi/processors/PutCouchbaseKey.py b/docker/test/integration/minifi/processors/PutCouchbaseKey.py
new file mode 100644
index 000000000..5e94aaa07
--- /dev/null
+++ b/docker/test/integration/minifi/processors/PutCouchbaseKey.py
@@ -0,0 +1,24 @@
+# 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.
+from ..core.Processor import Processor
+
+
+class PutCouchbaseKey(Processor):
+ def __init__(self, context, schedule={'scheduling strategy': 'EVENT_DRIVEN'}):
+ super(PutCouchbaseKey, self).__init__(
+ context=context,
+ clazz='PutCouchbaseKey',
+ auto_terminate=['success', 'failure'],
+ schedule=schedule)
diff --git a/cmake/Asio.cmake b/extensions/couchbase/CMakeLists.txt
similarity index 51%
copy from cmake/Asio.cmake
copy to extensions/couchbase/CMakeLists.txt
index 28f4bb928..0299ce8fd 100644
--- a/cmake/Asio.cmake
+++ b/extensions/couchbase/CMakeLists.txt
@@ -1,3 +1,4 @@
+#
# 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
@@ -14,18 +15,22 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+#
-include(FetchContent)
+if (NOT (ENABLE_ALL OR ENABLE_COUCHBASE))
+ return()
+endif()
-FetchContent_Declare(asio
- URL https://github.com/chriskohlhoff/asio/archive/refs/tags/asio-1-28-1.tar.gz
- URL_HASH SHA256=5ff6111ec8cbe73a168d997c547f562713aa7bd004c5c02326f0e9d579a5f2ce)
+include(Couchbase)
+include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)
-FetchContent_GetProperties(asio)
-if(NOT asio_POPULATED)
- FetchContent_Populate(asio)
- add_library(asio INTERFACE)
- target_include_directories(asio SYSTEM INTERFACE ${asio_SOURCE_DIR}/asio/include)
- find_package(Threads)
- target_link_libraries(asio INTERFACE Threads::Threads)
-endif()
+file(GLOB SOURCES "*.cpp" "controllerservices/*.cpp" "processors/*.cpp")
+
+add_minifi_library(minifi-couchbase SHARED ${SOURCES})
+
+target_include_directories(minifi-couchbase PRIVATE BEFORE "${CMAKE_CURRENT_SOURCE_DIR}" "controllerservices" "processors" ${COUCHBASE_INCLUDE_DIR})
+target_link_libraries(minifi-couchbase ${LIBMINIFI} couchbase_cxx_client_static hdr_histogram_static snappy llhttp::llhttp)
+
+register_extension(minifi-couchbase "COUCHBASE EXTENSIONS" COUCHBASE-EXTENSIONS "This enables Couchbase support" "extensions/couchbase/tests")
+
+register_extension_linter(minifi-couchbase-extensions-linter)
diff --git a/extensions/couchbase/CouchbaseCollection.h b/extensions/couchbase/CouchbaseCollection.h
new file mode 100644
index 000000000..7968bdd8c
--- /dev/null
+++ b/extensions/couchbase/CouchbaseCollection.h
@@ -0,0 +1,40 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include "couchbase/upsert_options.hxx"
+#include "nonstd/expected.hpp"
+
+namespace org::apache::nifi::minifi::couchbase {
+
+struct CouchbaseUpsertResult {
+ std::string bucket_name;
+ std::uint64_t cas{0};
+ std::uint64_t sequence_number{0};
+ std::uint64_t partition_uuid{0};
+ std::uint16_t partition_id{0};
+};
+
+class CouchbaseCollection {
+ public:
+ virtual nonstd::expected<CouchbaseUpsertResult, std::error_code> upsert(const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) = 0;
+ virtual ~CouchbaseCollection() = default;
+};
+
+} // namespace org::apache::nifi::minifi::couchbase
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
new file mode 100644
index 000000000..02b43283a
--- /dev/null
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -0,0 +1,122 @@
+/**
+ *
+ * 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.
+ */
+
+#include "CouchbaseClusterService.h"
+
+#include "core/Resource.h"
+
+namespace org::apache::nifi::minifi::couchbase {
+
+nonstd::expected<CouchbaseUpsertResult, std::error_code> RemoteCouchbaseCollection::upsert(const std::string& document_id, const std::vector<std::byte>& buffer,
+ const ::couchbase::upsert_options& options) {
+ auto [err, resp] = collection_.upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
+ if (err.ec()) {
+ client_.setConnectionError();
+ return nonstd::make_unexpected(err.ec());
+ } else {
+ const uint64_t partition_uuid = (resp.mutation_token().has_value() ? resp.mutation_token()->partition_uuid() : 0);
+ const uint64_t sequence_number = (resp.mutation_token().has_value() ? resp.mutation_token()->sequence_number() : 0);
+ const uint16_t partition_id = (resp.mutation_token().has_value() ? resp.mutation_token()->partition_id() : 0);
+ return CouchbaseUpsertResult {
+ collection_.bucket_name(),
+ resp.cas().value(),
+ partition_uuid,
+ sequence_number,
+ partition_id
+ };
+ }
+}
+
+std::unique_ptr<CouchbaseCollection> CouchBaseClient::getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name) {
+ if (!establishConnection()) {
+ return nullptr;
+ }
+ return std::make_unique<RemoteCouchbaseCollection>(cluster_.bucket(bucket_name).scope(scope_name).collection(collection_name), *this);
+}
+
+void CouchBaseClient::setConnectionError() {
+ std::lock_guard<std::mutex> lock(state_mutex_);
+ state_ = State::UNKNOWN;
+}
+
+void CouchBaseClient::close() {
+ std::lock_guard<std::mutex> lock(state_mutex_);
+ if (state_ == State::CONNECTED || state_ == State::UNKNOWN) {
+ cluster_.close().wait();
+ state_ = State::DISCONNECTED;
+ }
+}
+
+bool CouchBaseClient::establishConnection() {
+ std::lock_guard<std::mutex> lock(state_mutex_);
+ if (state_ == State::CONNECTED) {
+ return true;
+ }
+
+ if (state_ == State::UNKNOWN) {
+ auto [err, resp] = cluster_.ping().get();
+ if (err.ec()) {
+ close();
+ } else {
+ state_ = State::CONNECTED;
+ return true;
+ }
+ }
+
+ auto options = ::couchbase::cluster_options(username_, password_);
+ auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, options).get();
+ if (connect_err.ec()) {
+ logger_->log_error("Failed to connect to Couchbase cluster: {}", connect_err.message());
+ return false;
+ }
+ cluster_ = std::move(cluster);
+ state_ = State::CONNECTED;
+ return true;
+}
+
+namespace controllers {
+
+void CouchbaseClusterService::initialize() {
+ setSupportedProperties(Properties);
+}
+
+void CouchbaseClusterService::onEnable() {
+ std::string connection_string;
+ getProperty(ConnectionString, connection_string);
+ std::string username;
+ getProperty(UserName, username);
+ std::string password;
+ getProperty(UserPassword, password);
+ client_ = std::make_unique<CouchBaseClient>(connection_string, username, password, logger_);
+}
+
+gsl::not_null<std::shared_ptr<CouchbaseClusterService>> CouchbaseClusterService::getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property) {
+ std::shared_ptr<CouchbaseClusterService> couchbase_cluster_service;
+ if (auto connection_controller_name = context.getProperty(property)) {
+ couchbase_cluster_service = std::dynamic_pointer_cast<CouchbaseClusterService>(context.getControllerService(*connection_controller_name));
+ }
+ if (!couchbase_cluster_service) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing Couchbase Cluster Service");
+ }
+ return gsl::make_not_null(couchbase_cluster_service);
+}
+
+REGISTER_RESOURCE(CouchbaseClusterService, ControllerService);
+
+} // namespace controllers
+} // namespace org::apache::nifi::minifi::couchbase
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
new file mode 100644
index 000000000..0fc3c0cc3
--- /dev/null
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -0,0 +1,150 @@
+/**
+ *
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "core/controller/ControllerService.h"
+#include "core/PropertyDefinition.h"
+#include "core/PropertyDefinitionBuilder.h"
+#include "core/PropertyType.h"
+#include "couchbase/cluster.hxx"
+#include "core/ProcessContext.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "CouchbaseCollection.h"
+#include "couchbase/codec/raw_binary_transcoder.hxx"
+
+namespace org::apache::nifi::minifi::couchbase {
+
+class CouchBaseClient;
+
+class RemoteCouchbaseCollection : public CouchbaseCollection {
+ public:
+ explicit RemoteCouchbaseCollection(::couchbase::collection collection, CouchBaseClient& client)
+ : collection_(std::move(collection)),
+ client_(client) {
+ }
+
+ nonstd::expected<CouchbaseUpsertResult, std::error_code> upsert(const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) override;
+
+ private:
+ ::couchbase::collection collection_;
+ CouchBaseClient& client_;
+};
+
+class CouchBaseClient {
+ public:
+ enum class State {
+ DISCONNECTED,
+ CONNECTED,
+ UNKNOWN,
+ };
+
+ CouchBaseClient(std::string connection_string, std::string username, std::string password, const std::shared_ptr<core::logging::Logger>& logger)
+ : connection_string_(std::move(connection_string)), username_(std::move(username)), password_(std::move(password)), logger_(logger) {
+ }
+
+ std::unique_ptr<CouchbaseCollection> getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name);
+ void setConnectionError();
+ void close();
+
+ private:
+ bool establishConnection();
+
+ std::mutex state_mutex_;
+ State state_ = State::DISCONNECTED;
+ std::string connection_string_;
+ std::string username_;
+ std::string password_;
+ ::couchbase::cluster cluster_;
+ std::shared_ptr<core::logging::Logger> logger_;
+};
+
+namespace controllers {
+
+class CouchbaseClusterService : public core::controller::ControllerService {
+ public:
+ explicit CouchbaseClusterService(std::string_view name, const minifi::utils::Identifier &uuid = {})
+ : ControllerService(name, uuid) {
+ }
+
+ explicit CouchbaseClusterService(std::string_view name, const std::shared_ptr<Configure>& /*configuration*/)
+ : ControllerService(name) {
+ }
+
+ EXTENSIONAPI static constexpr const char* Description = "Provides a centralized Couchbase connection and bucket passwords management. Bucket passwords can be specified via dynamic properties.";
+
+ EXTENSIONAPI static constexpr auto ConnectionString = core::PropertyDefinitionBuilder<>::createProperty("Connection String")
+ .withDescription("The hostnames or ip addresses of the bootstraping nodes and optional parameters. Syntax) couchbase://node1,node2,nodeN?param1=value1¶m2=value2¶mN=valueN")
+ .isRequired(true)
+ .build();
+ EXTENSIONAPI static constexpr auto UserName = core::PropertyDefinitionBuilder<>::createProperty("User Name")
+ .withDescription("The user name to authenticate MiNiFi as a Couchbase client.")
+ .build();
+ EXTENSIONAPI static constexpr auto UserPassword = core::PropertyDefinitionBuilder<>::createProperty("User Password")
+ .withDescription("The user password to authenticate MiNiFi as a Couchbase client.")
+ .isSensitive(true)
+ .build();
+
+ EXTENSIONAPI static constexpr auto Properties = std::to_array<core::PropertyReference>({
+ ConnectionString,
+ UserName,
+ UserPassword
+ });
+
+
+ EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
+ ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_CONTROLLER_SERVICES
+
+ void initialize() override;
+
+ void yield() override {
+ };
+
+ bool isWorkAvailable() override {
+ return false;
+ };
+
+ bool isRunning() const override {
+ return getState() == core::controller::ControllerServiceState::ENABLED;
+ }
+
+ void onEnable() override;
+ void notifyStop() override {
+ if (client_) {
+ client_->close();
+ }
+ }
+
+ virtual std::unique_ptr<CouchbaseCollection> getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name) {
+ gsl_Expects(client_);
+ return client_->getCollection(bucket_name, scope_name, collection_name);
+ }
+
+ static gsl::not_null<std::shared_ptr<CouchbaseClusterService>> getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property);
+
+ private:
+ std::unique_ptr<CouchBaseClient> client_;
+ std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CouchbaseClusterService>::getLogger(uuid_);
+};
+
+} // namespace controllers
+} // namespace org::apache::nifi::minifi::couchbase
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.cpp b/extensions/couchbase/processors/PutCouchbaseKey.cpp
new file mode 100644
index 000000000..4f4062d2d
--- /dev/null
+++ b/extensions/couchbase/processors/PutCouchbaseKey.cpp
@@ -0,0 +1,91 @@
+/**
+ *
+ * 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.
+ */
+
+#include "PutCouchbaseKey.h"
+#include "utils/gsl.h"
+#include "core/Resource.h"
+#include "utils/ProcessorConfigUtils.h"
+
+namespace org::apache::nifi::minifi::couchbase::processors {
+
+void PutCouchbaseKey::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
+ couchbase_cluster_service_ = controllers::CouchbaseClusterService::getFromProperty(context, PutCouchbaseKey::CouchbaseClusterControllerService);
+ persist_to_ = utils::parseEnumProperty<::couchbase::persist_to>(context, PutCouchbaseKey::PersistTo);
+ replicate_to_ = utils::parseEnumProperty<::couchbase::replicate_to>(context, PutCouchbaseKey::ReplicateTo);
+}
+
+void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
+ gsl_Expects(couchbase_cluster_service_);
+
+ auto flow_file = session.get();
+ if (!flow_file) {
+ context.yield();
+ return;
+ }
+
+ std::string bucket_name;
+ if (!context.getProperty(BucketName, bucket_name, flow_file.get()) || bucket_name.empty()) {
+ logger_->log_error("Bucket '{}' is invalid or empty!", bucket_name);
+ session.transfer(flow_file, Failure);
+ return;
+ }
+
+ std::string scope_name;
+ if (!context.getProperty(ScopeName, scope_name, flow_file.get()) || scope_name.empty()) {
+ scope_name = ::couchbase::scope::default_name;
+ }
+
+ std::string collection_name;
+ if (!context.getProperty(CollectionName, collection_name, flow_file.get()) || collection_name.empty()) {
+ collection_name = ::couchbase::collection::default_name;
+ }
+
+ std::string document_id;
+ if (!context.getProperty(DocumentId, document_id, flow_file.get()) || document_id.empty()) {
+ document_id = flow_file->getUUIDStr();
+ }
+
+ auto collection = couchbase_cluster_service_->getCollection(bucket_name, scope_name, collection_name);
+ if (!collection) {
+ logger_->log_error("Failed to get collection '{}.{}.{}', transferring to retry relationship", bucket_name, scope_name, collection_name);
+ session.transfer(flow_file, Retry);
+ return;
+ }
+ ::couchbase::upsert_options options;
+ options.durability(persist_to_, replicate_to_);
+ auto result = session.readBuffer(flow_file);
+ if (auto upsert_result = collection->upsert(document_id, result.buffer, options)) {
+ session.putAttribute(*flow_file, "couchbase.bucket", upsert_result->bucket_name);
+ session.putAttribute(*flow_file, "couchbase.doc.id", document_id);
+ session.putAttribute(*flow_file, "couchbase.doc.cas", std::to_string(upsert_result->cas));
+ session.putAttribute(*flow_file, "couchbase.doc.sequence.number", std::to_string(upsert_result->sequence_number));
+ session.putAttribute(*flow_file, "couchbase.partition.uuid", std::to_string(upsert_result->partition_uuid));
+ session.putAttribute(*flow_file, "couchbase.partition.id", std::to_string(upsert_result->partition_id));
+ session.transfer(flow_file, Success);
+ } else if (upsert_result.error().value() == static_cast<int>(::couchbase::errc::common::unambiguous_timeout)) {
+ logger_->log_error("Failed to upsert document '{}' due to timeout, transferring to retry relationship", document_id);
+ session.transfer(flow_file, Retry);
+ } else {
+ logger_->log_error("Failed to upsert document '{}': {}", document_id, upsert_result.error().message());
+ session.transfer(flow_file, Failure);
+ }
+}
+
+REGISTER_RESOURCE(PutCouchbaseKey, Processor);
+
+} // namespace org::apache::nifi::minifi::couchbase::processors
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.h b/extensions/couchbase/processors/PutCouchbaseKey.h
new file mode 100644
index 000000000..91975e62a
--- /dev/null
+++ b/extensions/couchbase/processors/PutCouchbaseKey.h
@@ -0,0 +1,152 @@
+/**
+ *
+ * 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.
+ */
+#pragma once
+
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "core/AbstractProcessor.h"
+#include "core/ProcessSession.h"
+#include "utils/Enum.h"
+#include "core/logging/LoggerConfiguration.h"
+#include "CouchbaseClusterService.h"
+#include "couchbase/persist_to.hxx"
+#include "couchbase/replicate_to.hxx"
+
+namespace magic_enum::customize {
+
+template <>
+constexpr customize_t enum_name<::couchbase::persist_to>(::couchbase::persist_to value) noexcept {
+ switch (value) {
+ case ::couchbase::persist_to::none:
+ return "NONE";
+ case ::couchbase::persist_to::active:
+ return "ACTIVE";
+ case ::couchbase::persist_to::one:
+ return "ONE";
+ case ::couchbase::persist_to::two:
+ return "TWO";
+ case ::couchbase::persist_to::three:
+ return "THREE";
+ case ::couchbase::persist_to::four:
+ return "FOUR";
+ }
+ return invalid_tag;
+}
+
+template <>
+constexpr customize_t enum_name<::couchbase::replicate_to>(::couchbase::replicate_to value) noexcept {
+ switch (value) {
+ case ::couchbase::replicate_to::none:
+ return "NONE";
+ case ::couchbase::replicate_to::one:
+ return "ONE";
+ case ::couchbase::replicate_to::two:
+ return "TWO";
+ case ::couchbase::replicate_to::three:
+ return "THREE";
+ }
+ return invalid_tag;
+}
+} // namespace magic_enum::customize
+
+namespace org::apache::nifi::minifi::couchbase::processors {
+
+class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
+ public:
+ using core::AbstractProcessor<PutCouchbaseKey>::AbstractProcessor;
+
+ EXTENSIONAPI static constexpr const char* Description = "Put a document to Couchbase Server via Key/Value access.";
+
+ EXTENSIONAPI static constexpr auto CouchbaseClusterControllerService = core::PropertyDefinitionBuilder<>::createProperty("Couchbase Cluster Controller Service")
+ .withDescription("A Couchbase Cluster Controller Service which manages connections to a Couchbase cluster.")
+ .withAllowedTypes<controllers::CouchbaseClusterService>()
+ .isRequired(true)
+ .build();
+ EXTENSIONAPI static constexpr auto BucketName = core::PropertyDefinitionBuilder<>::createProperty("Bucket Name")
+ .withDescription("The name of bucket to access.")
+ .withDefaultValue("default")
+ .isRequired(true)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto ScopeName = core::PropertyDefinitionBuilder<>::createProperty("Scope Name")
+ .withDescription("Scope to use inside the bucket. If not specified, the _default scope is used.")
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto CollectionName = core::PropertyDefinitionBuilder<>::createProperty("Collection Name")
+ .withDescription("Collection to use inside the bucket scope. If not specified, the _default collection is used.")
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto DocumentId = core::PropertyDefinitionBuilder<>::createProperty("Document Id")
+ .withDescription("A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.")
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto PersistTo = core::PropertyDefinitionBuilder<6>::createProperty("Persist To")
+ .withDescription("Durability constraint about disk persistence.")
+ .isRequired(true)
+ .withDefaultValue(magic_enum::enum_name(::couchbase::persist_to::none))
+ .withAllowedValues(magic_enum::enum_names<::couchbase::persist_to>())
+ .build();
+ EXTENSIONAPI static constexpr auto ReplicateTo = core::PropertyDefinitionBuilder<4>::createProperty("Replicate To")
+ .withDescription("Durability constraint about replication.")
+ .isRequired(true)
+ .withDefaultValue(magic_enum::enum_name(::couchbase::replicate_to::none))
+ .withAllowedValues(magic_enum::enum_names<::couchbase::replicate_to>())
+ .build();
+
+ EXTENSIONAPI static constexpr auto Properties = std::to_array<core::PropertyReference>({
+ CouchbaseClusterControllerService,
+ BucketName,
+ ScopeName,
+ CollectionName,
+ DocumentId,
+ PersistTo,
+ ReplicateTo
+ });
+
+ EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "All FlowFiles that are written to Couchbase Server are routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Failure = core::RelationshipDefinition{"failure", "All FlowFiles failed to be written to Couchbase Server and not retry-able are routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Retry = core::RelationshipDefinition{"retry", "All FlowFiles failed to be written to Couchbase Server but can be retried are routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Relationships = std::array{Success, Failure, Retry};
+
+ EXTENSIONAPI static constexpr auto CouchbaseBucket = core::OutputAttributeDefinition<>{"couchbase.bucket", {Success}, "Bucket where the document was stored."};
+ EXTENSIONAPI static constexpr auto CouchbaseDocId = core::OutputAttributeDefinition<>{"couchbase.doc.id", {Success}, "Id of the document."};
+ EXTENSIONAPI static constexpr auto CouchbaseDocCas = core::OutputAttributeDefinition<>{"couchbase.doc.cas", {Success}, "CAS of the document."};
+ EXTENSIONAPI static constexpr auto CouchbaseDocSequenceNumber = core::OutputAttributeDefinition<>{"couchbase.doc.sequence.number", {Success}, "Sequence number associated with the document."};
+ EXTENSIONAPI static constexpr auto CouchbasePartitionUUID = core::OutputAttributeDefinition<>{"couchbase.partition.uuid", {Success}, "UUID of partition."};
+ EXTENSIONAPI static constexpr auto CouchbasePartitionId = core::OutputAttributeDefinition<>{"couchbase.partition.id", {Success}, "ID of partition (also known as vBucket)."};
+ EXTENSIONAPI static constexpr auto OutputAttributes = std::array<core::OutputAttributeReference, 6>{
+ CouchbaseBucket, CouchbaseDocId, CouchbaseDocCas, CouchbaseDocSequenceNumber, CouchbasePartitionUUID, CouchbasePartitionId};
+
+ EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
+ EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
+ EXTENSIONAPI static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_REQUIRED;
+ EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
+
+ void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& sessionFactory) override;
+ void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override;
+
+ private:
+ std::shared_ptr<controllers::CouchbaseClusterService> couchbase_cluster_service_;
+ std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<PutCouchbaseKey>::getLogger(uuid_);
+ ::couchbase::persist_to persist_to_ = ::couchbase::persist_to::none;
+ ::couchbase::replicate_to replicate_to_ = ::couchbase::replicate_to::none;
+};
+
+} // namespace org::apache::nifi::minifi::couchbase::processors
diff --git a/extensions/couchbase/tests/CMakeLists.txt b/extensions/couchbase/tests/CMakeLists.txt
new file mode 100644
index 000000000..18caaeff0
--- /dev/null
+++ b/extensions/couchbase/tests/CMakeLists.txt
@@ -0,0 +1,39 @@
+#
+# 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.
+#
+
+file(GLOB COUCHBASE_TESTS "*.cpp")
+
+SET(COUCHBASE_TEST_COUNT 0)
+FOREACH(testfile ${COUCHBASE_TESTS})
+ get_filename_component(testfilename "${testfile}" NAME_WE)
+ add_minifi_executable("${testfilename}" "${testfile}")
+ target_include_directories(${testfilename} PRIVATE BEFORE "${CMAKE_SOURCE_DIR}/extensions/standard-processors")
+ target_include_directories(${testfilename} PRIVATE BEFORE "${CMAKE_SOURCE_DIR}/extensions/couchbase/controllerservices")
+ target_include_directories(${testfilename} PRIVATE BEFORE "${CMAKE_SOURCE_DIR}/extensions/couchbase/processors")
+ target_include_directories(${testfilename} PRIVATE BEFORE "${CMAKE_SOURCE_DIR}/extensions/couchbase/")
+ target_include_directories(${testfilename} PRIVATE BEFORE "${COUCHBASE_INCLUDE_DIR}")
+
+ createTests("${testfilename}")
+ target_link_libraries(${testfilename} Catch2WithMain)
+ target_link_libraries(${testfilename} minifi-couchbase)
+ target_link_libraries(${testfilename} minifi-standard-processors)
+ MATH(EXPR COUCHBASE_TEST_COUNT "${COUCHBASE_TEST_COUNT}+1")
+ add_test(NAME "${testfilename}" COMMAND "${testfilename}" WORKING_DIRECTORY ${TEST_DIR})
+ENDFOREACH()
+message("-- Finished building ${COUCHBASE_TEST_COUNT} Couchbase related test file(s)...")
diff --git a/extensions/couchbase/tests/MockCouchbaseClusterService.h b/extensions/couchbase/tests/MockCouchbaseClusterService.h
new file mode 100644
index 000000000..86e749dfe
--- /dev/null
+++ b/extensions/couchbase/tests/MockCouchbaseClusterService.h
@@ -0,0 +1,111 @@
+/**
+ * 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.
+ */
+
+#pragma once
+#include <utility>
+#include <string>
+#include "CouchbaseClusterService.h"
+#include "unit/Catch.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::couchbase::test {
+
+struct GetCollectionParameters {
+ std::string bucket_name;
+ std::string scope_name;
+ std::string collection_name;
+};
+
+struct UpsertParameters {
+ std::string document_id;
+ std::vector<std::byte> buffer;
+ ::couchbase::upsert_options options;
+};
+
+class MockCouchbaseCollection : public CouchbaseCollection {
+ public:
+ MockCouchbaseCollection(UpsertParameters& parameters, std::error_code& upsert_error_code, std::string_view bucket_name)
+ : parameters_(parameters),
+ upsert_error_code_(upsert_error_code),
+ bucket_name_(bucket_name) {}
+
+ nonstd::expected<CouchbaseUpsertResult, std::error_code> upsert(const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) override {
+ parameters_.document_id = document_id;
+ parameters_.buffer = buffer;
+ parameters_.options = options;
+
+ if (upsert_error_code_) {
+ return nonstd::make_unexpected(upsert_error_code_);
+ } else {
+ return CouchbaseUpsertResult{bucket_name_, 1, 2, 3, 4};
+ }
+ }
+
+ private:
+ UpsertParameters& parameters_;
+ std::error_code& upsert_error_code_;
+ std::string bucket_name_;
+};
+
+class MockCouchbaseClusterService : public controllers::CouchbaseClusterService {
+ public:
+ using CouchbaseClusterService::CouchbaseClusterService;
+ EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
+ ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_CONTROLLER_SERVICES
+
+ void onEnable() override {}
+ void notifyStop() override {}
+
+ std::unique_ptr<CouchbaseCollection> getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name) override {
+ get_collection_parameters_.bucket_name = bucket_name;
+ get_collection_parameters_.scope_name = scope_name;
+ get_collection_parameters_.collection_name = collection_name;
+ if (!get_collection_succeeds_) {
+ return nullptr;
+ } else {
+ return std::make_unique<MockCouchbaseCollection>(upsert_parameters_, upsert_error_code_, bucket_name);
+ }
+ }
+
+ GetCollectionParameters getGetCollectionParameters() const {
+ return get_collection_parameters_;
+ }
+
+ UpsertParameters getUpsertParameters() const {
+ return upsert_parameters_;
+ }
+
+ void setUpsertErrorCode(const std::error_code& error) {
+ upsert_error_code_ = error;
+ }
+
+ void setGetCollectionSucceeds(bool succeeds) {
+ get_collection_succeeds_ = succeeds;
+ }
+
+ bool getGetCollectionSucceeds() const {
+ return get_collection_succeeds_;
+ }
+
+ private:
+ GetCollectionParameters get_collection_parameters_;
+ UpsertParameters upsert_parameters_;
+ std::error_code upsert_error_code_;
+ bool get_collection_succeeds_{true};
+};
+} // namespace org::apache::nifi::minifi::couchbase::test
diff --git a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
new file mode 100644
index 000000000..e5953beb7
--- /dev/null
+++ b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
@@ -0,0 +1,169 @@
+/**
+ *
+ * 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.
+ */
+
+#include "unit/TestBase.h"
+#include "unit/Catch.h"
+#include "unit/SingleProcessorTestController.h"
+#include "processors/PutCouchbaseKey.h"
+#include "MockCouchbaseClusterService.h"
+#include "core/Resource.h"
+
+namespace org::apache::nifi::minifi::couchbase::test {
+
+REGISTER_RESOURCE(MockCouchbaseClusterService, ControllerService);
+
+struct ExpectedCallOptions {
+ std::string bucket_name;
+ std::string scope_name;
+ std::string collection_name;
+ ::couchbase::persist_to persist_to;
+ ::couchbase::replicate_to replicate_to;
+ std::string doc_id;
+};
+
+class PutCouchbaseKeyTestController : public TestController {
+ public:
+ PutCouchbaseKeyTestController() {
+ auto controller_service_node = controller_.plan->addController("MockCouchbaseClusterService", "MockCouchbaseClusterService");
+ mock_couchbase_cluster_service_ = std::static_pointer_cast<MockCouchbaseClusterService>(controller_service_node->getControllerServiceImplementation());
+ proc_->setProperty(processors::PutCouchbaseKey::CouchbaseClusterControllerService, "MockCouchbaseClusterService");
+ }
+
+ [[nodiscard]] static std::vector<std::byte> stringToByteVector(const std::string& str) {
+ std::vector<std::byte> byte_vector;
+ byte_vector.reserve(str.size());
+ for (char ch : str) {
+ byte_vector.push_back(static_cast<std::byte>(ch));
+ }
+ return byte_vector;
+ }
+
+ void verifyResults(const minifi::test::ProcessorTriggerResult& results, const minifi::core::Relationship& expected_result, const ExpectedCallOptions& expected_call_options,
+ const std::string& input) const {
+ std::shared_ptr<core::FlowFile> flow_file;
+ if (expected_result == processors::PutCouchbaseKey::Success) {
+ REQUIRE(results.at(processors::PutCouchbaseKey::Success).size() == 1);
+ REQUIRE(results.at(processors::PutCouchbaseKey::Failure).empty());
+ REQUIRE(results.at(processors::PutCouchbaseKey::Retry).empty());
+ flow_file = results.at(processors::PutCouchbaseKey::Success)[0];
+ } else if (expected_result == processors::PutCouchbaseKey::Failure) {
+ REQUIRE(results.at(processors::PutCouchbaseKey::Success).empty());
+ REQUIRE(results.at(processors::PutCouchbaseKey::Failure).size() == 1);
+ REQUIRE(results.at(processors::PutCouchbaseKey::Retry).empty());
+ flow_file = results.at(processors::PutCouchbaseKey::Failure)[0];
+ REQUIRE(LogTestController::getInstance().contains("Failed to upsert document", 1s));
+ } else {
+ REQUIRE(results.at(processors::PutCouchbaseKey::Success).empty());
+ REQUIRE(results.at(processors::PutCouchbaseKey::Failure).empty());
+ REQUIRE(results.at(processors::PutCouchbaseKey::Retry).size() == 1);
+ flow_file = results.at(processors::PutCouchbaseKey::Retry)[0];
+ }
+
+ auto get_collection_parameters = mock_couchbase_cluster_service_->getGetCollectionParameters();
+ CHECK(get_collection_parameters.bucket_name == expected_call_options.bucket_name);
+ CHECK(get_collection_parameters.collection_name == expected_call_options.collection_name);
+ CHECK(get_collection_parameters.scope_name == expected_call_options.scope_name);
+
+ if (!mock_couchbase_cluster_service_->getGetCollectionSucceeds()) {
+ return;
+ }
+
+ auto upsert_parameters = mock_couchbase_cluster_service_->getUpsertParameters();
+ std::string expected_doc_id = expected_call_options.doc_id.empty() ? flow_file->getUUID().to_string() : expected_call_options.doc_id;
+ CHECK(upsert_parameters.document_id == expected_doc_id);
+ CHECK(upsert_parameters.buffer == stringToByteVector(input));
+
+ auto upsert_options = upsert_parameters.options.build();
+ CHECK(upsert_options.persist_to == expected_call_options.persist_to);
+ CHECK(upsert_options.replicate_to == expected_call_options.replicate_to);
+
+ if (expected_result != processors::PutCouchbaseKey::Success) {
+ return;
+ }
+
+ CHECK(flow_file->getAttribute("couchbase.bucket").value() == expected_call_options.bucket_name);
+ CHECK(flow_file->getAttribute("couchbase.doc.id").value() == expected_doc_id);
+ CHECK(flow_file->getAttribute("couchbase.doc.cas").value() == "1");
+ CHECK(flow_file->getAttribute("couchbase.doc.sequence.number").value() == "2");
+ CHECK(flow_file->getAttribute("couchbase.partition.uuid").value() == "3");
+ CHECK(flow_file->getAttribute("couchbase.partition.id").value() == "4");
+ }
+
+ protected:
+ std::shared_ptr<core::Processor> proc_ = std::make_shared<processors::PutCouchbaseKey>("PutCouchbaseKey");
+ minifi::test::SingleProcessorTestController controller_{proc_};
+ std::shared_ptr<MockCouchbaseClusterService> mock_couchbase_cluster_service_;
+};
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Invalid Couchbase cluster controller service", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::CouchbaseClusterControllerService, "invalid");
+ REQUIRE_THROWS_AS(controller_.trigger({minifi::test::InputFlowFileData{"{\"name\": \"John\"}\n{\"name\": \"Jill\"}"}}), minifi::Exception);
+}
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Invalid bucket name", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::BucketName, "");
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{"{\"name\": \"John\"}\n{\"name\": \"Jill\"}"}});
+ REQUIRE(results[processors::PutCouchbaseKey::Failure].size() == 1);
+ REQUIRE(LogTestController::getInstance().contains("Bucket '' is invalid or empty!", 1s));
+}
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with default properties", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
+ const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+}
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with optional properties", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
+ proc_->setProperty(processors::PutCouchbaseKey::ScopeName, "scope1");
+ proc_->setProperty(processors::PutCouchbaseKey::CollectionName, "collection1");
+ proc_->setProperty(processors::PutCouchbaseKey::DocumentId, "important_doc");
+ proc_->setProperty(processors::PutCouchbaseKey::PersistTo, "ACTIVE");
+ proc_->setProperty(processors::PutCouchbaseKey::ReplicateTo, "TWO");
+ const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "scope1", "collection1", ::couchbase::persist_to::active,
+ ::couchbase::replicate_to::two, "important_doc"}, input);
+}
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put fails with default properties", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
+ mock_couchbase_cluster_service_->setUpsertErrorCode(std::make_error_code(std::errc::invalid_argument));
+ const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::PutCouchbaseKey::Failure, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+}
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when get collection fails", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
+ mock_couchbase_cluster_service_->setGetCollectionSucceeds(false);
+ const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+}
+
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when unambiguous timeout is returned", "[putcouchbasekey]") {
+ proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
+ mock_couchbase_cluster_service_->setUpsertErrorCode(::couchbase::errc::common::unambiguous_timeout);
+ const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+}
+
+} // namespace org::apache::nifi::minifi::couchbase::test
diff --git a/libminifi/include/Exception.h b/libminifi/include/Exception.h
index dff108d84..086303bc2 100644
--- a/libminifi/include/Exception.h
+++ b/libminifi/include/Exception.h
@@ -43,11 +43,12 @@ enum ExceptionType {
REGEX_EXCEPTION,
REPOSITORY_EXCEPTION,
PARAMETER_EXCEPTION,
+ CONTROLLER_ENABLE_EXCEPTION,
MAX_EXCEPTION
};
static const char *ExceptionStr[MAX_EXCEPTION] = { "File Operation", "Flow File Operation", "Processor Operation", "Process Session Operation", "Process Schedule Operation", "Site2Site Protocol",
- "General Operation", "Regex Operation", "Repository Operation", "Parameter Operation"};
+ "General Operation", "Regex Operation", "Repository Operation", "Parameter Operation", "Controller Enable Operation" };
inline const char *ExceptionTypeToString(ExceptionType type) {
if (type < MAX_EXCEPTION)
diff --git a/libminifi/include/utils/SmallString.h b/libminifi/include/utils/SmallString.h
index 2ff5dd9b6..edf9466d2 100644
--- a/libminifi/include/utils/SmallString.h
+++ b/libminifi/include/utils/SmallString.h
@@ -100,7 +100,7 @@ struct fmt::formatter<org::apache::nifi::minifi::utils::SmallString<N>> {
}
template <typename FormatContext>
- auto format(const org::apache::nifi::minifi::utils::SmallString<N>& small_string, FormatContext& ctx) {
+ auto format(const org::apache::nifi::minifi::utils::SmallString<N>& small_string, FormatContext& ctx) const {
return string_view_formatter.format(small_string.view(), ctx);
}
};
diff --git a/libminifi/test/unit/LoggerTests.cpp b/libminifi/test/unit/LoggerTests.cpp
index fc87ee500..9c9ec0d51 100644
--- a/libminifi/test/unit/LoggerTests.cpp
+++ b/libminifi/test/unit/LoggerTests.cpp
@@ -386,7 +386,7 @@ TEST_CASE("fmt formatting works with the logger") {
logger->log_critical("{} in hex is {:#x}", 13, 13);
logger->log_critical("Unix epoch: {}", std::chrono::system_clock::time_point());
logger->log_critical("{:%Q %q} equals to {:%Q %q}", 2h, std::chrono::duration_cast<std::chrono::seconds>(2h));
- CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 1m equals to 60s"));
+ CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 1min equals to 60s"));
CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 13 in hex is 0xd"));
CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] Unix epoch: 1970-01-01 00:00:00.00"));
CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 2 h equals to 7200 s"));
diff --git a/thirdparty/bustache/add-append.patch b/thirdparty/bustache/add-append.patch
new file mode 100644
index 000000000..fb7f24149
--- /dev/null
+++ b/thirdparty/bustache/add-append.patch
@@ -0,0 +1,39 @@
+diff --git a/include/bustache/model.hpp b/include/bustache/model.hpp
+index 575969a..6df2cff 100644
+--- a/include/bustache/model.hpp
++++ b/include/bustache/model.hpp
+@@ -8,7 +8,7 @@
+ #define BUSTACHE_MODEL_HPP_INCLUDED
+
+ #include <bustache/format.hpp>
+-#include <version>
++#include <version>
+ #include <vector>
+ #include <cstring>
+ #include <concepts>
+@@ -330,6 +330,16 @@ namespace bustache::detail
+ buf[count++] = c;
+ }
+
++ template <typename U>
++ void append(const U* begin, const U* end)
++ {
++ size_t content_count = end - begin;
++ for (size_t i = 0; i < content_count; ++i) {
++ push_back(begin[i]);
++ }
++ begin += content_count;
++ }
++
+ void flush() { os(buf, count); }
+
+ std::size_t count = 0;
+@@ -592,7 +602,7 @@ namespace bustache
+ {
+ static constexpr model kind = model::atom;
+ };
+-
++
+ template<String T>
+ struct impl_test<T>
+ {
diff --git a/thirdparty/couchbase/remove-thirdparty.patch b/thirdparty/couchbase/remove-thirdparty.patch
new file mode 100644
index 000000000..e57ca18f5
--- /dev/null
+++ b/thirdparty/couchbase/remove-thirdparty.patch
@@ -0,0 +1,241 @@
+diff --git a/cmake/ThirdPartyDependencies.cmake b/cmake/ThirdPartyDependencies.cmake
+index f02af02..1a51f82 100644
+--- a/cmake/ThirdPartyDependencies.cmake
++++ b/cmake/ThirdPartyDependencies.cmake
+@@ -15,48 +15,48 @@ function(declare_system_library target)
+ $<TARGET_PROPERTY:${target},INTERFACE_INCLUDE_DIRECTORIES>)
+ endfunction()
+
+-if(NOT TARGET fmt::fmt)
+- # https://github.com/fmtlib/fmt/releases
+- cpmaddpackage(
+- NAME
+- fmt
+- GIT_TAG
+- 11.0.1
+- VERSION
+- 11.0.1
+- GITHUB_REPOSITORY
+- "fmtlib/fmt"
+- EXCLUDE_FROM_ALL ON
+- OPTIONS
+- "FMT_INSTALL OFF"
+- # Unicode support for MSVC enabled in CompilerWarnings.cmake
+- "FMT_UNICODE OFF"
+- "FMT_DOC OFF"
+- "BUILD_SHARED_LIBS OFF"
+- "CMAKE_C_VISIBILITY_PRESET hidden"
+- "CMAKE_CXX_VISIBILITY_PRESET hidden"
+- "CMAKE_POSITION_INDEPENDENT_CODE ON")
+-endif()
+-
+-if(NOT TARGET spdlog::spdlog)
+- # https://github.com/gabime/spdlog/releases
+- cpmaddpackage(
+- NAME
+- spdlog
+- VERSION
+- 1.14.1
+- GITHUB_REPOSITORY
+- "gabime/spdlog"
+- EXCLUDE_FROM_ALL ON
+- OPTIONS
+- "SPDLOG_INSTALL OFF"
+- "BUILD_SHARED_LIBS OFF"
+- "CMAKE_C_VISIBILITY_PRESET hidden"
+- "CMAKE_CXX_VISIBILITY_PRESET hidden"
+- "CMAKE_POSITION_INDEPENDENT_CODE ON"
+- "SPDLOG_BUILD_SHARED OFF"
+- "SPDLOG_FMT_EXTERNAL ON")
+-endif()
++# if(NOT TARGET fmt::fmt)
++# # https://github.com/fmtlib/fmt/releases
++# cpmaddpackage(
++# NAME
++# fmt
++# GIT_TAG
++# 11.0.1
++# VERSION
++# 11.0.1
++# GITHUB_REPOSITORY
++# "fmtlib/fmt"
++# EXCLUDE_FROM_ALL ON
++# OPTIONS
++# "FMT_INSTALL OFF"
++# # Unicode support for MSVC enabled in CompilerWarnings.cmake
++# "FMT_UNICODE OFF"
++# "FMT_DOC OFF"
++# "BUILD_SHARED_LIBS OFF"
++# "CMAKE_C_VISIBILITY_PRESET hidden"
++# "CMAKE_CXX_VISIBILITY_PRESET hidden"
++# "CMAKE_POSITION_INDEPENDENT_CODE ON")
++# endif()
++
++# if(NOT TARGET spdlog::spdlog)
++# # https://github.com/gabime/spdlog/releases
++# cpmaddpackage(
++# NAME
++# spdlog
++# VERSION
++# 1.14.1
++# GITHUB_REPOSITORY
++# "gabime/spdlog"
++# EXCLUDE_FROM_ALL ON
++# OPTIONS
++# "SPDLOG_INSTALL OFF"
++# "BUILD_SHARED_LIBS OFF"
++# "CMAKE_C_VISIBILITY_PRESET hidden"
++# "CMAKE_CXX_VISIBILITY_PRESET hidden"
++# "CMAKE_POSITION_INDEPENDENT_CODE ON"
++# "SPDLOG_BUILD_SHARED OFF"
++# "SPDLOG_FMT_EXTERNAL ON")
++# endif()
+
+ if(NOT TARGET Microsoft.GSL::GSL)
+ # https://github.com/microsoft/GSL/releases
+@@ -172,70 +172,70 @@ if(NOT TARGET taocpp::json)
+ endif()
+
+
+-if(NOT TARGET asio::asio)
+- # https://github.com/chriskohlhoff/asio/tags
+- cpmaddpackage(
+- NAME
+- asio
+- GIT_TAG
+- asio-1-31-0
+- VERSION
+- 1.31.0
+- GITHUB_REPOSITORY
+- "chriskohlhoff/asio"
+- EXCLUDE_FROM_ALL ON)
+-endif()
+-
+-# ASIO doesn't use CMake, we have to configure it manually. Extra notes for using on Windows:
+-#
+-# 1) If _WIN32_WINNT is not set, ASIO assumes _WIN32_WINNT=0x0501, i.e. Windows XP target, which is definitely not the
+-# platform which most users target.
+-#
+-# 2) WIN32_LEAN_AND_MEAN is defined to make Winsock2 work.
+-if(asio_ADDED)
+- add_library(asio STATIC ${asio_SOURCE_DIR}/asio/src/asio.cpp ${asio_SOURCE_DIR}/asio/src/asio_ssl.cpp)
+-
+- target_include_directories(asio SYSTEM PUBLIC ${asio_SOURCE_DIR}/asio/include)
+- target_compile_definitions(asio PRIVATE ASIO_STANDALONE=1 ASIO_NO_DEPRECATED=1 ASIO_SEPARATE_COMPILATION=1)
+- target_link_libraries(asio PRIVATE Threads::Threads OpenSSL::SSL OpenSSL::Crypto)
+- set_target_properties(
+- asio
+- PROPERTIES C_VISIBILITY_PRESET hidden
+- CXX_VISIBILITY_PRESET hidden
+- POSITION_INDEPENDENT_CODE TRUE)
+-
+- if(WIN32)
+- # macro see @ https://stackoverflow.com/a/40217291/1746503
+- macro(get_win32_winnt version)
+- if(CMAKE_SYSTEM_VERSION)
+- set(ver ${CMAKE_SYSTEM_VERSION})
+- string(REGEX MATCH "^([0-9]+).([0-9])" ver ${ver})
+- string(REGEX MATCH "^([0-9]+)" verMajor ${ver})
+- # Check for Windows 10, b/c we'll need to convert to hex 'A'.
+- if("${verMajor}" MATCHES "10")
+- set(verMajor "A")
+- string(REGEX REPLACE "^([0-9]+)" ${verMajor} ver ${ver})
+- endif("${verMajor}" MATCHES "10")
+- # Remove all remaining '.' characters.
+- string(REPLACE "." "" ver ${ver})
+- # Prepend each digit with a zero.
+- string(REGEX REPLACE "([0-9A-Z])" "0\\1" ver ${ver})
+- set(${version} "0x${ver}")
+- endif()
+- endmacro()
+-
+- if(NOT DEFINED _WIN32_WINNT)
+- get_win32_winnt(ver)
+- set(_WIN32_WINNT ${ver})
+- endif()
+-
+- message(STATUS "Set _WIN32_WINNT=${_WIN32_WINNT}")
+-
+- target_compile_definitions(asio INTERFACE _WIN32_WINNT=${_WIN32_WINNT} WIN32_LEAN_AND_MEAN)
+- endif()
+-
+- add_library(asio::asio ALIAS asio)
+-endif()
++# if(NOT TARGET asio::asio)
++# # https://github.com/chriskohlhoff/asio/tags
++# cpmaddpackage(
++# NAME
++# asio
++# GIT_TAG
++# asio-1-31-0
++# VERSION
++# 1.31.0
++# GITHUB_REPOSITORY
++# "chriskohlhoff/asio"
++# EXCLUDE_FROM_ALL ON)
++# endif()
++
++# # ASIO doesn't use CMake, we have to configure it manually. Extra notes for using on Windows:
++# #
++# # 1) If _WIN32_WINNT is not set, ASIO assumes _WIN32_WINNT=0x0501, i.e. Windows XP target, which is definitely not the
++# # platform which most users target.
++# #
++# # 2) WIN32_LEAN_AND_MEAN is defined to make Winsock2 work.
++# if(asio_ADDED)
++# add_library(asio STATIC ${asio_SOURCE_DIR}/asio/src/asio.cpp ${asio_SOURCE_DIR}/asio/src/asio_ssl.cpp)
++
++# target_include_directories(asio SYSTEM PUBLIC ${asio_SOURCE_DIR}/asio/include)
++# target_compile_definitions(asio PRIVATE ASIO_STANDALONE=1 ASIO_NO_DEPRECATED=1 ASIO_SEPARATE_COMPILATION=1)
++# target_link_libraries(asio PRIVATE Threads::Threads OpenSSL::SSL OpenSSL::Crypto)
++# set_target_properties(
++# asio
++# PROPERTIES C_VISIBILITY_PRESET hidden
++# CXX_VISIBILITY_PRESET hidden
++# POSITION_INDEPENDENT_CODE TRUE)
++
++# if(WIN32)
++# # macro see @ https://stackoverflow.com/a/40217291/1746503
++# macro(get_win32_winnt version)
++# if(CMAKE_SYSTEM_VERSION)
++# set(ver ${CMAKE_SYSTEM_VERSION})
++# string(REGEX MATCH "^([0-9]+).([0-9])" ver ${ver})
++# string(REGEX MATCH "^([0-9]+)" verMajor ${ver})
++# # Check for Windows 10, b/c we'll need to convert to hex 'A'.
++# if("${verMajor}" MATCHES "10")
++# set(verMajor "A")
++# string(REGEX REPLACE "^([0-9]+)" ${verMajor} ver ${ver})
++# endif("${verMajor}" MATCHES "10")
++# # Remove all remaining '.' characters.
++# string(REPLACE "." "" ver ${ver})
++# # Prepend each digit with a zero.
++# string(REGEX REPLACE "([0-9A-Z])" "0\\1" ver ${ver})
++# set(${version} "0x${ver}")
++# endif()
++# endmacro()
++
++# if(NOT DEFINED _WIN32_WINNT)
++# get_win32_winnt(ver)
++# set(_WIN32_WINNT ${ver})
++# endif()
++
++# message(STATUS "Set _WIN32_WINNT=${_WIN32_WINNT}")
++
++# target_compile_definitions(asio INTERFACE _WIN32_WINNT=${_WIN32_WINNT} WIN32_LEAN_AND_MEAN)
++# endif()
++
++# add_library(asio::asio ALIAS asio)
++# endif()
+
+ add_library(jsonsl OBJECT ${PROJECT_SOURCE_DIR}/third_party/jsonsl/jsonsl.c)
+ set_target_properties(jsonsl PROPERTIES C_VISIBILITY_PRESET hidden POSITION_INDEPENDENT_CODE TRUE)
+@@ -245,7 +245,7 @@ declare_system_library(snappy)
+ declare_system_library(llhttp::llhttp)
+ declare_system_library(hdr_histogram_static)
+ declare_system_library(Microsoft.GSL::GSL)
+-declare_system_library(spdlog::spdlog)
+-declare_system_library(fmt::fmt)
+-declare_system_library(asio)
++# declare_system_library(spdlog::spdlog)
++# declare_system_library(fmt::fmt)
++# declare_system_library(asio)
+ declare_system_library(taocpp::json)
diff --git a/win_build_vs.bat b/win_build_vs.bat
index a0058a6d7..f0699b885 100644
--- a/win_build_vs.bat
+++ b/win_build_vs.bat
@@ -45,6 +45,7 @@ set enable_prometheus=ON
set enable_gcp=ON
set enable_elastic=ON
set enable_grafana_loki=OFF
+set enable_couchbase=OFF
set test_custom_wel_provider=OFF
set generator="Visual Studio 17 2022"
set cpack=OFF
@@ -83,6 +84,7 @@ for %%x in (%*) do (
if [%%~x] EQU [/NO_OPS] set enable_ops=OFF
if [%%~x] EQU [/NO_PYTHON_SCRIPTING] set enable_python_scripting=OFF
if [%%~x] EQU [/LOKI] set enable_grafana_loki=ON
+ if [%%~x] EQU [/COUCHBASE] set enable_couchbase=ON
if [%%~x] EQU [/32] set build_platform=Win32
if [%%~x] EQU [/D] set cmake_build_type=RelWithDebInfo
if [%%~x] EQU [/DD] set cmake_build_type=Debug
@@ -111,7 +113,7 @@ cmake -G %generator% %build_platform_cmd% -DMINIFI_INCLUDE_VC_REDIST_MERGE_MODUL
-DENABLE_OPENCV=%enable_opencv% -DENABLE_PROMETHEUS=%enable_prometheus% -DENABLE_ELASTICSEARCH=%enable_elastic% -DUSE_SHARED_LIBS=OFF -DENABLE_CONTROLLER=OFF ^
-DENABLE_BUSTACHE=%enable_bustache% -DENABLE_ENCRYPT_CONFIG=%enable_encrypt_config% -DENABLE_LUA_SCRIPTING=%enable_lua_scripting% -DENABLE_SMB=%enable_smb% ^
-DENABLE_MQTT=%enable_mqtt% -DENABLE_OPC=%enable_opc% -DENABLE_OPS=%enable_ops% ^
- -DENABLE_PYTHON_SCRIPTING=%enable_python_scripting% -DENABLE_GRAFANA_LOKI=%enable_grafana_loki% ^
+ -DENABLE_PYTHON_SCRIPTING=%enable_python_scripting% -DENABLE_GRAFANA_LOKI=%enable_grafana_loki% -DENABLE_COUCHBASE=%enable_couchbase% ^
-DBUILD_ROCKSDB=ON -DUSE_SYSTEM_UUID=OFF -DENABLE_LIBARCHIVE=ON -DENABLE_WEL=ON -DMINIFI_FAIL_ON_WARNINGS=OFF -DSKIP_TESTS=%skiptests% -DMINIFI_INCLUDE_VC_REDIST_DLLS=%vc_redist% ^
%strict_gsl_checks% -DMINIFI_INCLUDE_UCRT_DLLS=%ucrt% %sccache_arg% %EXTRA_CMAKE_ARGUMENTS% "%scriptdir%" && %buildcmd%
IF %ERRORLEVEL% NEQ 0 EXIT /b %ERRORLEVEL%
(nifi-minifi-cpp) 10/15: Update extensions in README.md
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit e1086ceaecc376f9df6a28556ddd30fe74d2ae36
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 21 14:12:58 2024 +0200
Update extensions in README.md
---
README.md | 1 +
1 file changed, 1 insertion(+)
diff --git a/README.md b/README.md
index 8285e4b4d..069d142ad 100644
--- a/README.md
+++ b/README.md
@@ -75,6 +75,7 @@ The next table outlines CMAKE flags that correspond with MiNiFi extensions. Exte
| AWS | [AWSCredentialsService](CONTROLLERS.md#awscredentialsservice)<br/>[PutS3Object](PROCESSORS.md#puts3object)<br/>[DeleteS3Object](PROCESSORS.md#deletes3object)<br/>[FetchS3Object](PROCESSORS.md#fetchs3object)<br/>[ListS3](PROCESSORS.md#lists3) [...]
| Azure | [AzureStorageCredentialsService](CONTROLLERS.md#azurestoragecredentialsservice)<br/>[PutAzureBlobStorage](PROCESSORS.md#putazureblobstorage)<br/>[DeleteAzureBlobStorage](PROCESSORS.md#deleteazureblobstorage)<br/>[FetchAzureBlobStorage](PROCESSORS.md#fetchazureblobstorage)<br/>[ListAzureBlobStorage](PROCESSORS.md#listazureblobstorage)<br/>[PutAzureDataLakeStorage](PROCESSORS.md#putazuredatalakestorage)<br/>[DeleteAzureDataLakeStorage](PROCESSORS.md#del [...]
| CivetWeb | [ListenHTTP](PROCESSORS.md#listenhttp) [...]
+| Couchbase | [CouchbaseClusterService](CONTROLLERS.md#couchbaseclusterservice)<br/>[PutCouchbaseKey](PROCESSORS.md#putcouchbasekey) [...]
| Elasticsearch | [ElasticsearchCredentialsControllerService](CONTROLLERS.md#elasticsearchcredentialscontrollerservice)<br/>[PostElasticsearch](PROCESSORS.md#postelasticsearch) [...]
| ExecuteProcess (Linux and macOS) | [ExecuteProcess](PROCESSORS.md#executeprocess) [...]
| Google Cloud Platform | [DeleteGCSObject](PROCESSORS.md#deletegcsobject)<br>[FetchGCSObject](PROCESSORS.md#fetchgcsobject)<br>[GCPCredentialsControllerService](CONTROLLERS.md#gcpcredentialscontrollerservice)<br>[ListGCSBucket](PROCESSORS.md#listgcsbucket)<br>[PutGCSObject](PROCESSORS.md#putgcsobject) [...]
(nifi-minifi-cpp) 09/15: Refactor tests
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 57d1c6403eb5bc877add814cbbe7a8a82af5c5af
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 21 13:56:36 2024 +0200
Refactor tests
---
docker/test/integration/features/couchbase.feature | 12 ++++++------
docker/test/integration/features/steps/steps.py | 14 ++++----------
2 files changed, 10 insertions(+), 16 deletions(-)
diff --git a/docker/test/integration/features/couchbase.feature b/docker/test/integration/features/couchbase.feature
index f4e352cbe..446821f25 100644
--- a/docker/test/integration/features/couchbase.feature
+++ b/docker/test/integration/features/couchbase.feature
@@ -19,14 +19,14 @@ Feature: Executing Couchbase operations from MiNiFi-C++
Given the content of "/tmp/output" is monitored
Scenario: A MiNiFi instance can insert json data to test bucket with PutCouchbaseKey processor
- Given a Couchbase server is set up
- And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the PutCouchbaseKey processor is set to "Json"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
- And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up with the name "CouchbaseClusterService"
And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
And the "success" relationship of the PutCouchbaseKey processor is connected to the LogAttribute
@@ -43,14 +43,14 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1", "field2": "value2"}' of type "Json" in Couchbase
Scenario: A MiNiFi instance can insert binary data to test bucket with PutCouchbaseKey processor
- Given a Couchbase server is set up
- And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
And the "Document Type" property of the PutCouchbaseKey processor is set to "Binary"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
- And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up with the name "CouchbaseClusterService"
And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
And the "success" relationship of the PutCouchbaseKey processor is connected to the LogAttribute
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 0a76672c3..a6909926e 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -1359,24 +1359,18 @@ def step_impl(context, modbus_cmd):
# Couchbase
-@given("a Couchbase server is set up")
-def step_impl(context):
- context.test.acquire_container(context=context, name="couchbase-server", engine="couchbase-server")
-
-
@when(u'a Couchbase server is started')
def step_impl(context):
context.test.start_couchbase_server(context)
-@given("a CouchbaseClusterService is setup up for {processor_name} with the name \"{service_name}\"")
-def step_impl(context, processor_name, service_name):
+@given("a CouchbaseClusterService is setup up with the name \"{service_name}\"")
+def step_impl(context, service_name):
couchbase_cluster_controller_service = CouchbaseClusterService(
name=service_name,
connection_string="couchbase://{server_hostname}".format(server_hostname=context.test.get_container_name_with_postfix("couchbase-server")))
- processor = context.test.get_node_by_name(processor_name)
- processor.controller_services.append(couchbase_cluster_controller_service)
- processor.set_property("Couchbase Cluster Controller Service", couchbase_cluster_controller_service.name)
+ container = context.test.acquire_container(context=context, name="minifi-cpp-flow")
+ container.add_controller(couchbase_cluster_controller_service)
@then("a document with id \"{doc_id}\" in bucket \"{bucket_name}\" is present with data '{data}' of type \"{data_type}\" in Couchbase")
(nifi-minifi-cpp) 08/15: Review update
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit abc4c2d2684f5b25f2ac0952cc9e9a2032da30f1
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 14 20:42:47 2024 +0200
Review update
---
.../controllerservices/CouchbaseClusterService.cpp | 18 ++++++++++--------
.../controllerservices/CouchbaseClusterService.h | 2 +-
2 files changed, 11 insertions(+), 9 deletions(-)
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index d40ada072..8abf7985c 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -35,8 +35,9 @@ CouchbaseErrorType CouchbaseClient::getErrorType(const std::error_code& error_co
}
nonstd::expected<::couchbase::collection, CouchbaseErrorType> CouchbaseClient::getCollection(const CouchbaseCollection& collection) {
- if (auto error_type = establishConnection()) {
- return nonstd::make_unexpected(*error_type);
+ auto connection_result = establishConnection();
+ if (!connection_result) {
+ return nonstd::make_unexpected(connection_result.error());
}
return cluster_->bucket(collection.bucket_name).scope(collection.scope_name).collection(collection.collection_name);
}
@@ -88,19 +89,19 @@ void CouchbaseClient::close() {
}
}
-std::optional<CouchbaseErrorType> CouchbaseClient::establishConnection() {
+nonstd::expected<void, CouchbaseErrorType> CouchbaseClient::establishConnection() {
if (cluster_) {
- return std::nullopt;
+ return {};
}
auto options = ::couchbase::cluster_options(username_, password_);
auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, options).get();
if (connect_err.ec()) {
logger_->log_error("Failed to connect to Couchbase cluster with error code: '{}' and message: '{}'", connect_err.ec(), connect_err.message());
- return getErrorType(connect_err.ec());
+ return nonstd::make_unexpected(getErrorType(connect_err.ec()));
}
cluster_ = std::move(cluster);
- return std::nullopt;
+ return {};
}
namespace controllers {
@@ -121,8 +122,9 @@ void CouchbaseClusterService::onEnable() {
}
client_ = std::make_unique<CouchbaseClient>(connection_string, username, password, logger_);
- if (auto result = client_->establishConnection()) {
- if (result == CouchbaseErrorType::FATAL) {
+ auto result = client_->establishConnection();
+ if (!result) {
+ if (result.error() == CouchbaseErrorType::FATAL) {
throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Failed to connect to Couchbase cluster with fatal error");
}
logger_->log_warn("Failed to connect to Couchbase cluster with temporary error, will retry connection when a Couchbase processor is triggered");
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 0d46975b0..0103404c2 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -65,7 +65,7 @@ class CouchbaseClient {
nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type, const std::string& document_id,
const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options);
- std::optional<CouchbaseErrorType> establishConnection();
+ nonstd::expected<void, CouchbaseErrorType> establishConnection();
void close();
private:
(nifi-minifi-cpp) 12/15: Fix test on Windows
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 660fde2ce5ff9d10c47967f312b80b899ec2e05f
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Oct 22 11:02:14 2024 +0200
Fix test on Windows
---
PROCESSORS.md | 20 ++++++++++----------
extensions/couchbase/processors/PutCouchbaseKey.cpp | 2 +-
extensions/couchbase/processors/PutCouchbaseKey.h | 3 ++-
extensions/couchbase/tests/PutCouchbaseKeyTests.cpp | 21 ++++++++++++++-------
4 files changed, 27 insertions(+), 19 deletions(-)
diff --git a/PROCESSORS.md b/PROCESSORS.md
index fbccbb9c4..274393223 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -2213,16 +2213,16 @@ Put a document to Couchbase Server via Key/Value access.
In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
-| Name | Default Value | Allowable Values | Description |
-|------------------------------------------|---------------|----------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| **Couchbase Cluster Controller Service** | | | A Couchbase Cluster Controller Service which manages connections to a Couchbase cluster. |
-| **Bucket Name** | default | | The name of bucket to access.<br/>**Supports Expression Language: true** |
-| Scope Name | | | Scope to use inside the bucket. If not specified, the _default scope is used.<br/>**Supports Expression Language: true** |
-| Collection Name | | | Collection to use inside the bucket scope. If not specified, the _default collection is used.<br/>**Supports Expression Language: true** |
-| **Document Type** | Json | Json<br/>Binary<br/>String | Content type to store data as. |
-| Document Id | | | A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.<br/>**Supports Expression Language: true** |
-| **Persist To** | NONE | NONE<br/>ACTIVE<br/>ONE<br/>TWO<br/>THREE<br/>FOUR | Durability constraint about disk persistence. |
-| **Replicate To** | NONE | NONE<br/>ONE<br/>TWO<br/>THREE | Durability constraint about replication. |
+| Name | Default Value | Allowable Values | Description |
+|------------------------------------------|---------------|----------------------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Couchbase Cluster Controller Service** | | | A Couchbase Cluster Controller Service which manages connections to a Couchbase cluster. |
+| **Bucket Name** | default | | The name of bucket to access.<br/>**Supports Expression Language: true** |
+| Scope Name | | | Scope to use inside the bucket. If not specified, the _default scope is used.<br/>**Supports Expression Language: true** |
+| Collection Name | | | Collection to use inside the bucket scope. If not specified, the _default collection is used.<br/>**Supports Expression Language: true** |
+| **Document Type** | Json | Json<br/>Binary<br/>String | Content type to store data as. |
+| Document Id | | | A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, either the FlowFile uuid attribute or if that's not found a generated uuid will be used.<br/>**Supports Expression Language: true** |
+| **Persist To** | NONE | NONE<br/>ACTIVE<br/>ONE<br/>TWO<br/>THREE<br/>FOUR | Durability constraint about disk persistence. |
+| **Replicate To** | NONE | NONE<br/>ONE<br/>TWO<br/>THREE | Durability constraint about replication. |
### Relationships
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.cpp b/extensions/couchbase/processors/PutCouchbaseKey.cpp
index 77708debb..6f8ed54ce 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.cpp
+++ b/extensions/couchbase/processors/PutCouchbaseKey.cpp
@@ -56,7 +56,7 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
std::string document_id;
if (!context.getProperty(DocumentId, document_id, flow_file.get()) || document_id.empty()) {
- document_id = flow_file->getUUIDStr();
+ document_id = flow_file->getAttribute(core::SpecialFlowAttribute::UUID).value_or(utils::IdGenerator::getIdGenerator()->generate().to_string());
}
::couchbase::upsert_options options;
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.h b/extensions/couchbase/processors/PutCouchbaseKey.h
index ca79b7dba..56ba8c52f 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.h
+++ b/extensions/couchbase/processors/PutCouchbaseKey.h
@@ -100,7 +100,8 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
.withAllowedValues(magic_enum::enum_names<CouchbaseValueType>())
.build();
EXTENSIONAPI static constexpr auto DocumentId = core::PropertyDefinitionBuilder<>::createProperty("Document Id")
- .withDescription("A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.")
+ .withDescription("A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. "
+ "If not specified, either the FlowFile uuid attribute or if that's not found a generated uuid will be used.")
.supportsExpressionLanguage(true)
.build();
EXTENSIONAPI static constexpr auto PersistTo = core::PropertyDefinitionBuilder<6>::createProperty("Persist To")
diff --git a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
index 7adb02394..2ffe683a8 100644
--- a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
+++ b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
@@ -27,6 +27,8 @@ namespace org::apache::nifi::minifi::couchbase::test {
REGISTER_RESOURCE(MockCouchbaseClusterService, ControllerService);
+const std::string TEST_UUID = "a53f0e78-b91a-4a82-939b-639174edb00b";
+
struct ExpectedCallOptions {
std::string bucket_name;
std::string scope_name;
@@ -40,6 +42,11 @@ struct ExpectedCallOptions {
class PutCouchbaseKeyTestController : public TestController {
public:
PutCouchbaseKeyTestController() {
+ LogTestController::getInstance().setDebug<TestPlan>();
+ LogTestController::getInstance().setDebug<minifi::core::Processor>();
+ LogTestController::getInstance().setTrace<minifi::core::ProcessSession>();
+ LogTestController::getInstance().setDebug<controllers::CouchbaseClusterService>();
+ LogTestController::getInstance().setDebug<processors::PutCouchbaseKey>();
auto controller_service_node = controller_.plan->addController("MockCouchbaseClusterService", "MockCouchbaseClusterService");
mock_couchbase_cluster_service_ = std::static_pointer_cast<MockCouchbaseClusterService>(controller_service_node->getControllerServiceImplementation());
proc_->setProperty(processors::PutCouchbaseKey::CouchbaseClusterControllerService, "MockCouchbaseClusterService");
@@ -82,7 +89,7 @@ class PutCouchbaseKeyTestController : public TestController {
auto upsert_parameters = mock_couchbase_cluster_service_->getUpsertParameters();
CHECK(upsert_parameters.document_type == expected_call_options.document_type);
- std::string expected_doc_id = expected_call_options.doc_id.empty() ? flow_file->getUUID().to_string() : expected_call_options.doc_id;
+ auto expected_doc_id = expected_call_options.doc_id.empty() ? TEST_UUID : expected_call_options.doc_id;
CHECK(upsert_parameters.document_id == expected_doc_id);
CHECK(upsert_parameters.buffer == stringToByteVector(input));
@@ -110,12 +117,12 @@ class PutCouchbaseKeyTestController : public TestController {
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Invalid Couchbase cluster controller service", "[putcouchbasekey]") {
proc_->setProperty(processors::PutCouchbaseKey::CouchbaseClusterControllerService, "invalid");
- REQUIRE_THROWS_AS(controller_.trigger({minifi::test::InputFlowFileData{"{\"name\": \"John\"}\n{\"name\": \"Jill\"}"}}), minifi::Exception);
+ REQUIRE_THROWS_AS(controller_.trigger({minifi::test::InputFlowFileData{"{\"name\": \"John\"}\n{\"name\": \"Jill\"}", {{"uuid", TEST_UUID}}}}), minifi::Exception);
}
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Invalid bucket name", "[putcouchbasekey]") {
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "");
- auto results = controller_.trigger({minifi::test::InputFlowFileData{"{\"name\": \"John\"}\n{\"name\": \"Jill\"}"}});
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{"{\"name\": \"John\"}\n{\"name\": \"Jill\"}", {{"uuid", TEST_UUID}}}});
REQUIRE(results[processors::PutCouchbaseKey::Failure].size() == 1);
REQUIRE(LogTestController::getInstance().contains("Bucket '' is invalid or empty!", 1s));
}
@@ -123,7 +130,7 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Invalid bucket name", "[putcouc
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with default properties", "[putcouchbasekey]") {
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
- auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input, {{"uuid", TEST_UUID}}}});
verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "_default", "_default",
::couchbase::persist_to::none, ::couchbase::replicate_to::none, CouchbaseValueType::Json, ""}, input);
}
@@ -137,7 +144,7 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with optional pro
proc_->setProperty(processors::PutCouchbaseKey::PersistTo, "ACTIVE");
proc_->setProperty(processors::PutCouchbaseKey::ReplicateTo, "TWO");
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
- auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input, {{"uuid", TEST_UUID}}}});
verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "scope1", "collection1", ::couchbase::persist_to::active,
::couchbase::replicate_to::two, CouchbaseValueType::Binary, "important_doc"}, input);
}
@@ -146,7 +153,7 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put fails with default properti
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
mock_couchbase_cluster_service_->setUpsertError(CouchbaseErrorType::FATAL);
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
- auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input, {{"uuid", TEST_UUID}}}});
verifyResults(results, processors::PutCouchbaseKey::Failure, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none,
CouchbaseValueType::Json, ""}, input);
}
@@ -155,7 +162,7 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retr
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
mock_couchbase_cluster_service_->setUpsertError(CouchbaseErrorType::TEMPORARY);
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
- auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input, {{"uuid", TEST_UUID}}}});
verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none,
CouchbaseValueType::Json, ""}, input);
}
(nifi-minifi-cpp) 11/15: Fix test after rebase
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3fcbc57af8b4ada66af229689529df2b03ca72cd
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 21 15:10:14 2024 +0200
Fix test after rebase
---
libminifi/test/unit/LoggerTests.cpp | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/libminifi/test/unit/LoggerTests.cpp b/libminifi/test/unit/LoggerTests.cpp
index 9c9ec0d51..fc87ee500 100644
--- a/libminifi/test/unit/LoggerTests.cpp
+++ b/libminifi/test/unit/LoggerTests.cpp
@@ -386,7 +386,7 @@ TEST_CASE("fmt formatting works with the logger") {
logger->log_critical("{} in hex is {:#x}", 13, 13);
logger->log_critical("Unix epoch: {}", std::chrono::system_clock::time_point());
logger->log_critical("{:%Q %q} equals to {:%Q %q}", 2h, std::chrono::duration_cast<std::chrono::seconds>(2h));
- CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 1min equals to 60s"));
+ CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 1m equals to 60s"));
CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 13 in hex is 0xd"));
CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] Unix epoch: 1970-01-01 00:00:00.00"));
CHECK(LogTestController::getInstance().contains("[org::apache::nifi::minifi::core::logging::Logger] [critical] 2 h equals to 7200 s"));
(nifi-minifi-cpp) 07/15: Add Document Type property
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 7e8e56c35237da6012f13832316dd7cb9d7aa6f2
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 14 20:31:27 2024 +0200
Add Document Type property
---
PROCESSORS.md | 1 +
docker/requirements.txt | 1 +
.../test/integration/cluster/DockerTestCluster.py | 5 ++
.../cluster/checkers/CouchbaseChecker.py | 69 ++++++++++++++++++++++
.../cluster/containers/CouchbaseServerContainer.py | 1 +
.../features/MiNiFi_integration_test_driver.py | 3 +
docker/test/integration/features/couchbase.feature | 28 ++++++++-
docker/test/integration/features/steps/steps.py | 5 ++
.../controllerservices/CouchbaseClusterService.cpp | 16 ++++-
.../controllerservices/CouchbaseClusterService.h | 16 +++--
.../couchbase/processors/PutCouchbaseKey.cpp | 3 +-
extensions/couchbase/processors/PutCouchbaseKey.h | 8 +++
.../couchbase/tests/MockCouchbaseClusterService.h | 6 +-
.../couchbase/tests/PutCouchbaseKeyTests.cpp | 14 +++--
14 files changed, 160 insertions(+), 16 deletions(-)
diff --git a/PROCESSORS.md b/PROCESSORS.md
index df3148523..fbccbb9c4 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -2219,6 +2219,7 @@ In the list below, the names of required properties appear in bold. Any other pr
| **Bucket Name** | default | | The name of bucket to access.<br/>**Supports Expression Language: true** |
| Scope Name | | | Scope to use inside the bucket. If not specified, the _default scope is used.<br/>**Supports Expression Language: true** |
| Collection Name | | | Collection to use inside the bucket scope. If not specified, the _default collection is used.<br/>**Supports Expression Language: true** |
+| **Document Type** | Json | Json<br/>Binary<br/>String | Content type to store data as. |
| Document Id | | | A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.<br/>**Supports Expression Language: true** |
| **Persist To** | NONE | NONE<br/>ACTIVE<br/>ONE<br/>TWO<br/>THREE<br/>FOUR | Durability constraint about disk persistence. |
| **Replicate To** | NONE | NONE<br/>ONE<br/>TWO<br/>THREE | Durability constraint about replication. |
diff --git a/docker/requirements.txt b/docker/requirements.txt
index 294f836ff..09e799589 100644
--- a/docker/requirements.txt
+++ b/docker/requirements.txt
@@ -12,3 +12,4 @@ azure-storage-blob==12.13.0
prometheus-api-client==0.5.0
humanfriendly==10.0
requests<2.29 # https://github.com/docker/docker-py/issues/3113
+couchbase==4.3.2
diff --git a/docker/test/integration/cluster/DockerTestCluster.py b/docker/test/integration/cluster/DockerTestCluster.py
index f5fb913b7..cf6cc699b 100644
--- a/docker/test/integration/cluster/DockerTestCluster.py
+++ b/docker/test/integration/cluster/DockerTestCluster.py
@@ -35,6 +35,7 @@ from .checkers.PrometheusChecker import PrometheusChecker
from .checkers.SplunkChecker import SplunkChecker
from .checkers.GrafanaLokiChecker import GrafanaLokiChecker
from .checkers.ModbusChecker import ModbusChecker
+from .checkers.CouchbaseChecker import CouchbaseChecker
from utils import get_peak_memory_usage, get_minifi_pid, get_memory_usage, retry_check
@@ -54,6 +55,7 @@ class DockerTestCluster:
self.grafana_loki_checker = GrafanaLokiChecker()
self.minifi_controller_executor = MinifiControllerExecutor(self.container_communicator)
self.modbus_checker = ModbusChecker(self.container_communicator)
+ self.couchbase_checker = CouchbaseChecker()
def cleanup(self):
self.container_store.cleanup()
@@ -424,3 +426,6 @@ class DockerTestCluster:
def enable_ssl_in_nifi(self):
self.container_store.enable_ssl_in_nifi()
+
+ def is_data_present_in_couchbase(self, doc_id: str, bucket_name: str, expected_data: str, expected_data_type: str):
+ return self.couchbase_checker.is_data_present_in_couchbase(doc_id, bucket_name, expected_data, expected_data_type)
diff --git a/docker/test/integration/cluster/checkers/CouchbaseChecker.py b/docker/test/integration/cluster/checkers/CouchbaseChecker.py
new file mode 100644
index 000000000..07a332cbf
--- /dev/null
+++ b/docker/test/integration/cluster/checkers/CouchbaseChecker.py
@@ -0,0 +1,69 @@
+# 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.
+import logging
+import json
+from couchbase.cluster import Cluster
+from couchbase.options import ClusterOptions
+from couchbase.auth import PasswordAuthenticator
+from couchbase.transcoder import RawBinaryTranscoder, RawStringTranscoder
+
+
+class CouchbaseChecker:
+ def is_data_present_in_couchbase(self, doc_id: str, bucket_name: str, expected_data: str, expected_data_type: str):
+ try:
+ cluster = Cluster('couchbase://localhost', ClusterOptions(
+ PasswordAuthenticator('Administrator', 'password123')))
+
+ bucket = cluster.bucket(bucket_name)
+ collection = bucket.default_collection()
+
+ if expected_data_type.lower() == "binary":
+ binary_flag = 0x03 << 24
+ result = collection.get(doc_id, transcoder=RawBinaryTranscoder())
+ flags = result.flags
+ if not flags & binary_flag:
+ logging.error(f"Expected binary data for document '{doc_id}' but no binary flags were found.")
+ return False
+
+ content = result.content_as[bytes]
+ return content.decode('utf-8') == expected_data
+
+ if expected_data_type.lower() == "json":
+ json_flag = 0x02 << 24
+ result = collection.get(doc_id)
+ flags = result.flags
+ if not flags & json_flag:
+ logging.error(f"Expected JSON data for document '{doc_id}' but no JSON flags were found.")
+ return False
+
+ content = result.content_as[dict]
+ return content == json.loads(expected_data)
+
+ if expected_data_type.lower() == "string":
+ string_flag = 0x04 << 24
+ result = collection.get(doc_id, transcoder=RawStringTranscoder())
+ flags = result.flags
+ if not flags & string_flag:
+ logging.error(f"Expected string data for document '{doc_id}' but no string flags were found.")
+ return False
+
+ content = result.content_as[str]
+ return content == expected_data
+
+ logging.error(f"Unsupported data type '{expected_data_type}'")
+ return False
+ except Exception as e:
+ logging.error(f"Error while fetching document '{doc_id}' from bucket '{bucket_name}': {e}")
+ return False
diff --git a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
index 63d2a84c6..1c4353046 100644
--- a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
+++ b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
@@ -51,4 +51,5 @@ class CouchbaseServerContainer(Container):
detach=True,
name=self.name,
network=self.network.name,
+ ports={'11210/tcp': 11210},
entrypoint=self.command)
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index 40b2fe56c..9b3ae5e4f 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -445,3 +445,6 @@ class MiNiFi_integration_test:
def enable_ssl_in_nifi(self):
self.cluster.enable_ssl_in_nifi()
+
+ def check_is_data_present_on_couchbase(self, doc_id: str, bucket_name: str, expected_data: str, expected_data_type: str):
+ assert self.cluster.is_data_present_in_couchbase(doc_id, bucket_name, expected_data, expected_data_type)
diff --git a/docker/test/integration/features/couchbase.feature b/docker/test/integration/features/couchbase.feature
index 2e4fcd8e6..f4e352cbe 100644
--- a/docker/test/integration/features/couchbase.feature
+++ b/docker/test/integration/features/couchbase.feature
@@ -18,12 +18,13 @@ Feature: Executing Couchbase operations from MiNiFi-C++
Background:
Given the content of "/tmp/output" is monitored
- Scenario: A MiNiFi instance can insert data to test bucket with PutCouchbaseKey processor
+ Scenario: A MiNiFi instance can insert json data to test bucket with PutCouchbaseKey processor
Given a Couchbase server is set up
And a GetFile processor with the "Input Directory" property set to "/tmp/input"
And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the PutCouchbaseKey processor is set to "Json"
And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
@@ -39,3 +40,28 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And the Minifi logs match the following regex: "key:couchbase.doc.sequence.number value:[1-9][0-9]*" in less than 1 seconds
And the Minifi logs match the following regex: "key:couchbase.partition.uuid value:[1-9][0-9]*" in less than 1 seconds
And the Minifi logs match the following regex: "key:couchbase.partition.id value:[1-9][0-9]*" in less than 1 seconds
+ And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1", "field2": "value2"}' of type "Json" in Couchbase
+
+ Scenario: A MiNiFi instance can insert binary data to test bucket with PutCouchbaseKey processor
+ Given a Couchbase server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the PutCouchbaseKey processor is set to "Binary"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 60 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.sequence.number value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.partition.uuid value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.partition.id value:[1-9][0-9]*" in less than 1 seconds
+ And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1"}' of type "Binary" in Couchbase
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 7f5857f67..0a76672c3 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -1377,3 +1377,8 @@ def step_impl(context, processor_name, service_name):
processor = context.test.get_node_by_name(processor_name)
processor.controller_services.append(couchbase_cluster_controller_service)
processor.set_property("Couchbase Cluster Controller Service", couchbase_cluster_controller_service.name)
+
+
+@then("a document with id \"{doc_id}\" in bucket \"{bucket_name}\" is present with data '{data}' of type \"{data_type}\" in Couchbase")
+def step_impl(context, doc_id: str, bucket_name: str, data: str, data_type: str):
+ context.test.check_is_data_present_on_couchbase(doc_id, bucket_name, data, data_type)
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index f529a7fc9..d40ada072 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -17,6 +17,9 @@
*/
#include "CouchbaseClusterService.h"
+#include "couchbase/codec/raw_binary_transcoder.hxx"
+#include "couchbase/codec/raw_string_transcoder.hxx"
+#include "couchbase/codec/raw_json_transcoder.hxx"
#include "core/Resource.h"
@@ -39,13 +42,22 @@ nonstd::expected<::couchbase::collection, CouchbaseErrorType> CouchbaseClient::g
}
nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::upsert(const CouchbaseCollection& collection,
- const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
+ CouchbaseValueType document_type, const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
auto collection_result = getCollection(collection);
if (!collection_result.has_value()) {
return nonstd::make_unexpected(collection_result.error());
}
- auto [upsert_err, upsert_resp] = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
+ std::pair<::couchbase::error, ::couchbase::mutation_result> result;
+ if (document_type == CouchbaseValueType::Json) {
+ result = collection_result->upsert<::couchbase::codec::raw_json_transcoder>(document_id, buffer, options).get();
+ } else if (document_type == CouchbaseValueType::String) {
+ std::string data_str(reinterpret_cast<const char*>(buffer.data()), buffer.size());
+ result = collection_result->upsert<::couchbase::codec::raw_string_transcoder>(document_id, data_str, options).get();
+ } else {
+ result = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
+ }
+ auto& [upsert_err, upsert_resp] = result;
if (upsert_err.ec()) {
// ambiguous_timeout should not be retried as we do not know if the insert was successful or not
if (getErrorType(upsert_err.ec()) == CouchbaseErrorType::TEMPORARY && upsert_err.ec().value() != static_cast<int>(::couchbase::errc::common::ambiguous_timeout)) {
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 6b85c37b8..0d46975b0 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -29,8 +29,6 @@
#include "couchbase/cluster.hxx"
#include "core/ProcessContext.h"
#include "core/logging/LoggerConfiguration.h"
-#include "couchbase/codec/raw_binary_transcoder.hxx"
-#include "couchbase/error.hxx"
namespace org::apache::nifi::minifi::couchbase {
@@ -48,6 +46,12 @@ struct CouchbaseUpsertResult {
std::uint16_t partition_id{0};
};
+enum class CouchbaseValueType {
+ Json,
+ Binary,
+ String
+};
+
enum class CouchbaseErrorType {
FATAL,
TEMPORARY,
@@ -59,8 +63,8 @@ class CouchbaseClient {
: connection_string_(std::move(connection_string)), username_(std::move(username)), password_(std::move(password)), logger_(logger) {
}
- nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, const std::string& document_id, const std::vector<std::byte>& buffer,
- const ::couchbase::upsert_options& options);
+ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type, const std::string& document_id,
+ const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options);
std::optional<CouchbaseErrorType> establishConnection();
void close();
@@ -142,10 +146,10 @@ class CouchbaseClusterService : public core::controller::ControllerService {
}
}
- virtual nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection,
+ virtual nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type,
const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
gsl_Expects(client_);
- return client_->upsert(collection, document_id, buffer, options);
+ return client_->upsert(collection, document_type, document_id, buffer, options);
}
static gsl::not_null<std::shared_ptr<CouchbaseClusterService>> getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property);
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.cpp b/extensions/couchbase/processors/PutCouchbaseKey.cpp
index fb36ec4a8..77708debb 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.cpp
+++ b/extensions/couchbase/processors/PutCouchbaseKey.cpp
@@ -25,6 +25,7 @@ namespace org::apache::nifi::minifi::couchbase::processors {
void PutCouchbaseKey::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
couchbase_cluster_service_ = controllers::CouchbaseClusterService::getFromProperty(context, PutCouchbaseKey::CouchbaseClusterControllerService);
+ document_type_ = utils::parseEnumProperty<CouchbaseValueType>(context, PutCouchbaseKey::DocumentType);
persist_to_ = utils::parseEnumProperty<::couchbase::persist_to>(context, PutCouchbaseKey::PersistTo);
replicate_to_ = utils::parseEnumProperty<::couchbase::replicate_to>(context, PutCouchbaseKey::ReplicateTo);
}
@@ -61,7 +62,7 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
::couchbase::upsert_options options;
options.durability(persist_to_, replicate_to_);
auto result = session.readBuffer(flow_file);
- if (auto upsert_result = couchbase_cluster_service_->upsert(collection, document_id, result.buffer, options)) {
+ if (auto upsert_result = couchbase_cluster_service_->upsert(collection, document_type_, document_id, result.buffer, options)) {
session.putAttribute(*flow_file, "couchbase.bucket", upsert_result->bucket_name);
session.putAttribute(*flow_file, "couchbase.doc.id", document_id);
session.putAttribute(*flow_file, "couchbase.doc.cas", std::to_string(upsert_result->cas));
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.h b/extensions/couchbase/processors/PutCouchbaseKey.h
index 91975e62a..ca79b7dba 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.h
+++ b/extensions/couchbase/processors/PutCouchbaseKey.h
@@ -93,6 +93,12 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
.withDescription("Collection to use inside the bucket scope. If not specified, the _default collection is used.")
.supportsExpressionLanguage(true)
.build();
+ EXTENSIONAPI static constexpr auto DocumentType = core::PropertyDefinitionBuilder<3>::createProperty("Document Type")
+ .withDescription("Content type to store data as.")
+ .isRequired(true)
+ .withDefaultValue(magic_enum::enum_name(CouchbaseValueType::Json))
+ .withAllowedValues(magic_enum::enum_names<CouchbaseValueType>())
+ .build();
EXTENSIONAPI static constexpr auto DocumentId = core::PropertyDefinitionBuilder<>::createProperty("Document Id")
.withDescription("A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. If not specified, the FlowFile UUID will be used.")
.supportsExpressionLanguage(true)
@@ -115,6 +121,7 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
BucketName,
ScopeName,
CollectionName,
+ DocumentType,
DocumentId,
PersistTo,
ReplicateTo
@@ -145,6 +152,7 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
private:
std::shared_ptr<controllers::CouchbaseClusterService> couchbase_cluster_service_;
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<PutCouchbaseKey>::getLogger(uuid_);
+ CouchbaseValueType document_type_ = CouchbaseValueType::Json;
::couchbase::persist_to persist_to_ = ::couchbase::persist_to::none;
::couchbase::replicate_to replicate_to_ = ::couchbase::replicate_to::none;
};
diff --git a/extensions/couchbase/tests/MockCouchbaseClusterService.h b/extensions/couchbase/tests/MockCouchbaseClusterService.h
index 521ab3bdf..a2757f16c 100644
--- a/extensions/couchbase/tests/MockCouchbaseClusterService.h
+++ b/extensions/couchbase/tests/MockCouchbaseClusterService.h
@@ -31,6 +31,7 @@ const std::uint64_t COUCHBASE_PUT_RESULT_PARTITION_UUID = 7890123456;
const std::uint16_t COUCHBASE_PUT_RESULT_PARTITION_ID = 1234;
struct UpsertParameters {
+ CouchbaseValueType document_type;
std::string document_id;
std::vector<std::byte> buffer;
::couchbase::upsert_options options;
@@ -45,9 +46,10 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
void onEnable() override {}
void notifyStop() override {}
- nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, const std::string& document_id, const std::vector<std::byte>& buffer,
- const ::couchbase::upsert_options& options) override {
+ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type, const std::string& document_id,
+ const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) override {
collection_ = collection;
+ upsert_parameters_.document_type = document_type;
upsert_parameters_.document_id = document_id;
upsert_parameters_.buffer = buffer;
upsert_parameters_.options = options;
diff --git a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
index b65f39d00..7adb02394 100644
--- a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
+++ b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
@@ -33,6 +33,7 @@ struct ExpectedCallOptions {
std::string collection_name;
::couchbase::persist_to persist_to;
::couchbase::replicate_to replicate_to;
+ CouchbaseValueType document_type;
std::string doc_id;
};
@@ -80,6 +81,7 @@ class PutCouchbaseKeyTestController : public TestController {
CHECK(get_collection_parameters.scope_name == expected_call_options.scope_name);
auto upsert_parameters = mock_couchbase_cluster_service_->getUpsertParameters();
+ CHECK(upsert_parameters.document_type == expected_call_options.document_type);
std::string expected_doc_id = expected_call_options.doc_id.empty() ? flow_file->getUUID().to_string() : expected_call_options.doc_id;
CHECK(upsert_parameters.document_id == expected_doc_id);
CHECK(upsert_parameters.buffer == stringToByteVector(input));
@@ -122,20 +124,22 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with default prop
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
- verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+ verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "_default", "_default",
+ ::couchbase::persist_to::none, ::couchbase::replicate_to::none, CouchbaseValueType::Json, ""}, input);
}
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with optional properties", "[putcouchbasekey]") {
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
proc_->setProperty(processors::PutCouchbaseKey::ScopeName, "scope1");
proc_->setProperty(processors::PutCouchbaseKey::CollectionName, "collection1");
+ proc_->setProperty(processors::PutCouchbaseKey::DocumentType, "Binary");
proc_->setProperty(processors::PutCouchbaseKey::DocumentId, "important_doc");
proc_->setProperty(processors::PutCouchbaseKey::PersistTo, "ACTIVE");
proc_->setProperty(processors::PutCouchbaseKey::ReplicateTo, "TWO");
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
verifyResults(results, processors::PutCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "scope1", "collection1", ::couchbase::persist_to::active,
- ::couchbase::replicate_to::two, "important_doc"}, input);
+ ::couchbase::replicate_to::two, CouchbaseValueType::Binary, "important_doc"}, input);
}
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put fails with default properties", "[putcouchbasekey]") {
@@ -143,7 +147,8 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put fails with default properti
mock_couchbase_cluster_service_->setUpsertError(CouchbaseErrorType::FATAL);
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
- verifyResults(results, processors::PutCouchbaseKey::Failure, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+ verifyResults(results, processors::PutCouchbaseKey::Failure, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none,
+ CouchbaseValueType::Json, ""}, input);
}
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when temporary error is returned", "[putcouchbasekey]") {
@@ -151,7 +156,8 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retr
mock_couchbase_cluster_service_->setUpsertError(CouchbaseErrorType::TEMPORARY);
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
- verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
+ verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none,
+ CouchbaseValueType::Json, ""}, input);
}
} // namespace org::apache::nifi::minifi::couchbase::test
(nifi-minifi-cpp) 06/15: Refactor
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 50f738d0c20d231f23f94d583dbd0abb7e45aa39
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Mon Oct 14 10:29:17 2024 +0200
Refactor
---
extensions/couchbase/tests/MockCouchbaseClusterService.h | 7 ++++++-
extensions/couchbase/tests/PutCouchbaseKeyTests.cpp | 8 ++++----
2 files changed, 10 insertions(+), 5 deletions(-)
diff --git a/extensions/couchbase/tests/MockCouchbaseClusterService.h b/extensions/couchbase/tests/MockCouchbaseClusterService.h
index 5f2bc08a6..521ab3bdf 100644
--- a/extensions/couchbase/tests/MockCouchbaseClusterService.h
+++ b/extensions/couchbase/tests/MockCouchbaseClusterService.h
@@ -25,6 +25,11 @@ using namespace std::literals::chrono_literals;
namespace org::apache::nifi::minifi::couchbase::test {
+const std::uint64_t COUCHBASE_PUT_RESULT_CAS = 9876;
+const std::uint64_t COUCHBASE_PUT_RESULT_SEQUENCE_NUMBER = 345;
+const std::uint64_t COUCHBASE_PUT_RESULT_PARTITION_UUID = 7890123456;
+const std::uint16_t COUCHBASE_PUT_RESULT_PARTITION_ID = 1234;
+
struct UpsertParameters {
std::string document_id;
std::vector<std::byte> buffer;
@@ -50,7 +55,7 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
if (upsert_error_) {
return nonstd::make_unexpected(*upsert_error_);
} else {
- return CouchbaseUpsertResult{std::string(collection_.bucket_name), 1, 2, 3, 4};
+ return CouchbaseUpsertResult{collection_.bucket_name, COUCHBASE_PUT_RESULT_CAS, COUCHBASE_PUT_RESULT_SEQUENCE_NUMBER, COUCHBASE_PUT_RESULT_PARTITION_UUID, COUCHBASE_PUT_RESULT_PARTITION_ID};
}
}
diff --git a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
index ec2272d33..b65f39d00 100644
--- a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
+++ b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
@@ -94,10 +94,10 @@ class PutCouchbaseKeyTestController : public TestController {
CHECK(flow_file->getAttribute("couchbase.bucket").value() == expected_call_options.bucket_name);
CHECK(flow_file->getAttribute("couchbase.doc.id").value() == expected_doc_id);
- CHECK(flow_file->getAttribute("couchbase.doc.cas").value() == "1");
- CHECK(flow_file->getAttribute("couchbase.doc.sequence.number").value() == "2");
- CHECK(flow_file->getAttribute("couchbase.partition.uuid").value() == "3");
- CHECK(flow_file->getAttribute("couchbase.partition.id").value() == "4");
+ CHECK(flow_file->getAttribute("couchbase.doc.cas").value() == std::to_string(COUCHBASE_PUT_RESULT_CAS));
+ CHECK(flow_file->getAttribute("couchbase.doc.sequence.number").value() == std::to_string(COUCHBASE_PUT_RESULT_SEQUENCE_NUMBER));
+ CHECK(flow_file->getAttribute("couchbase.partition.uuid").value() == std::to_string(COUCHBASE_PUT_RESULT_PARTITION_UUID));
+ CHECK(flow_file->getAttribute("couchbase.partition.id").value() == std::to_string(COUCHBASE_PUT_RESULT_PARTITION_ID));
}
protected:
(nifi-minifi-cpp) 02/15: Handle temporary errors on connection initialization
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3593eeb92dadf647d266d170f0274f470756a278
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Oct 9 14:06:49 2024 +0200
Handle temporary errors on connection initialization
---
extensions/couchbase/CouchbaseCollection.h | 40 ----------
.../controllerservices/CouchbaseClusterService.cpp | 89 +++++++++++++++-------
.../controllerservices/CouchbaseClusterService.h | 61 ++++++++++-----
.../couchbase/processors/PutCouchbaseKey.cpp | 34 ++++-----
.../couchbase/tests/MockCouchbaseClusterService.h | 68 ++++-------------
.../couchbase/tests/PutCouchbaseKeyTests.cpp | 20 +----
6 files changed, 135 insertions(+), 177 deletions(-)
diff --git a/extensions/couchbase/CouchbaseCollection.h b/extensions/couchbase/CouchbaseCollection.h
deleted file mode 100644
index 7968bdd8c..000000000
--- a/extensions/couchbase/CouchbaseCollection.h
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- *
- * 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.
- */
-
-#pragma once
-
-#include "couchbase/upsert_options.hxx"
-#include "nonstd/expected.hpp"
-
-namespace org::apache::nifi::minifi::couchbase {
-
-struct CouchbaseUpsertResult {
- std::string bucket_name;
- std::uint64_t cas{0};
- std::uint64_t sequence_number{0};
- std::uint64_t partition_uuid{0};
- std::uint16_t partition_id{0};
-};
-
-class CouchbaseCollection {
- public:
- virtual nonstd::expected<CouchbaseUpsertResult, std::error_code> upsert(const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) = 0;
- virtual ~CouchbaseCollection() = default;
-};
-
-} // namespace org::apache::nifi::minifi::couchbase
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index 02b43283a..8db837198 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -22,19 +22,48 @@
namespace org::apache::nifi::minifi::couchbase {
-nonstd::expected<CouchbaseUpsertResult, std::error_code> RemoteCouchbaseCollection::upsert(const std::string& document_id, const std::vector<std::byte>& buffer,
- const ::couchbase::upsert_options& options) {
- auto [err, resp] = collection_.upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
- if (err.ec()) {
- client_.setConnectionError();
- return nonstd::make_unexpected(err.ec());
+CouchbaseErrorType CouchbaseClient::getErrorType(const std::error_code& error_code) {
+ for (const auto& temporary_error : temporary_connection_errors) {
+ if (static_cast<int>(temporary_error) == error_code.value()) {
+ return CouchbaseErrorType::TEMPORARY;
+ }
+ }
+ return CouchbaseErrorType::FATAL;
+}
+
+nonstd::expected<::couchbase::collection, CouchbaseErrorType> CouchbaseClient::getCollection(const CouchbaseCollection& collection) {
+ if (auto error_type = establishConnection()) {
+ return nonstd::make_unexpected(*error_type);
+ }
+ return cluster_.bucket(collection.bucket_name).scope(collection.scope_name).collection(collection.collection_name);
+}
+
+nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::upsert(const CouchbaseCollection& collection,
+ const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
+ auto collection_result = getCollection(collection);
+ if (!collection_result.has_value()) {
+ return nonstd::make_unexpected(collection_result.error());
+ }
+
+ auto [upsert_err, upsert_resp] = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
+ if (upsert_err.ec()) {
+ setConnectionError();
+ if (upsert_err.ec().value() == static_cast<int>(::couchbase::errc::common::unambiguous_timeout)) {
+ logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to timeout",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
+ return nonstd::make_unexpected(CouchbaseErrorType::TEMPORARY);
+ }
+ std::string cause = upsert_err.cause() ? upsert_err.cause()->message() : "";
+ logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' with error code: '{}', message: '{}'",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name, upsert_err.ec(), upsert_err.message());
+ return nonstd::make_unexpected(CouchbaseErrorType::FATAL);
} else {
- const uint64_t partition_uuid = (resp.mutation_token().has_value() ? resp.mutation_token()->partition_uuid() : 0);
- const uint64_t sequence_number = (resp.mutation_token().has_value() ? resp.mutation_token()->sequence_number() : 0);
- const uint16_t partition_id = (resp.mutation_token().has_value() ? resp.mutation_token()->partition_id() : 0);
+ const uint64_t partition_uuid = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_uuid() : 0);
+ const uint64_t sequence_number = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->sequence_number() : 0);
+ const uint16_t partition_id = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_id() : 0);
return CouchbaseUpsertResult {
- collection_.bucket_name(),
- resp.cas().value(),
+ std::string(collection.bucket_name),
+ upsert_resp.cas().value(),
partition_uuid,
sequence_number,
partition_id
@@ -42,19 +71,12 @@ nonstd::expected<CouchbaseUpsertResult, std::error_code> RemoteCouchbaseCollecti
}
}
-std::unique_ptr<CouchbaseCollection> CouchBaseClient::getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name) {
- if (!establishConnection()) {
- return nullptr;
- }
- return std::make_unique<RemoteCouchbaseCollection>(cluster_.bucket(bucket_name).scope(scope_name).collection(collection_name), *this);
-}
-
-void CouchBaseClient::setConnectionError() {
+void CouchbaseClient::setConnectionError() {
std::lock_guard<std::mutex> lock(state_mutex_);
state_ = State::UNKNOWN;
}
-void CouchBaseClient::close() {
+void CouchbaseClient::close() {
std::lock_guard<std::mutex> lock(state_mutex_);
if (state_ == State::CONNECTED || state_ == State::UNKNOWN) {
cluster_.close().wait();
@@ -62,31 +84,32 @@ void CouchBaseClient::close() {
}
}
-bool CouchBaseClient::establishConnection() {
+std::optional<CouchbaseErrorType> CouchbaseClient::establishConnection() {
std::lock_guard<std::mutex> lock(state_mutex_);
if (state_ == State::CONNECTED) {
- return true;
+ return std::nullopt;
}
if (state_ == State::UNKNOWN) {
- auto [err, resp] = cluster_.ping().get();
+ auto [err, upsert_resp] = cluster_.ping().get();
if (err.ec()) {
close();
} else {
state_ = State::CONNECTED;
- return true;
+ return std::nullopt;
}
}
auto options = ::couchbase::cluster_options(username_, password_);
auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, options).get();
if (connect_err.ec()) {
- logger_->log_error("Failed to connect to Couchbase cluster: {}", connect_err.message());
- return false;
+ std::string cause = connect_err.cause() ? connect_err.cause()->message() : "";
+ logger_->log_error("Failed to connect to Couchbase cluster with error code: '{}' and message: '{}'", connect_err.ec(), connect_err.message());
+ return getErrorType(connect_err.ec());
}
cluster_ = std::move(cluster);
state_ = State::CONNECTED;
- return true;
+ return std::nullopt;
}
namespace controllers {
@@ -102,7 +125,17 @@ void CouchbaseClusterService::onEnable() {
getProperty(UserName, username);
std::string password;
getProperty(UserPassword, password);
- client_ = std::make_unique<CouchBaseClient>(connection_string, username, password, logger_);
+ if (connection_string.empty() || username.empty() || password.empty()) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing connection string, username or password");
+ }
+
+ client_ = std::make_unique<CouchbaseClient>(connection_string, username, password, logger_);
+ if (auto result = client_->establishConnection()) {
+ if (result == CouchbaseErrorType::FATAL) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Failed to connect to Couchbase cluster with fatal error");
+ }
+ logger_->log_warn("Failed to connect to Couchbase cluster with temporary error, will retry connection when a Couchbase processor is triggered");
+ }
}
gsl::not_null<std::shared_ptr<CouchbaseClusterService>> CouchbaseClusterService::getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property) {
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 0fc3c0cc3..9b2fa6831 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -29,28 +29,31 @@
#include "couchbase/cluster.hxx"
#include "core/ProcessContext.h"
#include "core/logging/LoggerConfiguration.h"
-#include "CouchbaseCollection.h"
#include "couchbase/codec/raw_binary_transcoder.hxx"
+#include "couchbase/error.hxx"
namespace org::apache::nifi::minifi::couchbase {
-class CouchBaseClient;
-
-class RemoteCouchbaseCollection : public CouchbaseCollection {
- public:
- explicit RemoteCouchbaseCollection(::couchbase::collection collection, CouchBaseClient& client)
- : collection_(std::move(collection)),
- client_(client) {
- }
+struct CouchbaseCollection {
+ std::string bucket_name;
+ std::string scope_name;
+ std::string collection_name;
+};
- nonstd::expected<CouchbaseUpsertResult, std::error_code> upsert(const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) override;
+struct CouchbaseUpsertResult {
+ std::string bucket_name;
+ std::uint64_t cas{0};
+ std::uint64_t sequence_number{0};
+ std::uint64_t partition_uuid{0};
+ std::uint16_t partition_id{0};
+};
- private:
- ::couchbase::collection collection_;
- CouchBaseClient& client_;
+enum class CouchbaseErrorType {
+ FATAL,
+ TEMPORARY,
};
-class CouchBaseClient {
+class CouchbaseClient {
public:
enum class State {
DISCONNECTED,
@@ -58,16 +61,31 @@ class CouchBaseClient {
UNKNOWN,
};
- CouchBaseClient(std::string connection_string, std::string username, std::string password, const std::shared_ptr<core::logging::Logger>& logger)
+ CouchbaseClient(std::string connection_string, std::string username, std::string password, const std::shared_ptr<core::logging::Logger>& logger)
: connection_string_(std::move(connection_string)), username_(std::move(username)), password_(std::move(password)), logger_(logger) {
}
- std::unique_ptr<CouchbaseCollection> getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name);
- void setConnectionError();
+ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, const std::string& document_id, const std::vector<std::byte>& buffer,
+ const ::couchbase::upsert_options& options);
+ std::optional<CouchbaseErrorType> establishConnection();
void close();
private:
- bool establishConnection();
+ static constexpr std::array<::couchbase::errc::common, 9> temporary_connection_errors = {
+ ::couchbase::errc::common::temporary_failure,
+ ::couchbase::errc::common::request_canceled,
+ ::couchbase::errc::common::service_not_available,
+ ::couchbase::errc::common::internal_server_failure,
+ ::couchbase::errc::common::cas_mismatch,
+ ::couchbase::errc::common::ambiguous_timeout,
+ ::couchbase::errc::common::unambiguous_timeout,
+ ::couchbase::errc::common::rate_limited,
+ ::couchbase::errc::common::quota_limited
+ };
+
+ static CouchbaseErrorType getErrorType(const std::error_code& error_code);
+ nonstd::expected<::couchbase::collection, CouchbaseErrorType> getCollection(const CouchbaseCollection& collection);
+ void setConnectionError();
std::mutex state_mutex_;
State state_ = State::DISCONNECTED;
@@ -134,15 +152,16 @@ class CouchbaseClusterService : public core::controller::ControllerService {
}
}
- virtual std::unique_ptr<CouchbaseCollection> getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name) {
+ virtual nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection,
+ const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) {
gsl_Expects(client_);
- return client_->getCollection(bucket_name, scope_name, collection_name);
+ return client_->upsert(collection, document_id, buffer, options);
}
static gsl::not_null<std::shared_ptr<CouchbaseClusterService>> getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property);
private:
- std::unique_ptr<CouchBaseClient> client_;
+ std::unique_ptr<CouchbaseClient> client_;
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<CouchbaseClusterService>::getLogger(uuid_);
};
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.cpp b/extensions/couchbase/processors/PutCouchbaseKey.cpp
index 4f4062d2d..15771afca 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.cpp
+++ b/extensions/couchbase/processors/PutCouchbaseKey.cpp
@@ -38,21 +38,19 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
return;
}
- std::string bucket_name;
- if (!context.getProperty(BucketName, bucket_name, flow_file.get()) || bucket_name.empty()) {
- logger_->log_error("Bucket '{}' is invalid or empty!", bucket_name);
+ CouchbaseCollection collection;
+ if (!context.getProperty(BucketName, collection.bucket_name, flow_file.get()) || collection.bucket_name.empty()) {
+ logger_->log_error("Bucket '{}' is invalid or empty!", collection.bucket_name);
session.transfer(flow_file, Failure);
return;
}
- std::string scope_name;
- if (!context.getProperty(ScopeName, scope_name, flow_file.get()) || scope_name.empty()) {
- scope_name = ::couchbase::scope::default_name;
+ if (!context.getProperty(ScopeName, collection.scope_name, flow_file.get()) || collection.scope_name.empty()) {
+ collection.scope_name = ::couchbase::scope::default_name;
}
- std::string collection_name;
- if (!context.getProperty(CollectionName, collection_name, flow_file.get()) || collection_name.empty()) {
- collection_name = ::couchbase::collection::default_name;
+ if (!context.getProperty(CollectionName, collection.collection_name, flow_file.get()) || collection.collection_name.empty()) {
+ collection.collection_name = ::couchbase::collection::default_name;
}
std::string document_id;
@@ -60,28 +58,24 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
document_id = flow_file->getUUIDStr();
}
- auto collection = couchbase_cluster_service_->getCollection(bucket_name, scope_name, collection_name);
- if (!collection) {
- logger_->log_error("Failed to get collection '{}.{}.{}', transferring to retry relationship", bucket_name, scope_name, collection_name);
- session.transfer(flow_file, Retry);
- return;
- }
::couchbase::upsert_options options;
options.durability(persist_to_, replicate_to_);
auto result = session.readBuffer(flow_file);
- if (auto upsert_result = collection->upsert(document_id, result.buffer, options)) {
- session.putAttribute(*flow_file, "couchbase.bucket", upsert_result->bucket_name);
+ if (auto upsert_result = couchbase_cluster_service_->upsert(collection, document_id, result.buffer, options)) {
+ session.putAttribute(*flow_file, "couchbase.bucket", std::string(upsert_result->bucket_name));
session.putAttribute(*flow_file, "couchbase.doc.id", document_id);
session.putAttribute(*flow_file, "couchbase.doc.cas", std::to_string(upsert_result->cas));
session.putAttribute(*flow_file, "couchbase.doc.sequence.number", std::to_string(upsert_result->sequence_number));
session.putAttribute(*flow_file, "couchbase.partition.uuid", std::to_string(upsert_result->partition_uuid));
session.putAttribute(*flow_file, "couchbase.partition.id", std::to_string(upsert_result->partition_id));
session.transfer(flow_file, Success);
- } else if (upsert_result.error().value() == static_cast<int>(::couchbase::errc::common::unambiguous_timeout)) {
- logger_->log_error("Failed to upsert document '{}' due to timeout, transferring to retry relationship", document_id);
+ } else if (upsert_result.error() == CouchbaseErrorType::TEMPORARY) {
+ logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to timeout, transferring to retry relationship",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
session.transfer(flow_file, Retry);
} else {
- logger_->log_error("Failed to upsert document '{}': {}", document_id, upsert_result.error().message());
+ logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}', transferring to failure relationship",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
session.transfer(flow_file, Failure);
}
}
diff --git a/extensions/couchbase/tests/MockCouchbaseClusterService.h b/extensions/couchbase/tests/MockCouchbaseClusterService.h
index 86e749dfe..396bff4fd 100644
--- a/extensions/couchbase/tests/MockCouchbaseClusterService.h
+++ b/extensions/couchbase/tests/MockCouchbaseClusterService.h
@@ -25,43 +25,12 @@ using namespace std::literals::chrono_literals;
namespace org::apache::nifi::minifi::couchbase::test {
-struct GetCollectionParameters {
- std::string bucket_name;
- std::string scope_name;
- std::string collection_name;
-};
-
struct UpsertParameters {
std::string document_id;
std::vector<std::byte> buffer;
::couchbase::upsert_options options;
};
-class MockCouchbaseCollection : public CouchbaseCollection {
- public:
- MockCouchbaseCollection(UpsertParameters& parameters, std::error_code& upsert_error_code, std::string_view bucket_name)
- : parameters_(parameters),
- upsert_error_code_(upsert_error_code),
- bucket_name_(bucket_name) {}
-
- nonstd::expected<CouchbaseUpsertResult, std::error_code> upsert(const std::string& document_id, const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options) override {
- parameters_.document_id = document_id;
- parameters_.buffer = buffer;
- parameters_.options = options;
-
- if (upsert_error_code_) {
- return nonstd::make_unexpected(upsert_error_code_);
- } else {
- return CouchbaseUpsertResult{bucket_name_, 1, 2, 3, 4};
- }
- }
-
- private:
- UpsertParameters& parameters_;
- std::error_code& upsert_error_code_;
- std::string bucket_name_;
-};
-
class MockCouchbaseClusterService : public controllers::CouchbaseClusterService {
public:
using CouchbaseClusterService::CouchbaseClusterService;
@@ -71,41 +40,36 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
void onEnable() override {}
void notifyStop() override {}
- std::unique_ptr<CouchbaseCollection> getCollection(std::string_view bucket_name, std::string_view scope_name, std::string_view collection_name) override {
- get_collection_parameters_.bucket_name = bucket_name;
- get_collection_parameters_.scope_name = scope_name;
- get_collection_parameters_.collection_name = collection_name;
- if (!get_collection_succeeds_) {
- return nullptr;
+ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, const std::string& document_id, const std::vector<std::byte>& buffer,
+ const ::couchbase::upsert_options& options) override {
+ collection_ = collection;
+ upsert_parameters_.document_id = document_id;
+ upsert_parameters_.buffer = buffer;
+ upsert_parameters_.options = options;
+
+ if (upsert_error_) {
+ return nonstd::make_unexpected(*upsert_error_);
} else {
- return std::make_unique<MockCouchbaseCollection>(upsert_parameters_, upsert_error_code_, bucket_name);
+ return CouchbaseUpsertResult{std::string(collection_.bucket_name), 1, 2, 3, 4};
}
}
- GetCollectionParameters getGetCollectionParameters() const {
- return get_collection_parameters_;
- }
-
UpsertParameters getUpsertParameters() const {
return upsert_parameters_;
}
- void setUpsertErrorCode(const std::error_code& error) {
- upsert_error_code_ = error;
- }
-
- void setGetCollectionSucceeds(bool succeeds) {
- get_collection_succeeds_ = succeeds;
+ CouchbaseCollection getCollectionParameter() const {
+ return collection_;
}
- bool getGetCollectionSucceeds() const {
- return get_collection_succeeds_;
+ void setUpsertError(const CouchbaseErrorType upsert_error) {
+ upsert_error_ = upsert_error;
}
private:
- GetCollectionParameters get_collection_parameters_;
+ CouchbaseCollection collection_;
UpsertParameters upsert_parameters_;
- std::error_code upsert_error_code_;
+ std::optional<CouchbaseErrorType> upsert_error_;
bool get_collection_succeeds_{true};
};
} // namespace org::apache::nifi::minifi::couchbase::test
diff --git a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
index e5953beb7..ec2272d33 100644
--- a/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
+++ b/extensions/couchbase/tests/PutCouchbaseKeyTests.cpp
@@ -74,15 +74,11 @@ class PutCouchbaseKeyTestController : public TestController {
flow_file = results.at(processors::PutCouchbaseKey::Retry)[0];
}
- auto get_collection_parameters = mock_couchbase_cluster_service_->getGetCollectionParameters();
+ auto get_collection_parameters = mock_couchbase_cluster_service_->getCollectionParameter();
CHECK(get_collection_parameters.bucket_name == expected_call_options.bucket_name);
CHECK(get_collection_parameters.collection_name == expected_call_options.collection_name);
CHECK(get_collection_parameters.scope_name == expected_call_options.scope_name);
- if (!mock_couchbase_cluster_service_->getGetCollectionSucceeds()) {
- return;
- }
-
auto upsert_parameters = mock_couchbase_cluster_service_->getUpsertParameters();
std::string expected_doc_id = expected_call_options.doc_id.empty() ? flow_file->getUUID().to_string() : expected_call_options.doc_id;
CHECK(upsert_parameters.document_id == expected_doc_id);
@@ -144,23 +140,15 @@ TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put succeeeds with optional pro
TEST_CASE_METHOD(PutCouchbaseKeyTestController, "Put fails with default properties", "[putcouchbasekey]") {
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
- mock_couchbase_cluster_service_->setUpsertErrorCode(std::make_error_code(std::errc::invalid_argument));
+ mock_couchbase_cluster_service_->setUpsertError(CouchbaseErrorType::FATAL);
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
verifyResults(results, processors::PutCouchbaseKey::Failure, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
}
-TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when get collection fails", "[putcouchbasekey]") {
- proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
- mock_couchbase_cluster_service_->setGetCollectionSucceeds(false);
- const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
- auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
- verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
-}
-
-TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when unambiguous timeout is returned", "[putcouchbasekey]") {
+TEST_CASE_METHOD(PutCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when temporary error is returned", "[putcouchbasekey]") {
proc_->setProperty(processors::PutCouchbaseKey::BucketName, "mybucket");
- mock_couchbase_cluster_service_->setUpsertErrorCode(::couchbase::errc::common::unambiguous_timeout);
+ mock_couchbase_cluster_service_->setUpsertError(CouchbaseErrorType::TEMPORARY);
const std::string input = "{\"name\": \"John\"}\n{\"name\": \"Jill\"}";
auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
verifyResults(results, processors::PutCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", ::couchbase::persist_to::none, ::couchbase::replicate_to::none, ""}, input);
(nifi-minifi-cpp) 13/15: MINIFICPP-2469 Create GetCouchbaseKey processor
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 87e65111b3717df4ac918e17233e2089c8841a33
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Tue Oct 8 16:54:16 2024 +0200
MINIFICPP-2469 Create GetCouchbaseKey processor
---
PROCESSORS.md | 40 +++++
.../cluster/containers/CouchbaseServerContainer.py | 4 +-
docker/test/integration/features/couchbase.feature | 108 ++++++++++++++
.../{PutCouchbaseKey.py => GetCouchbaseKey.py} | 8 +-
.../minifi/processors/PutCouchbaseKey.py | 2 +-
.../controllerservices/CouchbaseClusterService.cpp | 62 ++++++--
.../controllerservices/CouchbaseClusterService.h | 17 ++-
.../couchbase/processors/GetCouchbaseKey.cpp | 111 ++++++++++++++
.../{PutCouchbaseKey.h => GetCouchbaseKey.h} | 93 +++---------
extensions/couchbase/processors/PutCouchbaseKey.h | 1 -
.../couchbase/tests/GetCouchbaseKeyTests.cpp | 163 +++++++++++++++++++++
.../couchbase/tests/MockCouchbaseClusterService.h | 35 ++++-
libminifi/include/core/Processor.h | 4 +-
13 files changed, 555 insertions(+), 93 deletions(-)
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 274393223..bb9ea9e32 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -48,6 +48,7 @@ limitations under the License.
- [FetchSmb](#FetchSmb)
- [FocusArchiveEntry](#FocusArchiveEntry)
- [GenerateFlowFile](#GenerateFlowFile)
+- [GetCouchbaseKey](#GetCouchbaseKey)
- [GetFile](#GetFile)
- [GetTCP](#GetTCP)
- [HashContent](#HashContent)
@@ -1094,6 +1095,45 @@ In the list below, the names of required properties appear in bold. Any other pr
| success | success operational on the flow record |
+## GetCouchbaseKey
+
+### Description
+
+Get a document from Couchbase Server via Key/Value access. The ID of the document to fetch may be supplied by setting the <Document Id> property. NOTE: if the Document Id property is not set, the contents of the FlowFile will be read to determine the Document Id, which means that the contents of the entire FlowFile will be buffered in memory.
+
+### Properties
+
+In the list below, the names of required properties appear in bold. Any other properties (not in bold) are considered optional. The table also indicates any default values, and whether a property supports the NiFi Expression Language.
+
+| Name | Default Value | Allowable Values | Description |
+|------------------------------------------|---------------|----------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| **Couchbase Cluster Controller Service** | | | A Couchbase Cluster Controller Service which manages connections to a Couchbase cluster. |
+| **Bucket Name** | default | | The name of bucket to access.<br/>**Supports Expression Language: true** |
+| Scope Name | | | Scope to use inside the bucket. If not specified, the _default scope is used.<br/>**Supports Expression Language: true** |
+| Collection Name | | | Collection to use inside the bucket scope. If not specified, the _default collection is used.<br/>**Supports Expression Language: true** |
+| **Document Type** | Json | Json<br/>Binary<br/>String | Content type of the retrieved value. |
+| Document Id | | | A static, fixed Couchbase document id, or an expression to construct the Couchbase document id.<br/>**Supports Expression Language: true** |
+| Put Value to Attribute | | | If set, the retrieved value will be put into an attribute of the FlowFile instead of a the content of the FlowFile. The attribute key to put to is determined by evaluating value of this property.<br/>**Supports Expression Language: true** |
+
+### Relationships
+
+| Name | Description |
+|----------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| success | Values retrieved from Couchbase Server are written as outgoing FlowFiles content or put into an attribute of the incoming FlowFile and routed to this relationship. |
+| failure | All FlowFiles failed to fetch from Couchbase Server and not retry-able are routed to this relationship. |
+| retry | All FlowFiles failed to fetch from Couchbase Server but can be retried are routed to this relationship. |
+| original | The original input FlowFile is routed to this relationship when the value is retrieved from Couchbase Server and routed to 'success'. |
+
+### Output Attributes
+
+| Attribute | Relationship | Description |
+|----------------------|--------------|---------------------------------------|
+| couchbase.bucket | success | Bucket where the document was stored. |
+| couchbase.doc.id | success | Id of the document. |
+| couchbase.doc.cas | success | CAS of the document. |
+| couchbase.doc.expiry | success | Expiration of the document. |
+
+
## GetFile
### Description
diff --git a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
index 1c4353046..9baf01f08 100644
--- a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
+++ b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
@@ -33,7 +33,7 @@ class CouchbaseServerContainer(Container):
["couchbase-cli", "cluster-init", "-c", "localhost", "--cluster-username", "Administrator", "--cluster-password", "password123", "--services", "data,index,query",
"--cluster-ramsize", "2048", "--cluster-index-ramsize", "256"],
["couchbase-cli", "bucket-create", "-c", "localhost", "--username", "Administrator", "--password", "password123", "--bucket", "test_bucket", "--bucket-type", "couchbase",
- "--bucket-ramsize", "1024"]
+ "--bucket-ramsize", "1024", "--max-ttl", "36000"]
]
for command in commands:
(code, _) = self.client.containers.get(self.name).exec_run(command)
@@ -47,7 +47,7 @@ class CouchbaseServerContainer(Container):
return
self.docker_container = self.client.containers.run(
- "couchbase:community-7.6.2",
+ "couchbase:enterprise-7.2.5",
detach=True,
name=self.name,
network=self.network.name,
diff --git a/docker/test/integration/features/couchbase.feature b/docker/test/integration/features/couchbase.feature
index 446821f25..d9a4bb664 100644
--- a/docker/test/integration/features/couchbase.feature
+++ b/docker/test/integration/features/couchbase.feature
@@ -65,3 +65,111 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And the Minifi logs match the following regex: "key:couchbase.partition.uuid value:[1-9][0-9]*" in less than 1 seconds
And the Minifi logs match the following regex: "key:couchbase.partition.id value:[1-9][0-9]*" in less than 1 seconds
And a document with id "test_doc_id" in bucket "test_bucket" is present with data '{"field1": "value1"}' of type "Binary" in Couchbase
+
+ Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor
+ Given a Couchbase server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
+ And the "success" relationship of the GetCouchbaseKey processor is connected to the PutFile
+ And the "success" relationship of the PutFile processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then a flowfile with the JSON content '{"field1": "value1", "field2": "value2"}' is placed in the monitored directory in less than 60 seconds
+ And the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 10 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
+
+ Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor using binary storage
+ Given a Couchbase server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the PutCouchbaseKey processor is set to "Binary"
+ And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the GetCouchbaseKey processor is set to "Binary"
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
+ And the "success" relationship of the GetCouchbaseKey processor is connected to the PutFile
+ And the "success" relationship of the PutFile processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then a flowfile with the JSON content '{"field1": "value1", "field2": "value2"}' is placed in the monitored directory in less than 60 seconds
+ And the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 10 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
+
+ Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor and put the result in an attribute
+ Given a Couchbase server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the PutCouchbaseKey processor is set to "String"
+ And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the GetCouchbaseKey processor is set to "String"
+ And the "Put Value to Attribute" property of the GetCouchbaseKey processor is set to "get_couchbase_result"
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
+ And the "success" relationship of the GetCouchbaseKey processor is connected to the PutFile
+ And the "success" relationship of the PutFile processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then a flowfile with the JSON content '{"field1": "value1", "field2": "value2"}' is placed in the monitored directory in less than 60 seconds
+ And the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 10 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
+ And the Minifi logs contain the following message: 'key:get_couchbase_result value:{"field1": "value1", "field2": "value2"}' in less than 1 seconds
+
+ Scenario: GetCouchbaseKey transfers FlowFile to failure relationship on Couchbase value type mismatch
+ Given a Couchbase server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the PutCouchbaseKey processor is set to "String"
+ And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And the "Document Type" property of the GetCouchbaseKey processor is set to "Binary"
+ And a CouchbaseClusterService is setup up for PutCouchbaseKey with the name "CouchbaseClusterService"
+ And a CouchbaseClusterService is setup up for GetCouchbaseKey with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then the Minifi logs contain the following message: "Failed to get content for document 'test_doc_id' from collection 'test_bucket._default._default' with the following exception: 'raw_binary_transcoder expects document to have BINARY common flags" in less than 60 seconds
diff --git a/docker/test/integration/minifi/processors/PutCouchbaseKey.py b/docker/test/integration/minifi/processors/GetCouchbaseKey.py
similarity index 84%
copy from docker/test/integration/minifi/processors/PutCouchbaseKey.py
copy to docker/test/integration/minifi/processors/GetCouchbaseKey.py
index 5e94aaa07..0b48dcbdd 100644
--- a/docker/test/integration/minifi/processors/PutCouchbaseKey.py
+++ b/docker/test/integration/minifi/processors/GetCouchbaseKey.py
@@ -15,10 +15,10 @@
from ..core.Processor import Processor
-class PutCouchbaseKey(Processor):
+class GetCouchbaseKey(Processor):
def __init__(self, context, schedule={'scheduling strategy': 'EVENT_DRIVEN'}):
- super(PutCouchbaseKey, self).__init__(
+ super(GetCouchbaseKey, self).__init__(
context=context,
- clazz='PutCouchbaseKey',
- auto_terminate=['success', 'failure'],
+ clazz='GetCouchbaseKey',
+ auto_terminate=['success', 'failure', 'retry'],
schedule=schedule)
diff --git a/docker/test/integration/minifi/processors/PutCouchbaseKey.py b/docker/test/integration/minifi/processors/PutCouchbaseKey.py
index 5e94aaa07..341338771 100644
--- a/docker/test/integration/minifi/processors/PutCouchbaseKey.py
+++ b/docker/test/integration/minifi/processors/PutCouchbaseKey.py
@@ -20,5 +20,5 @@ class PutCouchbaseKey(Processor):
super(PutCouchbaseKey, self).__init__(
context=context,
clazz='PutCouchbaseKey',
- auto_terminate=['success', 'failure'],
+ auto_terminate=['success', 'failure', 'retry'],
schedule=schedule)
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index 8abf7985c..bcab7b90f 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -22,6 +22,7 @@
#include "couchbase/codec/raw_json_transcoder.hxx"
#include "core/Resource.h"
+#include "utils/TimeUtil.h"
namespace org::apache::nifi::minifi::couchbase {
@@ -70,16 +71,57 @@ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::ups
document_id, collection.bucket_name, collection.scope_name, collection.collection_name, upsert_err.ec(), upsert_err.message());
return nonstd::make_unexpected(CouchbaseErrorType::FATAL);
} else {
- const uint64_t partition_uuid = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_uuid() : 0);
- const uint64_t sequence_number = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->sequence_number() : 0);
- const uint16_t partition_id = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_id() : 0);
- return CouchbaseUpsertResult {
- collection.bucket_name,
- upsert_resp.cas().value(),
- partition_uuid,
- sequence_number,
- partition_id
- };
+ CouchbaseUpsertResult result;
+ result.bucket_name = collection.bucket_name;
+ result.cas = upsert_resp.cas().value();
+ result.partition_uuid = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_uuid() : 0);
+ result.sequence_number = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->sequence_number() : 0);
+ result.partition_id = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_id() : 0);
+ return result;
+ }
+}
+
+nonstd::expected<CouchbaseGetResult, CouchbaseErrorType> CouchbaseClient::get(const CouchbaseCollection& collection,
+ const std::string& document_id, CouchbaseValueType return_type) {
+ auto collection_result = getCollection(collection);
+ if (!collection_result.has_value()) {
+ return nonstd::make_unexpected(collection_result.error());
+ }
+
+ ::couchbase::get_options options;
+ options.with_expiry(true);
+ auto [get_err, resp] = collection_result->get(document_id, options).get();
+ if (get_err.ec()) {
+ if (get_err.ec().value() == static_cast<int>(::couchbase::errc::common::unambiguous_timeout) || get_err.ec().value() == static_cast<int>(::couchbase::errc::common::ambiguous_timeout)) {
+ logger_->log_error("Failed to get document '{}' from collection '{}.{}.{}' due to timeout",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
+ return nonstd::make_unexpected(CouchbaseErrorType::TEMPORARY);
+ }
+ std::string cause = get_err.cause() ? get_err.cause()->message() : "";
+ logger_->log_error("Failed to get document '{}' from collection '{}.{}.{}' with error code: '{}', message: '{}'",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name, get_err.ec(), get_err.message());
+ return nonstd::make_unexpected(CouchbaseErrorType::FATAL);
+ } else {
+ try {
+ CouchbaseGetResult result;
+ result.bucket_name = collection.bucket_name;
+ result.cas = resp.cas().value();
+ if (return_type == CouchbaseValueType::Json) {
+ result.value = resp.content_as<::couchbase::codec::binary, ::couchbase::codec::raw_json_transcoder>();
+ } else if (return_type == CouchbaseValueType::String) {
+ result.value = resp.content_as<::couchbase::codec::raw_string_transcoder>();
+ } else {
+ result.value = resp.content_as<::couchbase::codec::raw_binary_transcoder>();
+ }
+ if (resp.expiry_time().has_value()) {
+ result.expiry = utils::timeutils::getTimeStr(*resp.expiry_time());
+ }
+ return result;
+ } catch(const std::exception& ex) {
+ logger_->log_error("Failed to get content for document '{}' from collection '{}.{}.{}' with the following exception: '{}'",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name, ex.what());
+ return nonstd::make_unexpected(CouchbaseErrorType::FATAL);
+ }
}
}
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 0103404c2..1a065d518 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -21,6 +21,7 @@
#include <memory>
#include <string>
#include <utility>
+#include <variant>
#include "core/controller/ControllerService.h"
#include "core/PropertyDefinition.h"
@@ -38,9 +39,17 @@ struct CouchbaseCollection {
std::string collection_name;
};
-struct CouchbaseUpsertResult {
+struct CouchbaseCallResult {
std::string bucket_name;
std::uint64_t cas{0};
+};
+
+struct CouchbaseGetResult : public CouchbaseCallResult {
+ std::string expiry;
+ std::variant<std::vector<std::byte>, std::string> value;
+};
+
+struct CouchbaseUpsertResult : public CouchbaseCallResult {
std::uint64_t sequence_number{0};
std::uint64_t partition_uuid{0};
std::uint16_t partition_id{0};
@@ -65,6 +74,7 @@ class CouchbaseClient {
nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> upsert(const CouchbaseCollection& collection, CouchbaseValueType document_type, const std::string& document_id,
const std::vector<std::byte>& buffer, const ::couchbase::upsert_options& options);
+ nonstd::expected<CouchbaseGetResult, CouchbaseErrorType> get(const CouchbaseCollection& collection, const std::string& document_id, CouchbaseValueType return_type);
nonstd::expected<void, CouchbaseErrorType> establishConnection();
void close();
@@ -152,6 +162,11 @@ class CouchbaseClusterService : public core::controller::ControllerService {
return client_->upsert(collection, document_type, document_id, buffer, options);
}
+ virtual nonstd::expected<CouchbaseGetResult, CouchbaseErrorType> get(const CouchbaseCollection& collection, const std::string& document_id, CouchbaseValueType return_type) {
+ gsl_Expects(client_);
+ return client_->get(collection, document_id, return_type);
+ }
+
static gsl::not_null<std::shared_ptr<CouchbaseClusterService>> getFromProperty(const core::ProcessContext& context, const core::PropertyReference& property);
private:
diff --git a/extensions/couchbase/processors/GetCouchbaseKey.cpp b/extensions/couchbase/processors/GetCouchbaseKey.cpp
new file mode 100644
index 000000000..89cf160f6
--- /dev/null
+++ b/extensions/couchbase/processors/GetCouchbaseKey.cpp
@@ -0,0 +1,111 @@
+/**
+ *
+ * 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.
+ */
+
+#include "GetCouchbaseKey.h"
+#include "utils/gsl.h"
+#include "core/Resource.h"
+#include "utils/ProcessorConfigUtils.h"
+
+namespace org::apache::nifi::minifi::couchbase::processors {
+
+void GetCouchbaseKey::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
+ couchbase_cluster_service_ = controllers::CouchbaseClusterService::getFromProperty(context, GetCouchbaseKey::CouchbaseClusterControllerService);
+ document_type_ = utils::parseEnumProperty<CouchbaseValueType>(context, GetCouchbaseKey::DocumentType);
+}
+
+void GetCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
+ gsl_Expects(couchbase_cluster_service_);
+
+ auto flow_file = session.get();
+ if (!flow_file) {
+ context.yield();
+ return;
+ }
+
+ CouchbaseCollection collection;
+ if (!context.getProperty(BucketName, collection.bucket_name, flow_file.get()) || collection.bucket_name.empty()) {
+ logger_->log_error("Bucket '{}' is invalid or empty!", collection.bucket_name);
+ session.transfer(flow_file, Failure);
+ return;
+ }
+
+ if (!context.getProperty(ScopeName, collection.scope_name, flow_file.get()) || collection.scope_name.empty()) {
+ collection.scope_name = ::couchbase::scope::default_name;
+ }
+
+ if (!context.getProperty(CollectionName, collection.collection_name, flow_file.get()) || collection.collection_name.empty()) {
+ collection.collection_name = ::couchbase::collection::default_name;
+ }
+
+ std::string document_id;
+ if (!context.getProperty(DocumentId, document_id, flow_file.get()) || document_id.empty()) {
+ auto ff_content = session.readBuffer(flow_file).buffer;
+ document_id = std::string(reinterpret_cast<const char*>(ff_content.data()), ff_content.size());
+ }
+
+ if (document_id.empty()) {
+ logger_->log_error("Document ID is empty, transferring FlowFile to failure relationship");
+ session.transfer(flow_file, Failure);
+ return;
+ }
+
+ std::string attribute_to_put_result_to;
+ context.getProperty(PutValueToAttribute, attribute_to_put_result_to, flow_file.get());
+
+ if (auto get_result = couchbase_cluster_service_->get(collection, document_id, document_type_)) {
+ if (!attribute_to_put_result_to.empty()) {
+ if (document_type_ == CouchbaseValueType::String) {
+ session.putAttribute(*flow_file, attribute_to_put_result_to, std::get<std::string>(get_result->value));
+ } else {
+ auto& binary_data = std::get<std::vector<std::byte>>(get_result->value);
+ std::string str_value{reinterpret_cast<const char*>(binary_data.data()), binary_data.size()};
+ session.putAttribute(*flow_file, attribute_to_put_result_to, str_value);
+ }
+ } else {
+ session.write(flow_file, [&, this](const std::shared_ptr<io::OutputStream>& stream) -> int64_t {
+ if (document_type_ == CouchbaseValueType::String) {
+ auto& value = std::get<std::string>(get_result->value);
+ stream->write(value);
+ return gsl::narrow<int64_t>(value.size());
+ } else {
+ auto& value = std::get<std::vector<std::byte>>(get_result->value);
+ stream->write(value);
+ return gsl::narrow<int64_t>(value.size());
+ }
+ });
+ }
+
+ session.putAttribute(*flow_file, "couchbase.bucket", get_result->bucket_name);
+ session.putAttribute(*flow_file, "couchbase.doc.id", document_id);
+ session.putAttribute(*flow_file, "couchbase.doc.cas", std::to_string(get_result->cas));
+ session.putAttribute(*flow_file, "couchbase.doc.expiry", get_result->expiry);
+ session.transfer(flow_file, Success);
+ } else if (get_result.error() == CouchbaseErrorType::TEMPORARY) {
+ logger_->log_error("Failed to get document '{}' from collection '{}.{}.{}' due to timeout, transferring to retry relationship",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
+ session.transfer(flow_file, Retry);
+ } else {
+ logger_->log_error("Failed to get document '{}' from collection '{}.{}.{}', transferring to failure relationship",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
+ session.transfer(flow_file, Failure);
+ }
+}
+
+REGISTER_RESOURCE(GetCouchbaseKey, Processor);
+
+} // namespace org::apache::nifi::minifi::couchbase::processors
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.h b/extensions/couchbase/processors/GetCouchbaseKey.h
similarity index 56%
copy from extensions/couchbase/processors/PutCouchbaseKey.h
copy to extensions/couchbase/processors/GetCouchbaseKey.h
index 56ba8c52f..c286561e4 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.h
+++ b/extensions/couchbase/processors/GetCouchbaseKey.h
@@ -23,56 +23,18 @@
#include "core/AbstractProcessor.h"
#include "core/ProcessSession.h"
-#include "utils/Enum.h"
#include "core/logging/LoggerConfiguration.h"
#include "CouchbaseClusterService.h"
-#include "couchbase/persist_to.hxx"
-#include "couchbase/replicate_to.hxx"
-
-namespace magic_enum::customize {
-
-template <>
-constexpr customize_t enum_name<::couchbase::persist_to>(::couchbase::persist_to value) noexcept {
- switch (value) {
- case ::couchbase::persist_to::none:
- return "NONE";
- case ::couchbase::persist_to::active:
- return "ACTIVE";
- case ::couchbase::persist_to::one:
- return "ONE";
- case ::couchbase::persist_to::two:
- return "TWO";
- case ::couchbase::persist_to::three:
- return "THREE";
- case ::couchbase::persist_to::four:
- return "FOUR";
- }
- return invalid_tag;
-}
-
-template <>
-constexpr customize_t enum_name<::couchbase::replicate_to>(::couchbase::replicate_to value) noexcept {
- switch (value) {
- case ::couchbase::replicate_to::none:
- return "NONE";
- case ::couchbase::replicate_to::one:
- return "ONE";
- case ::couchbase::replicate_to::two:
- return "TWO";
- case ::couchbase::replicate_to::three:
- return "THREE";
- }
- return invalid_tag;
-}
-} // namespace magic_enum::customize
namespace org::apache::nifi::minifi::couchbase::processors {
-class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
+class GetCouchbaseKey final : public core::AbstractProcessor<GetCouchbaseKey> {
public:
- using core::AbstractProcessor<PutCouchbaseKey>::AbstractProcessor;
+ using core::AbstractProcessor<GetCouchbaseKey>::AbstractProcessor;
- EXTENSIONAPI static constexpr const char* Description = "Put a document to Couchbase Server via Key/Value access.";
+ EXTENSIONAPI static constexpr const char* Description = "Get a document from Couchbase Server via Key/Value access. The ID of the document to fetch may be supplied by setting the "
+ "<Document Id> property. NOTE: if the Document Id property is not set, the contents of the FlowFile will be read to determine the Document Id, which means that the contents of "
+ "the entire FlowFile will be buffered in memory.";
EXTENSIONAPI static constexpr auto CouchbaseClusterControllerService = core::PropertyDefinitionBuilder<>::createProperty("Couchbase Cluster Controller Service")
.withDescription("A Couchbase Cluster Controller Service which manages connections to a Couchbase cluster.")
@@ -94,27 +56,19 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
.supportsExpressionLanguage(true)
.build();
EXTENSIONAPI static constexpr auto DocumentType = core::PropertyDefinitionBuilder<3>::createProperty("Document Type")
- .withDescription("Content type to store data as.")
+ .withDescription("Content type of the retrieved value.")
.isRequired(true)
.withDefaultValue(magic_enum::enum_name(CouchbaseValueType::Json))
.withAllowedValues(magic_enum::enum_names<CouchbaseValueType>())
.build();
EXTENSIONAPI static constexpr auto DocumentId = core::PropertyDefinitionBuilder<>::createProperty("Document Id")
- .withDescription("A static, fixed Couchbase document id, or an expression to construct the Couchbase document id. "
- "If not specified, either the FlowFile uuid attribute or if that's not found a generated uuid will be used.")
+ .withDescription("A static, fixed Couchbase document id, or an expression to construct the Couchbase document id.")
.supportsExpressionLanguage(true)
.build();
- EXTENSIONAPI static constexpr auto PersistTo = core::PropertyDefinitionBuilder<6>::createProperty("Persist To")
- .withDescription("Durability constraint about disk persistence.")
- .isRequired(true)
- .withDefaultValue(magic_enum::enum_name(::couchbase::persist_to::none))
- .withAllowedValues(magic_enum::enum_names<::couchbase::persist_to>())
- .build();
- EXTENSIONAPI static constexpr auto ReplicateTo = core::PropertyDefinitionBuilder<4>::createProperty("Replicate To")
- .withDescription("Durability constraint about replication.")
- .isRequired(true)
- .withDefaultValue(magic_enum::enum_name(::couchbase::replicate_to::none))
- .withAllowedValues(magic_enum::enum_names<::couchbase::replicate_to>())
+ EXTENSIONAPI static constexpr auto PutValueToAttribute = core::PropertyDefinitionBuilder<>::createProperty("Put Value to Attribute")
+ .withDescription("If set, the retrieved value will be put into an attribute of the FlowFile instead of a the content of the FlowFile. "
+ "The attribute key to put to is determined by evaluating value of this property.")
+ .supportsExpressionLanguage(true)
.build();
EXTENSIONAPI static constexpr auto Properties = std::to_array<core::PropertyReference>({
@@ -124,23 +78,23 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
CollectionName,
DocumentType,
DocumentId,
- PersistTo,
- ReplicateTo
+ PutValueToAttribute
});
- EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "All FlowFiles that are written to Couchbase Server are routed to this relationship."};
- EXTENSIONAPI static constexpr auto Failure = core::RelationshipDefinition{"failure", "All FlowFiles failed to be written to Couchbase Server and not retry-able are routed to this relationship."};
- EXTENSIONAPI static constexpr auto Retry = core::RelationshipDefinition{"retry", "All FlowFiles failed to be written to Couchbase Server but can be retried are routed to this relationship."};
- EXTENSIONAPI static constexpr auto Relationships = std::array{Success, Failure, Retry};
+ EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success",
+ "Values retrieved from Couchbase Server are written as outgoing FlowFiles content or put into an attribute of the incoming FlowFile and routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Failure = core::RelationshipDefinition{"failure", "All FlowFiles failed to fetch from Couchbase Server and not retry-able are routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Retry = core::RelationshipDefinition{"retry", "All FlowFiles failed to fetch from Couchbase Server but can be retried are routed to this relationship."};
+ EXTENSIONAPI static constexpr auto Original = core::RelationshipDefinition{"original",
+ "The original input FlowFile is routed to this relationship when the value is retrieved from Couchbase Server and routed to 'success'."};
+ EXTENSIONAPI static constexpr auto Relationships = std::array{Success, Failure, Retry, Original};
EXTENSIONAPI static constexpr auto CouchbaseBucket = core::OutputAttributeDefinition<>{"couchbase.bucket", {Success}, "Bucket where the document was stored."};
EXTENSIONAPI static constexpr auto CouchbaseDocId = core::OutputAttributeDefinition<>{"couchbase.doc.id", {Success}, "Id of the document."};
EXTENSIONAPI static constexpr auto CouchbaseDocCas = core::OutputAttributeDefinition<>{"couchbase.doc.cas", {Success}, "CAS of the document."};
- EXTENSIONAPI static constexpr auto CouchbaseDocSequenceNumber = core::OutputAttributeDefinition<>{"couchbase.doc.sequence.number", {Success}, "Sequence number associated with the document."};
- EXTENSIONAPI static constexpr auto CouchbasePartitionUUID = core::OutputAttributeDefinition<>{"couchbase.partition.uuid", {Success}, "UUID of partition."};
- EXTENSIONAPI static constexpr auto CouchbasePartitionId = core::OutputAttributeDefinition<>{"couchbase.partition.id", {Success}, "ID of partition (also known as vBucket)."};
- EXTENSIONAPI static constexpr auto OutputAttributes = std::array<core::OutputAttributeReference, 6>{
- CouchbaseBucket, CouchbaseDocId, CouchbaseDocCas, CouchbaseDocSequenceNumber, CouchbasePartitionUUID, CouchbasePartitionId};
+ EXTENSIONAPI static constexpr auto CouchbaseDocExpiry = core::OutputAttributeDefinition<>{"couchbase.doc.expiry", {Success}, "Expiration of the document."};
+ EXTENSIONAPI static constexpr auto OutputAttributes = std::array<core::OutputAttributeReference, 4>{
+ CouchbaseBucket, CouchbaseDocId, CouchbaseDocCas, CouchbaseDocExpiry};
EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
@@ -152,10 +106,7 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
private:
std::shared_ptr<controllers::CouchbaseClusterService> couchbase_cluster_service_;
- std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<PutCouchbaseKey>::getLogger(uuid_);
CouchbaseValueType document_type_ = CouchbaseValueType::Json;
- ::couchbase::persist_to persist_to_ = ::couchbase::persist_to::none;
- ::couchbase::replicate_to replicate_to_ = ::couchbase::replicate_to::none;
};
} // namespace org::apache::nifi::minifi::couchbase::processors
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.h b/extensions/couchbase/processors/PutCouchbaseKey.h
index 56ba8c52f..1472e3a20 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.h
+++ b/extensions/couchbase/processors/PutCouchbaseKey.h
@@ -152,7 +152,6 @@ class PutCouchbaseKey final : public core::AbstractProcessor<PutCouchbaseKey> {
private:
std::shared_ptr<controllers::CouchbaseClusterService> couchbase_cluster_service_;
- std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<PutCouchbaseKey>::getLogger(uuid_);
CouchbaseValueType document_type_ = CouchbaseValueType::Json;
::couchbase::persist_to persist_to_ = ::couchbase::persist_to::none;
::couchbase::replicate_to replicate_to_ = ::couchbase::replicate_to::none;
diff --git a/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp b/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp
new file mode 100644
index 000000000..69f84aa06
--- /dev/null
+++ b/extensions/couchbase/tests/GetCouchbaseKeyTests.cpp
@@ -0,0 +1,163 @@
+/**
+ *
+ * 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.
+ */
+
+#include "unit/TestBase.h"
+#include "unit/Catch.h"
+#include "unit/SingleProcessorTestController.h"
+#include "processors/GetCouchbaseKey.h"
+#include "MockCouchbaseClusterService.h"
+#include "core/Resource.h"
+
+namespace org::apache::nifi::minifi::couchbase::test {
+
+REGISTER_RESOURCE(MockCouchbaseClusterService, ControllerService);
+
+struct ExpectedCallOptions {
+ std::string bucket_name;
+ std::string scope_name;
+ std::string collection_name;
+ std::string doc_id;
+ couchbase::CouchbaseValueType document_type;
+};
+
+class GetCouchbaseKeyTestController : public TestController {
+ public:
+ GetCouchbaseKeyTestController() {
+ auto controller_service_node = controller_.plan->addController("MockCouchbaseClusterService", "MockCouchbaseClusterService");
+ mock_couchbase_cluster_service_ = std::static_pointer_cast<MockCouchbaseClusterService>(controller_service_node->getControllerServiceImplementation());
+ proc_->setProperty(processors::GetCouchbaseKey::CouchbaseClusterControllerService, "MockCouchbaseClusterService");
+ }
+
+ void verifyResults(const minifi::test::ProcessorTriggerResult& results, const minifi::core::Relationship& expected_result, const ExpectedCallOptions& expected_call_options,
+ const std::string& input) const {
+ std::shared_ptr<core::FlowFile> flow_file;
+ if (expected_result == processors::GetCouchbaseKey::Success) {
+ REQUIRE(results.at(processors::GetCouchbaseKey::Success).size() == 1);
+ REQUIRE(results.at(processors::GetCouchbaseKey::Failure).empty());
+ REQUIRE(results.at(processors::GetCouchbaseKey::Retry).empty());
+ flow_file = results.at(processors::GetCouchbaseKey::Success)[0];
+ } else if (expected_result == processors::GetCouchbaseKey::Failure) {
+ REQUIRE(results.at(processors::GetCouchbaseKey::Success).empty());
+ REQUIRE(results.at(processors::GetCouchbaseKey::Failure).size() == 1);
+ REQUIRE(results.at(processors::GetCouchbaseKey::Retry).empty());
+ flow_file = results.at(processors::GetCouchbaseKey::Failure)[0];
+ REQUIRE(LogTestController::getInstance().contains("Failed to get document", 1s));
+ } else {
+ REQUIRE(results.at(processors::GetCouchbaseKey::Success).empty());
+ REQUIRE(results.at(processors::GetCouchbaseKey::Failure).empty());
+ REQUIRE(results.at(processors::GetCouchbaseKey::Retry).size() == 1);
+ flow_file = results.at(processors::GetCouchbaseKey::Retry)[0];
+ }
+
+ auto get_collection_parameters = mock_couchbase_cluster_service_->getCollectionParameter();
+ CHECK(get_collection_parameters.bucket_name == expected_call_options.bucket_name);
+ CHECK(get_collection_parameters.collection_name == expected_call_options.collection_name);
+ CHECK(get_collection_parameters.scope_name == expected_call_options.scope_name);
+
+ auto get_parameters = mock_couchbase_cluster_service_->getGetParameters();
+ CHECK(get_parameters.document_id == expected_call_options.doc_id);
+ CHECK(get_parameters.document_type == expected_call_options.document_type);
+
+ if (expected_result != processors::GetCouchbaseKey::Success) {
+ return;
+ }
+
+ CHECK(flow_file->getAttribute("couchbase.bucket").value() == expected_call_options.bucket_name);
+ CHECK(flow_file->getAttribute("couchbase.doc.id").value() == expected_call_options.doc_id);
+ CHECK(flow_file->getAttribute("couchbase.doc.cas").value() == std::to_string(COUCHBASE_GET_RESULT_CAS));
+ CHECK(flow_file->getAttribute("couchbase.doc.expiry").value() == COUCHBASE_GET_RESULT_EXPIRY);
+ std::string value;
+ proc_->getProperty(processors::GetCouchbaseKey::PutValueToAttribute, value);
+ if (!value.empty()) {
+ CHECK(flow_file->getAttribute(value).value() == COUCHBASE_GET_RESULT_CONTENT);
+ CHECK(controller_.plan->getContent(flow_file) == input);
+ } else {
+ CHECK(controller_.plan->getContent(flow_file) == COUCHBASE_GET_RESULT_CONTENT);
+ }
+ }
+
+ protected:
+ std::shared_ptr<core::Processor> proc_ = std::make_shared<processors::GetCouchbaseKey>("GetCouchbaseKey");
+ minifi::test::SingleProcessorTestController controller_{proc_};
+ std::shared_ptr<MockCouchbaseClusterService> mock_couchbase_cluster_service_;
+};
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Invalid Couchbase cluster controller service", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::CouchbaseClusterControllerService, "invalid");
+ REQUIRE_THROWS_AS(controller_.trigger({minifi::test::InputFlowFileData{"couchbase_id"}}), minifi::Exception);
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Invalid bucket name", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "");
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{"couchbase_id"}});
+ REQUIRE(results[processors::GetCouchbaseKey::Failure].size() == 1);
+ REQUIRE(LogTestController::getInstance().contains("Bucket '' is invalid or empty!", 1s));
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Document ID is empty and no content is present to use", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "mybucket");
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{""}});
+ REQUIRE(results.at(processors::GetCouchbaseKey::Success).empty());
+ REQUIRE(results.at(processors::GetCouchbaseKey::Failure).size() == 1);
+ REQUIRE(results.at(processors::GetCouchbaseKey::Retry).empty());
+ REQUIRE(LogTestController::getInstance().contains("Document ID is empty, transferring FlowFile to failure relationship", 1s));
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Get succeeeds with default properties", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "mybucket");
+ const std::string input = "couchbase_id";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::GetCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "_default", "_default", input, couchbase::CouchbaseValueType::Json}, input);
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Get succeeeds with optional properties", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "mybucket");
+ proc_->setProperty(processors::GetCouchbaseKey::ScopeName, "scope1");
+ proc_->setProperty(processors::GetCouchbaseKey::CollectionName, "collection1");
+ proc_->setProperty(processors::GetCouchbaseKey::DocumentId, "important_doc");
+ proc_->setProperty(processors::GetCouchbaseKey::DocumentType, "Binary");
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{""}});
+ verifyResults(results, processors::GetCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "scope1", "collection1", "important_doc", couchbase::CouchbaseValueType::Binary}, "");
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Get fails with default properties", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "mybucket");
+ mock_couchbase_cluster_service_->setGetError(CouchbaseErrorType::FATAL);
+ const std::string input = "couchbase_id";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::GetCouchbaseKey::Failure, ExpectedCallOptions{"mybucket", "_default", "_default", input, couchbase::CouchbaseValueType::Json}, input);
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "FlowFile is transferred to retry relationship when temporary error is returned", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "mybucket");
+ mock_couchbase_cluster_service_->setGetError(CouchbaseErrorType::TEMPORARY);
+ const std::string input = "couchbase_id";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::GetCouchbaseKey::Retry, ExpectedCallOptions{"mybucket", "_default", "_default", input, couchbase::CouchbaseValueType::Json}, input);
+}
+
+TEST_CASE_METHOD(GetCouchbaseKeyTestController, "Get result is written to attribute", "[getcouchbasekey]") {
+ proc_->setProperty(processors::GetCouchbaseKey::BucketName, "mybucket");
+ proc_->setProperty(processors::GetCouchbaseKey::DocumentType, "String");
+ proc_->setProperty(processors::GetCouchbaseKey::PutValueToAttribute, "myattribute");
+ const std::string input = "couchbase_id";
+ auto results = controller_.trigger({minifi::test::InputFlowFileData{input}});
+ verifyResults(results, processors::GetCouchbaseKey::Success, ExpectedCallOptions{"mybucket", "_default", "_default", input, couchbase::CouchbaseValueType::String}, input);
+}
+
+} // namespace org::apache::nifi::minifi::couchbase::test
diff --git a/extensions/couchbase/tests/MockCouchbaseClusterService.h b/extensions/couchbase/tests/MockCouchbaseClusterService.h
index a2757f16c..6566e85d4 100644
--- a/extensions/couchbase/tests/MockCouchbaseClusterService.h
+++ b/extensions/couchbase/tests/MockCouchbaseClusterService.h
@@ -29,6 +29,9 @@ const std::uint64_t COUCHBASE_PUT_RESULT_CAS = 9876;
const std::uint64_t COUCHBASE_PUT_RESULT_SEQUENCE_NUMBER = 345;
const std::uint64_t COUCHBASE_PUT_RESULT_PARTITION_UUID = 7890123456;
const std::uint16_t COUCHBASE_PUT_RESULT_PARTITION_ID = 1234;
+const std::string COUCHBASE_GET_RESULT_EXPIRY = "2024/10/14 09:37:43.000Z";
+const std::string COUCHBASE_GET_RESULT_CONTENT = "abc";
+const uint64_t COUCHBASE_GET_RESULT_CAS = 1234567;
struct UpsertParameters {
CouchbaseValueType document_type;
@@ -36,6 +39,10 @@ struct UpsertParameters {
std::vector<std::byte> buffer;
::couchbase::upsert_options options;
};
+struct GetParameters {
+ std::string document_id;
+ CouchbaseValueType document_type;
+};
class MockCouchbaseClusterService : public controllers::CouchbaseClusterService {
public:
@@ -57,7 +64,23 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
if (upsert_error_) {
return nonstd::make_unexpected(*upsert_error_);
} else {
- return CouchbaseUpsertResult{collection_.bucket_name, COUCHBASE_PUT_RESULT_CAS, COUCHBASE_PUT_RESULT_SEQUENCE_NUMBER, COUCHBASE_PUT_RESULT_PARTITION_UUID, COUCHBASE_PUT_RESULT_PARTITION_ID};
+ return CouchbaseUpsertResult{{collection_.bucket_name, COUCHBASE_PUT_RESULT_CAS}, COUCHBASE_PUT_RESULT_SEQUENCE_NUMBER, COUCHBASE_PUT_RESULT_PARTITION_UUID, COUCHBASE_PUT_RESULT_PARTITION_ID};
+ }
+ }
+
+ nonstd::expected<CouchbaseGetResult, CouchbaseErrorType> get(const CouchbaseCollection& collection, const std::string& document_id, CouchbaseValueType document_type) override {
+ collection_ = collection;
+ get_parameters_.document_id = document_id;
+ get_parameters_.document_type = document_type;
+
+ if (get_error_) {
+ return nonstd::make_unexpected(*get_error_);
+ } else {
+ if (document_type == CouchbaseValueType::String) {
+ return CouchbaseGetResult{{collection_.bucket_name, COUCHBASE_GET_RESULT_CAS}, COUCHBASE_GET_RESULT_EXPIRY, COUCHBASE_GET_RESULT_CONTENT};
+ }
+ return CouchbaseGetResult{{collection_.bucket_name, COUCHBASE_GET_RESULT_CAS}, COUCHBASE_GET_RESULT_EXPIRY,
+ std::vector<std::byte>{static_cast<std::byte>('a'), static_cast<std::byte>('b'), static_cast<std::byte>('c')}};
}
}
@@ -65,6 +88,10 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
return upsert_parameters_;
}
+ GetParameters getGetParameters() const {
+ return get_parameters_;
+ }
+
CouchbaseCollection getCollectionParameter() const {
return collection_;
}
@@ -73,9 +100,15 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
upsert_error_ = upsert_error;
}
+ void setGetError(const CouchbaseErrorType get_error) {
+ get_error_ = get_error;
+ }
+
private:
CouchbaseCollection collection_;
UpsertParameters upsert_parameters_;
+ GetParameters get_parameters_;
std::optional<CouchbaseErrorType> upsert_error_;
+ std::optional<CouchbaseErrorType> get_error_;
};
} // namespace org::apache::nifi::minifi::couchbase::test
diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h
index f720733e8..6f25c176c 100644
--- a/libminifi/include/core/Processor.h
+++ b/libminifi/include/core/Processor.h
@@ -238,6 +238,8 @@ class Processor : public Connectable, public ConfigurableComponent, public state
std::string cron_period_;
gsl::not_null<std::shared_ptr<ProcessorMetrics>> metrics_;
+ std::shared_ptr<logging::Logger> logger_;
+
private:
mutable std::mutex mutex_;
std::atomic<std::chrono::steady_clock::time_point> yield_expiration_{};
@@ -254,8 +256,6 @@ class Processor : public Connectable, public ConfigurableComponent, public state
// an outgoing connection allows us to reach these nodes
std::unordered_map<Connection*, std::unordered_set<Processor*>> reachable_processors_;
-
- std::shared_ptr<logging::Logger> logger_;
};
} // namespace core
(nifi-minifi-cpp) 04/15: Fix clang issue and revise temporary failures
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3c4d115b794ce4b1e90f802c34a1a23e7e5bf801
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Oct 9 17:15:34 2024 +0200
Fix clang issue and revise temporary failures
---
.../couchbase/controllerservices/CouchbaseClusterService.cpp | 12 ++++++------
.../couchbase/controllerservices/CouchbaseClusterService.h | 1 -
extensions/couchbase/processors/PutCouchbaseKey.cpp | 4 ++--
extensions/couchbase/tests/MockCouchbaseClusterService.h | 1 -
4 files changed, 8 insertions(+), 10 deletions(-)
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index 8db837198..fbd8bfeea 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -48,12 +48,12 @@ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::ups
auto [upsert_err, upsert_resp] = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
if (upsert_err.ec()) {
setConnectionError();
- if (upsert_err.ec().value() == static_cast<int>(::couchbase::errc::common::unambiguous_timeout)) {
- logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to timeout",
- document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
+ // ambiguous_timeout should not be retried as we do not know if the insert was successful or not
+ if (getErrorType(upsert_err.ec()) == CouchbaseErrorType::TEMPORARY && upsert_err.ec().value() != static_cast<int>(::couchbase::errc::common::ambiguous_timeout)) {
+ logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to temporary issue, error code: '{}', message: '{}'",
+ document_id, collection.bucket_name, collection.scope_name, collection.collection_name, upsert_err.ec(), upsert_err.message());
return nonstd::make_unexpected(CouchbaseErrorType::TEMPORARY);
}
- std::string cause = upsert_err.cause() ? upsert_err.cause()->message() : "";
logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' with error code: '{}', message: '{}'",
document_id, collection.bucket_name, collection.scope_name, collection.collection_name, upsert_err.ec(), upsert_err.message());
return nonstd::make_unexpected(CouchbaseErrorType::FATAL);
@@ -62,7 +62,7 @@ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::ups
const uint64_t sequence_number = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->sequence_number() : 0);
const uint16_t partition_id = (upsert_resp.mutation_token().has_value() ? upsert_resp.mutation_token()->partition_id() : 0);
return CouchbaseUpsertResult {
- std::string(collection.bucket_name),
+ collection.bucket_name,
upsert_resp.cas().value(),
partition_uuid,
sequence_number,
@@ -94,6 +94,7 @@ std::optional<CouchbaseErrorType> CouchbaseClient::establishConnection() {
auto [err, upsert_resp] = cluster_.ping().get();
if (err.ec()) {
close();
+ state_ = State::DISCONNECTED;
} else {
state_ = State::CONNECTED;
return std::nullopt;
@@ -103,7 +104,6 @@ std::optional<CouchbaseErrorType> CouchbaseClient::establishConnection() {
auto options = ::couchbase::cluster_options(username_, password_);
auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, options).get();
if (connect_err.ec()) {
- std::string cause = connect_err.cause() ? connect_err.cause()->message() : "";
logger_->log_error("Failed to connect to Couchbase cluster with error code: '{}' and message: '{}'", connect_err.ec(), connect_err.message());
return getErrorType(connect_err.ec());
}
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 9b2fa6831..0890e0dcd 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -74,7 +74,6 @@ class CouchbaseClient {
static constexpr std::array<::couchbase::errc::common, 9> temporary_connection_errors = {
::couchbase::errc::common::temporary_failure,
::couchbase::errc::common::request_canceled,
- ::couchbase::errc::common::service_not_available,
::couchbase::errc::common::internal_server_failure,
::couchbase::errc::common::cas_mismatch,
::couchbase::errc::common::ambiguous_timeout,
diff --git a/extensions/couchbase/processors/PutCouchbaseKey.cpp b/extensions/couchbase/processors/PutCouchbaseKey.cpp
index 15771afca..fb36ec4a8 100644
--- a/extensions/couchbase/processors/PutCouchbaseKey.cpp
+++ b/extensions/couchbase/processors/PutCouchbaseKey.cpp
@@ -62,7 +62,7 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
options.durability(persist_to_, replicate_to_);
auto result = session.readBuffer(flow_file);
if (auto upsert_result = couchbase_cluster_service_->upsert(collection, document_id, result.buffer, options)) {
- session.putAttribute(*flow_file, "couchbase.bucket", std::string(upsert_result->bucket_name));
+ session.putAttribute(*flow_file, "couchbase.bucket", upsert_result->bucket_name);
session.putAttribute(*flow_file, "couchbase.doc.id", document_id);
session.putAttribute(*flow_file, "couchbase.doc.cas", std::to_string(upsert_result->cas));
session.putAttribute(*flow_file, "couchbase.doc.sequence.number", std::to_string(upsert_result->sequence_number));
@@ -70,7 +70,7 @@ void PutCouchbaseKey::onTrigger(core::ProcessContext& context, core::ProcessSess
session.putAttribute(*flow_file, "couchbase.partition.id", std::to_string(upsert_result->partition_id));
session.transfer(flow_file, Success);
} else if (upsert_result.error() == CouchbaseErrorType::TEMPORARY) {
- logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to timeout, transferring to retry relationship",
+ logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to temporary issue, transferring to retry relationship",
document_id, collection.bucket_name, collection.scope_name, collection.collection_name);
session.transfer(flow_file, Retry);
} else {
diff --git a/extensions/couchbase/tests/MockCouchbaseClusterService.h b/extensions/couchbase/tests/MockCouchbaseClusterService.h
index 396bff4fd..5f2bc08a6 100644
--- a/extensions/couchbase/tests/MockCouchbaseClusterService.h
+++ b/extensions/couchbase/tests/MockCouchbaseClusterService.h
@@ -70,6 +70,5 @@ class MockCouchbaseClusterService : public controllers::CouchbaseClusterService
CouchbaseCollection collection_;
UpsertParameters upsert_parameters_;
std::optional<CouchbaseErrorType> upsert_error_;
- bool get_collection_succeeds_{true};
};
} // namespace org::apache::nifi::minifi::couchbase::test
(nifi-minifi-cpp) 05/15: Remove connection state management
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 6e74934c5a064d5b3c6fdf07bbe5851314c3090e
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Oct 10 10:00:01 2024 +0200
Remove connection state management
---
.../controllerservices/CouchbaseClusterService.cpp | 29 +++-------------------
.../controllerservices/CouchbaseClusterService.h | 11 +-------
2 files changed, 5 insertions(+), 35 deletions(-)
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index fbd8bfeea..f529a7fc9 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -35,7 +35,7 @@ nonstd::expected<::couchbase::collection, CouchbaseErrorType> CouchbaseClient::g
if (auto error_type = establishConnection()) {
return nonstd::make_unexpected(*error_type);
}
- return cluster_.bucket(collection.bucket_name).scope(collection.scope_name).collection(collection.collection_name);
+ return cluster_->bucket(collection.bucket_name).scope(collection.scope_name).collection(collection.collection_name);
}
nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::upsert(const CouchbaseCollection& collection,
@@ -47,7 +47,6 @@ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::ups
auto [upsert_err, upsert_resp] = collection_result->upsert<::couchbase::codec::raw_binary_transcoder>(document_id, buffer, options).get();
if (upsert_err.ec()) {
- setConnectionError();
// ambiguous_timeout should not be retried as we do not know if the insert was successful or not
if (getErrorType(upsert_err.ec()) == CouchbaseErrorType::TEMPORARY && upsert_err.ec().value() != static_cast<int>(::couchbase::errc::common::ambiguous_timeout)) {
logger_->log_error("Failed to upsert document '{}' to collection '{}.{}.{}' due to temporary issue, error code: '{}', message: '{}'",
@@ -71,36 +70,17 @@ nonstd::expected<CouchbaseUpsertResult, CouchbaseErrorType> CouchbaseClient::ups
}
}
-void CouchbaseClient::setConnectionError() {
- std::lock_guard<std::mutex> lock(state_mutex_);
- state_ = State::UNKNOWN;
-}
-
void CouchbaseClient::close() {
- std::lock_guard<std::mutex> lock(state_mutex_);
- if (state_ == State::CONNECTED || state_ == State::UNKNOWN) {
- cluster_.close().wait();
- state_ = State::DISCONNECTED;
+ if (cluster_) {
+ cluster_->close().wait();
}
}
std::optional<CouchbaseErrorType> CouchbaseClient::establishConnection() {
- std::lock_guard<std::mutex> lock(state_mutex_);
- if (state_ == State::CONNECTED) {
+ if (cluster_) {
return std::nullopt;
}
- if (state_ == State::UNKNOWN) {
- auto [err, upsert_resp] = cluster_.ping().get();
- if (err.ec()) {
- close();
- state_ = State::DISCONNECTED;
- } else {
- state_ = State::CONNECTED;
- return std::nullopt;
- }
- }
-
auto options = ::couchbase::cluster_options(username_, password_);
auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, options).get();
if (connect_err.ec()) {
@@ -108,7 +88,6 @@ std::optional<CouchbaseErrorType> CouchbaseClient::establishConnection() {
return getErrorType(connect_err.ec());
}
cluster_ = std::move(cluster);
- state_ = State::CONNECTED;
return std::nullopt;
}
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 0890e0dcd..6b85c37b8 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -55,12 +55,6 @@ enum class CouchbaseErrorType {
class CouchbaseClient {
public:
- enum class State {
- DISCONNECTED,
- CONNECTED,
- UNKNOWN,
- };
-
CouchbaseClient(std::string connection_string, std::string username, std::string password, const std::shared_ptr<core::logging::Logger>& logger)
: connection_string_(std::move(connection_string)), username_(std::move(username)), password_(std::move(password)), logger_(logger) {
}
@@ -84,14 +78,11 @@ class CouchbaseClient {
static CouchbaseErrorType getErrorType(const std::error_code& error_code);
nonstd::expected<::couchbase::collection, CouchbaseErrorType> getCollection(const CouchbaseCollection& collection);
- void setConnectionError();
- std::mutex state_mutex_;
- State state_ = State::DISCONNECTED;
std::string connection_string_;
std::string username_;
std::string password_;
- ::couchbase::cluster cluster_;
+ std::optional<::couchbase::cluster> cluster_;
std::shared_ptr<core::logging::Logger> logger_;
};
(nifi-minifi-cpp) 03/15: Review update
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch MINIFICPP-2469
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 6c08d961526aa6caad47e2a9c9510269021a79e3
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Oct 9 16:26:07 2024 +0200
Review update
---
docker/test/integration/cluster/containers/CouchbaseServerContainer.py | 3 +--
thirdparty/bustache/add-append.patch | 3 +++
2 files changed, 4 insertions(+), 2 deletions(-)
diff --git a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
index a964690fa..63d2a84c6 100644
--- a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
+++ b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
@@ -21,6 +21,7 @@ class CouchbaseServerContainer(Container):
super().__init__(feature_context, name, 'couchbase-server', vols, network, image_store, command)
def get_startup_finished_log_entry(self):
+ # after startup the logs are only available in the container, only this message is shown
return "logs available in"
@retry_check(15, 2)
@@ -45,11 +46,9 @@ class CouchbaseServerContainer(Container):
if not self.set_deployed():
return
- port_list = [*range(8091, 8098), 9123, 11207, 11210, 11280, *range(18091, 18097)]
self.docker_container = self.client.containers.run(
"couchbase:community-7.6.2",
detach=True,
name=self.name,
network=self.network.name,
- ports={f'{port}/tcp': port for port in port_list},
entrypoint=self.command)
diff --git a/thirdparty/bustache/add-append.patch b/thirdparty/bustache/add-append.patch
index fb7f24149..78878c86d 100644
--- a/thirdparty/bustache/add-append.patch
+++ b/thirdparty/bustache/add-append.patch
@@ -1,3 +1,6 @@
+fmt library 11.0.2 requires the implementation of the append function aside
+from push_back method when formatting the bustache specific container
+
diff --git a/include/bustache/model.hpp b/include/bustache/model.hpp
index 575969a..6df2cff 100644
--- a/include/bustache/model.hpp