You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fg...@apache.org on 2026/01/07 20:41:56 UTC
(nifi-minifi-cpp) 01/07: MINIFICPP-2666 Move Kafka tests to modular docker tests
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 40ff0a929ff3892a4a4642bac8829eb59101cd39
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Jan 7 17:32:48 2026 +0100
MINIFICPP-2666 Move Kafka tests to modular docker tests
Closes #2059
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
---
behave_framework/pyproject.toml | 5 +-
.../minifi_test_framework/containers/container.py | 120 +++++++++-
.../src/minifi_test_framework/containers/file.py | 5 +-
.../containers/minifi_container.py | 21 +-
.../src/minifi_test_framework/core/helpers.py | 13 +-
.../src/minifi_test_framework/core/hooks.py | 2 +
.../core/minifi_test_context.py | 12 +-
.../src/minifi_test_framework/core/ssl_utils.py | 161 +++++++++++++
.../src/minifi_test_framework/minifi/processor.py | 4 +
.../minifi_test_framework/steps/checking_steps.py | 69 +++++-
.../src/minifi_test_framework/steps/core_steps.py | 14 +-
.../steps/flow_building_steps.py | 33 ++-
docker/RunBehaveTests.sh | 3 +-
docker/test/integration/cluster/ContainerStore.py | 9 -
.../test/integration/cluster/DockerTestCluster.py | 10 -
.../integration/cluster/checkers/KafkaHelper.py | 58 -----
.../cluster/containers/KafkaBrokerContainer.py | 163 -------------
.../features/MiNiFi_integration_test_driver.py | 8 -
.../test/integration/features/steps/kafka_steps.py | 188 ---------------
docker/test/integration/features/steps/steps.py | 1 -
.../integration/minifi/processors/ConsumeKafka.py | 37 ---
extensions/aws/tests/features/s3.feature | 22 +-
.../azure/tests/features/azure_storage.feature | 4 +-
.../kafka/tests}/features/consumekafka.feature | 228 +++++++++---------
.../kafka/tests/features/environment.py | 35 +--
.../kafka/tests/features/publishkafka.feature | 259 +++++++++++----------
.../tests/features/steps/kafka_server_container.py | 126 ++++++++++
extensions/kafka/tests/features/steps/steps.py | 214 +++++++++++++++++
.../tests/features/replace_text.feature | 4 +-
.../tests/features/routetext.feature | 2 +-
.../tests/features/splittext.feature | 4 +-
.../tests/features/syslog_listener.feature | 2 +-
32 files changed, 1055 insertions(+), 781 deletions(-)
diff --git a/behave_framework/pyproject.toml b/behave_framework/pyproject.toml
index 7533a205e..f5d2c3174 100644
--- a/behave_framework/pyproject.toml
+++ b/behave_framework/pyproject.toml
@@ -7,7 +7,10 @@ dependencies = [
"behavex==4.6.0",
"docker==7.1.0",
"PyYAML==6.0.3",
- "humanfriendly==10.0"
+ "humanfriendly==10.0",
+ "m2crypto==0.41.0",
+ "pyopenssl==25.0.0",
+ "pyjks==20.0.0"
]
[tool.setuptools]
diff --git a/behave_framework/src/minifi_test_framework/containers/container.py b/behave_framework/src/minifi_test_framework/containers/container.py
index dc0cf7353..e01c248d3 100644
--- a/behave_framework/src/minifi_test_framework/containers/container.py
+++ b/behave_framework/src/minifi_test_framework/containers/container.py
@@ -19,6 +19,7 @@ import logging
import os
import shlex
import tempfile
+import tarfile
import docker
from docker.models.networks import Network
@@ -48,23 +49,34 @@ class Container:
def add_host_file(self, host_path: str, container_path: str, mode: str = "ro"):
self.host_files.append(HostFile(container_path, host_path, mode))
- def deploy(self) -> bool:
- self._temp_dir = tempfile.TemporaryDirectory()
+ def _write_content_to_file(self, filepath: str, permissions: int | None, content: str | bytes):
+ write_mode = "w"
+ if isinstance(content, bytes):
+ write_mode = "wb"
+ with open(filepath, write_mode) as file:
+ file.write(content)
+ if permissions:
+ os.chmod(filepath, permissions)
+ def _configure_volumes_of_container_files(self):
for file in self.files:
- file_name = os.path.basename(file.path)
- temp_path = os.path.join(self._temp_dir.name, file_name)
- with open(temp_path, "w") as temp_file:
- temp_file.write(file.content)
+ temp_path = os.path.join(self._temp_dir.name, os.path.basename(file.path))
+ self._write_content_to_file(temp_path, file.permissions, file.content)
self.volumes[temp_path] = {"bind": file.path, "mode": file.mode}
+
+ def _configure_volumes_of_container_dirs(self):
for directory in self.dirs:
temp_path = self._temp_dir.name + directory.path
os.makedirs(temp_path, exist_ok=True)
for file_name, content in directory.files.items():
file_path = os.path.join(temp_path, file_name)
- with open(file_path, "w") as temp_file:
- temp_file.write(content)
+ self._write_content_to_file(file_path, None, content)
self.volumes[temp_path] = {"bind": directory.path, "mode": directory.mode}
+
+ def deploy(self) -> bool:
+ self._temp_dir = tempfile.TemporaryDirectory()
+ self._configure_volumes_of_container_files()
+ self._configure_volumes_of_container_dirs()
for host_file in self.host_files:
self.volumes[host_file.host_path] = {"bind": host_file.container_path, "mode": host_file.mode}
@@ -85,6 +97,30 @@ class Container:
raise
return True
+ def start(self):
+ if self.container:
+ self.container.start()
+ else:
+ logging.error("Container does not exist. Cannot start.")
+
+ def stop(self):
+ if self.container:
+ self.container.stop()
+ else:
+ logging.error("Container does not exist. Cannot stop.")
+
+ def kill(self):
+ if self.container:
+ self.container.kill()
+ else:
+ logging.error("Container does not exist. Cannot kill.")
+
+ def restart(self):
+ if self.container:
+ self.container.restart()
+ else:
+ logging.error("Container does not exist. Cannot restart.")
+
def clean_up(self):
if self.container:
try:
@@ -203,9 +239,14 @@ class Container:
return False
def get_number_of_files(self, directory_path: str) -> int:
- if not self.container or not self.not_empty_dir_exists(directory_path):
+ if not self.container:
+ logging.warning("Container not running")
return -1
+ if not self.not_empty_dir_exists(directory_path):
+ logging.warning(f"Container directory does not exist: {directory_path}")
+ return 0
+
count_command = f"sh -c 'find {directory_path} -maxdepth 1 -type f | wc -l'"
exit_code, output = self.exec_run(count_command)
@@ -214,13 +255,15 @@ class Container:
return -1
try:
- return int(output.strip())
+ file_count = int(output.strip())
+ logging.debug(f"Number of files in '{directory_path}': {file_count}")
+ return file_count
except (ValueError, IndexError):
logging.error(f"Error parsing output '{output}' from command '{count_command}'")
return -1
- def verify_file_contents(self, directory_path: str, expected_contents: list[str]) -> bool:
- if not self.container or not self.not_empty_dir_exists(directory_path):
+ def _verify_file_contents_in_running_container(self, directory_path: str, expected_contents: list[str]) -> bool:
+ if not self.not_empty_dir_exists(directory_path):
return False
safe_dir_path = shlex.quote(directory_path)
@@ -254,6 +297,59 @@ class Container:
return sorted(actual_file_contents) == sorted(expected_contents)
+ def _verify_file_contents_in_stopped_container(self, directory_path: str, expected_contents: list[str]) -> bool:
+ if not self.container:
+ return False
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ extracted_dir = self._extract_directory_from_container(directory_path, temp_dir)
+ if not extracted_dir:
+ return False
+
+ actual_file_contents = self._read_files_from_directory(extracted_dir)
+ if actual_file_contents is None:
+ return False
+
+ return sorted(actual_file_contents) == sorted(expected_contents)
+
+ def _extract_directory_from_container(self, directory_path: str, temp_dir: str) -> str | None:
+ try:
+ bits, _ = self.container.get_archive(directory_path)
+ temp_tar_path = os.path.join(temp_dir, "archive.tar")
+
+ with open(temp_tar_path, 'wb') as f:
+ for chunk in bits:
+ f.write(chunk)
+
+ with tarfile.open(temp_tar_path) as tar:
+ tar.extractall(path=temp_dir)
+
+ return os.path.join(temp_dir, os.path.basename(directory_path.strip('/')))
+ except Exception as e:
+ logging.error(f"Error extracting files from directory path '{directory_path}' from container '{self.container_name}': {e}")
+ return None
+
+ def _read_files_from_directory(self, directory_path: str) -> list[str] | None:
+ try:
+ file_contents = []
+ for entry in os.scandir(directory_path):
+ if entry.is_file():
+ with open(entry.path, 'r') as f:
+ file_contents.append(f.read())
+ return file_contents
+ except Exception as e:
+ logging.error(f"Error reading extracted files: {e}")
+ return None
+
+ def verify_file_contents(self, directory_path: str, expected_contents: list[str]) -> bool:
+ if not self.container:
+ return False
+
+ if self.container.status == "running":
+ return self._verify_file_contents_in_running_container(directory_path, expected_contents)
+
+ return self._verify_file_contents_in_stopped_container(directory_path, expected_contents)
+
def log_app_output(self) -> bool:
logs = self.get_logs()
logging.info("Logs of container '%s':", self.container_name)
diff --git a/behave_framework/src/minifi_test_framework/containers/file.py b/behave_framework/src/minifi_test_framework/containers/file.py
index 78c9aec7e..ee50af95e 100644
--- a/behave_framework/src/minifi_test_framework/containers/file.py
+++ b/behave_framework/src/minifi_test_framework/containers/file.py
@@ -16,7 +16,8 @@
#
class File:
- def __init__(self, path, content):
+ def __init__(self, path, content: str | bytes, mode="rw", permissions=None):
self.path = path
self.content = content
- self.mode = "rw"
+ self.mode = mode
+ self.permissions = permissions
diff --git a/behave_framework/src/minifi_test_framework/containers/minifi_container.py b/behave_framework/src/minifi_test_framework/containers/minifi_container.py
index b2cfeaeaf..84124ed8c 100644
--- a/behave_framework/src/minifi_test_framework/containers/minifi_container.py
+++ b/behave_framework/src/minifi_test_framework/containers/minifi_container.py
@@ -16,34 +16,43 @@
#
import logging
-from docker.models.networks import Network
+from OpenSSL import crypto
+from minifi_test_framework.core.minifi_test_context import MinifiTestContext
from minifi_test_framework.containers.file import File
from minifi_test_framework.minifi.flow_definition import FlowDefinition
+from minifi_test_framework.core.ssl_utils import make_cert_without_extended_usage
from .container import Container
class MinifiContainer(Container):
- def __init__(self, image_name: str, container_name: str, scenario_id: str, network: Network):
- super().__init__(image_name, f"{container_name}-{scenario_id}", network)
+ def __init__(self, container_name: str, test_context: MinifiTestContext):
+ super().__init__(test_context.minifi_container_image, f"{container_name}-{test_context.scenario_id}", test_context.network)
self.flow_config_str: str = ""
self.flow_definition = FlowDefinition()
self.properties: dict[str, str] = {}
self.log_properties: dict[str, str] = {}
- self.is_fhs = 'MINIFI_INSTALLATION_TYPE=FHS' in str(self.client.images.get(image_name).history())
+ minifi_client_cert, minifi_client_key = make_cert_without_extended_usage(common_name=self.container_name, ca_cert=test_context.root_ca_cert, ca_key=test_context.root_ca_key)
+ self.files.append(File("/tmp/resources/root_ca.crt", crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=test_context.root_ca_cert)))
+ self.files.append(File("/tmp/resources/minifi_client.crt", crypto.dump_certificate(type=crypto.FILETYPE_PEM, cert=minifi_client_cert)))
+ self.files.append(File("/tmp/resources/minifi_client.key", crypto.dump_privatekey(type=crypto.FILETYPE_PEM, pkey=minifi_client_key)))
+
+ self.is_fhs = 'MINIFI_INSTALLATION_TYPE=FHS' in str(self.client.images.get(test_context.minifi_container_image).history())
self._fill_default_properties()
self._fill_default_log_properties()
def deploy(self) -> bool:
+ flow_config = self.flow_definition.to_yaml()
+ logging.info(f"Deploying MiNiFi container '{self.container_name}' with flow configuration:\n{flow_config}")
if self.is_fhs:
- self.files.append(File("/etc/nifi-minifi-cpp/config.yml", self.flow_definition.to_yaml()))
+ self.files.append(File("/etc/nifi-minifi-cpp/config.yml", flow_config))
self.files.append(File("/etc/nifi-minifi-cpp/minifi.properties", self._get_properties_file_content()))
self.files.append(
File("/etc/nifi-minifi-cpp/minifi-log.properties", self._get_log_properties_file_content()))
else:
- self.files.append(File("/opt/minifi/minifi-current/conf/config.yml", self.flow_definition.to_yaml()))
+ self.files.append(File("/opt/minifi/minifi-current/conf/config.yml", flow_config))
self.files.append(
File("/opt/minifi/minifi-current/conf/minifi.properties", self._get_properties_file_content()))
self.files.append(File("/opt/minifi/minifi-current/conf/minifi-log.properties",
diff --git a/behave_framework/src/minifi_test_framework/core/helpers.py b/behave_framework/src/minifi_test_framework/core/helpers.py
index 3c551b8d4..e6ae51857 100644
--- a/behave_framework/src/minifi_test_framework/core/helpers.py
+++ b/behave_framework/src/minifi_test_framework/core/helpers.py
@@ -31,6 +31,15 @@ def log_due_to_failure(context: MinifiTestContext | None):
container.log_app_output()
+def check_condition_after_wait(condition: Callable[[], bool], context: MinifiTestContext | None, wait_time: int) -> bool:
+ time.sleep(wait_time)
+ if not condition():
+ logging.warning("Condition not met after wait")
+ log_due_to_failure(context)
+ return False
+ return True
+
+
def wait_for_condition(condition: Callable[[], bool], timeout_seconds: float, bail_condition: Callable[[], bool],
context: MinifiTestContext | None) -> bool:
if bail_condition():
@@ -50,8 +59,8 @@ def wait_for_condition(condition: Callable[[], bool], timeout_seconds: float, ba
sleep_time = min(1.0, remaining_time)
if sleep_time > 0:
time.sleep(sleep_time)
- except (Exception,):
- logging.warning("Exception while waiting for condition")
+ except Exception as ex:
+ logging.warning("Exception while waiting for condition: %s", ex)
log_due_to_failure(context)
return False
logging.warning("Timed out after %d seconds while waiting for condition", timeout_seconds)
diff --git a/behave_framework/src/minifi_test_framework/core/hooks.py b/behave_framework/src/minifi_test_framework/core/hooks.py
index 5513c9f51..80528fa96 100644
--- a/behave_framework/src/minifi_test_framework/core/hooks.py
+++ b/behave_framework/src/minifi_test_framework/core/hooks.py
@@ -23,6 +23,7 @@ import types
from behave.model import Scenario
from behave.runner import Context
+from minifi_test_framework.core.ssl_utils import make_self_signed_cert
from minifi_test_framework.core.minifi_test_context import MinifiTestContext
@@ -74,6 +75,7 @@ def common_before_scenario(context: Context, scenario: Scenario):
context.network = docker_client.networks.create(network_name)
context.containers = {}
context.resource_dir = None
+ context.root_ca_cert, context.root_ca_key = make_self_signed_cert("root CA")
for step in scenario.steps:
inject_scenario_id(context, step)
diff --git a/behave_framework/src/minifi_test_framework/core/minifi_test_context.py b/behave_framework/src/minifi_test_framework/core/minifi_test_context.py
index ab9110de0..3cfb0264e 100644
--- a/behave_framework/src/minifi_test_framework/core/minifi_test_context.py
+++ b/behave_framework/src/minifi_test_framework/core/minifi_test_context.py
@@ -15,11 +15,16 @@
# limitations under the License.
#
+from __future__ import annotations
+from typing import TYPE_CHECKING
from behave.runner import Context
from docker.models.networks import Network
from minifi_test_framework.containers.container import Container
-from minifi_test_framework.containers.minifi_container import MinifiContainer
+from OpenSSL import crypto
+
+if TYPE_CHECKING:
+ from minifi_test_framework.containers.minifi_container import MinifiContainer
DEFAULT_MINIFI_CONTAINER_NAME = "minifi-primary"
@@ -30,10 +35,13 @@ class MinifiTestContext(Context):
network: Network
minifi_container_image: str
resource_dir: str | None
+ root_ca_key: crypto.PKey
+ root_ca_cert: crypto.X509
def get_or_create_minifi_container(self, container_name: str) -> MinifiContainer:
if container_name not in self.containers:
- self.containers[container_name] = MinifiContainer(self.minifi_container_image, container_name, self.scenario_id, self.network)
+ from minifi_test_framework.containers.minifi_container import MinifiContainer
+ self.containers[container_name] = MinifiContainer(container_name, self)
return self.containers[container_name]
def get_or_create_default_minifi_container(self) -> MinifiContainer:
diff --git a/behave_framework/src/minifi_test_framework/core/ssl_utils.py b/behave_framework/src/minifi_test_framework/core/ssl_utils.py
new file mode 100644
index 000000000..d39cf9646
--- /dev/null
+++ b/behave_framework/src/minifi_test_framework/core/ssl_utils.py
@@ -0,0 +1,161 @@
+# 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 time
+import logging
+import random
+
+from M2Crypto import X509, EVP, RSA, ASN1
+from OpenSSL import crypto
+
+
+def gen_cert():
+ """
+ Generate TLS certificate request for testing
+ """
+
+ req, key = gen_req()
+ pub_key = req.get_pubkey()
+ subject = req.get_subject()
+ cert = X509.X509()
+ # noinspection PyTypeChecker
+ cert.set_serial_number(1)
+ cert.set_version(2)
+ cert.set_subject(subject)
+ t = int(time.time())
+ now = ASN1.ASN1_UTCTIME()
+ now.set_time(t)
+ now_plus_year = ASN1.ASN1_UTCTIME()
+ now_plus_year.set_time(t + 60 * 60 * 24 * 365)
+ cert.set_not_before(now)
+ cert.set_not_after(now_plus_year)
+ issuer = X509.X509_Name()
+ issuer.C = 'US'
+ issuer.CN = 'minifi-listen'
+ cert.set_issuer(issuer)
+ cert.set_pubkey(pub_key)
+ cert.sign(key, 'sha256')
+
+ return cert, key
+
+
+def rsa_gen_key_callback():
+ pass
+
+
+def gen_req():
+ """
+ Generate TLS certificate request for testing
+ """
+
+ logging.info('Generating test certificate request')
+ key = EVP.PKey()
+ req = X509.Request()
+ rsa = RSA.gen_key(1024, 65537, rsa_gen_key_callback)
+ key.assign_rsa(rsa)
+ req.set_pubkey(key)
+ name = req.get_subject()
+ name.C = 'US'
+ name.CN = 'minifi-listen'
+ req.sign(key, 'sha256')
+
+ return req, key
+
+
+def make_self_signed_cert(common_name):
+ ca_key = crypto.PKey()
+ ca_key.generate_key(crypto.TYPE_RSA, 2048)
+
+ ca_cert = crypto.X509()
+ ca_cert.set_version(2)
+ ca_cert.set_serial_number(random.randint(50000000, 100000000))
+
+ ca_subj = ca_cert.get_subject()
+ ca_subj.commonName = common_name
+
+ ca_cert.add_extensions([
+ crypto.X509Extension(b"subjectKeyIdentifier", False, b"hash", subject=ca_cert),
+ ])
+
+ ca_cert.add_extensions([
+ crypto.X509Extension(b"authorityKeyIdentifier", False, b"keyid:always", issuer=ca_cert),
+ ])
+
+ ca_cert.add_extensions([
+ crypto.X509Extension(b"basicConstraints", False, b"CA:TRUE"),
+ crypto.X509Extension(b"keyUsage", False, b"keyCertSign, cRLSign"),
+ ])
+
+ ca_cert.set_issuer(ca_subj)
+ ca_cert.set_pubkey(ca_key)
+
+ ca_cert.gmtime_adj_notBefore(0)
+ ca_cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)
+
+ ca_cert.sign(ca_key, 'sha256')
+
+ return ca_cert, ca_key
+
+
+def _make_cert(common_name, ca_cert, ca_key, extended_key_usage=None):
+ key = crypto.PKey()
+ key.generate_key(crypto.TYPE_RSA, 2048)
+
+ cert = crypto.X509()
+ cert.set_version(2)
+ cert.set_serial_number(random.randint(50000000, 100000000))
+
+ client_subj = cert.get_subject()
+ client_subj.commonName = common_name
+
+ cert.add_extensions([
+ crypto.X509Extension(b"basicConstraints", False, b"CA:FALSE"),
+ crypto.X509Extension(b"subjectKeyIdentifier", False, b"hash", subject=cert),
+ ])
+
+ extensions = [crypto.X509Extension(b"authorityKeyIdentifier", False, b"keyid:always", issuer=ca_cert),
+ crypto.X509Extension(b"keyUsage", False, b"digitalSignature")]
+
+ if extended_key_usage:
+ extensions.append(crypto.X509Extension(b"extendedKeyUsage", False, extended_key_usage))
+
+ cert.add_extensions([
+ crypto.X509Extension(b"subjectAltName", False, b"DNS.1:" + common_name.encode())
+ ])
+
+ cert.add_extensions(extensions)
+
+ cert.set_issuer(ca_cert.get_subject())
+ cert.set_pubkey(key)
+
+ cert.gmtime_adj_notBefore(0)
+ cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)
+
+ cert.sign(ca_key, 'sha256')
+
+ return cert, key
+
+
+def make_client_cert(common_name, ca_cert, ca_key):
+ return _make_cert(common_name=common_name, ca_cert=ca_cert, ca_key=ca_key, extended_key_usage=b"clientAuth")
+
+
+def make_server_cert(common_name, ca_cert, ca_key):
+ return _make_cert(common_name=common_name, ca_cert=ca_cert, ca_key=ca_key, extended_key_usage=b"serverAuth")
+
+
+def make_cert_without_extended_usage(common_name, ca_cert, ca_key):
+ return _make_cert(common_name=common_name, ca_cert=ca_cert, ca_key=ca_key, extended_key_usage=None)
diff --git a/behave_framework/src/minifi_test_framework/minifi/processor.py b/behave_framework/src/minifi_test_framework/minifi/processor.py
index 4b9be6622..fabbc733f 100644
--- a/behave_framework/src/minifi_test_framework/minifi/processor.py
+++ b/behave_framework/src/minifi_test_framework/minifi/processor.py
@@ -32,6 +32,10 @@ class Processor:
def add_property(self, property_name: str, property_value: str):
self.properties[property_name] = property_value
+ def remove_property(self, property_name: str):
+ if property_name in self.properties:
+ del self.properties[property_name]
+
def __repr__(self):
return f"({self.class_name}, {self.name}, {self.properties})"
diff --git a/behave_framework/src/minifi_test_framework/steps/checking_steps.py b/behave_framework/src/minifi_test_framework/steps/checking_steps.py
index 418c49c4f..f0376e5ee 100644
--- a/behave_framework/src/minifi_test_framework/steps/checking_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/checking_steps.py
@@ -17,16 +17,17 @@
import time
+import re
import humanfriendly
from behave import then, step
from minifi_test_framework.containers.http_proxy_container import HttpProxy
-from minifi_test_framework.core.helpers import wait_for_condition
+from minifi_test_framework.core.helpers import wait_for_condition, check_condition_after_wait
from minifi_test_framework.core.minifi_test_context import DEFAULT_MINIFI_CONTAINER_NAME, MinifiTestContext
-@then('there is a file with "{content}" content at {path} in less than {duration}')
+@then('a file with the content "{content}" is placed on the path "{path}" in less than {duration}')
def step_impl(context: MinifiTestContext, content: str, path: str, duration: str):
new_content = content.replace("\\n", "\n")
timeout_in_seconds = humanfriendly.parse_timespan(duration)
@@ -35,7 +36,7 @@ def step_impl(context: MinifiTestContext, content: str, path: str, duration: str
timeout_seconds=timeout_in_seconds, bail_condition=lambda: context.get_default_minifi_container().exited, context=context)
-@then('there is a single file with "{content}" content in the "{directory}" directory in less than {duration}')
+@then('a single file with the content "{content}" is placed in the "{directory}" directory in less than {duration}')
def step_impl(context: MinifiTestContext, content: str, directory: str, duration: str):
new_content = content.replace("\\n", "\n")
timeout_in_seconds = humanfriendly.parse_timespan(duration)
@@ -83,25 +84,44 @@ def step_impl(context: MinifiTestContext, message: str, duration: str):
context=context)
+@then("the Minifi logs match the following regex: \"{regex}\" in less than {duration}")
+def step_impl(context, regex, duration):
+ duration_seconds = humanfriendly.parse_timespan(duration)
+ assert wait_for_condition(condition=lambda: re.search(regex, context.get_default_minifi_container().get_logs()),
+ timeout_seconds=duration_seconds, bail_condition=lambda: context.get_default_minifi_container().exited,
+ context=context)
+
+
@step('no errors were generated on the http-proxy regarding "{url}"')
def step_impl(context: MinifiTestContext, url: str):
http_proxy_container = next(container for container in context.containers.values() if isinstance(container, HttpProxy))
assert http_proxy_container.check_http_proxy_access(url) or http_proxy_container.log_app_output()
-@then('there are {num_str} files in the "{directory}" directory in less than {duration}')
-@then('there is {num_str} file in the "{directory}" directory in less than {duration}')
-def step_impl(context: MinifiTestContext, num_str: str, directory: str, duration: str):
+@then('no files are placed in the "{directory}" directory in {duration} of running time')
+def step_impl(context, directory, duration):
duration_seconds = humanfriendly.parse_timespan(duration)
- assert wait_for_condition(condition=lambda: context.get_default_minifi_container().get_number_of_files(directory) == int(num_str),
+ assert check_condition_after_wait(condition=lambda: context.get_default_minifi_container().get_number_of_files(directory) == 0,
+ context=context, wait_time=duration_seconds)
+
+
+@then('{num:d} files are placed in the "{directory}" directory in less than {duration}')
+@then('{num:d} file is placed in the "{directory}" directory in less than {duration}')
+def step_impl(context: MinifiTestContext, num: int, directory: str, duration: str):
+ duration_seconds = humanfriendly.parse_timespan(duration)
+ if num == 0:
+ context.execute_steps(f'then no files are placed in the "{directory}" directory in {duration} of running time')
+ return
+ assert wait_for_condition(condition=lambda: context.get_default_minifi_container().get_number_of_files(directory) == num,
timeout_seconds=duration_seconds, bail_condition=lambda: context.get_default_minifi_container().exited,
context=context)
-@then('there are at least {num_str} files is in the "{directory}" directory in less than {duration}')
-def step_impl(context: MinifiTestContext, num_str: str, directory: str, duration: str):
+@then('at least {num:d} files are placed in the "{directory}" directory in less than {duration}')
+@then('at least {num:d} file is placed in the "{directory}" directory in less than {duration}')
+def step_impl(context: MinifiTestContext, num: int, directory: str, duration: str):
duration_seconds = humanfriendly.parse_timespan(duration)
- assert wait_for_condition(condition=lambda: context.get_default_minifi_container().get_number_of_files(directory) >= int(num_str),
+ assert wait_for_condition(condition=lambda: context.get_default_minifi_container().get_number_of_files(directory) >= num,
timeout_seconds=duration_seconds, bail_condition=lambda: context.get_default_minifi_container().exited,
context=context)
@@ -114,7 +134,7 @@ def step_impl(context: MinifiTestContext, directory: str, regex_str: str, durati
timeout_seconds=duration_seconds, bail_condition=lambda: context.get_default_minifi_container().exited, context=context)
-@then('the contents of {directory} in less than {timeout} are: "{content_one}" and "{content_two}"')
+@then('files with contents "{content_one}" and "{content_two}" are placed in the "{directory}" directory in less than {timeout}')
def step_impl(context: MinifiTestContext, directory: str, timeout: str, content_one: str, content_two: str):
timeout_seconds = humanfriendly.parse_timespan(timeout)
c1 = content_one.replace("\\n", "\n")
@@ -125,7 +145,7 @@ def step_impl(context: MinifiTestContext, directory: str, timeout: str, content_
context=context)
-@then('the contents of {directory} in less than {timeout} are: "{contents}"')
+@then('files with contents "{contents}" are placed in the "{directory}" directory in less than {timeout}')
def step_impl(context: MinifiTestContext, directory: str, timeout: str, contents: str):
timeout_seconds = humanfriendly.parse_timespan(timeout)
new_contents = contents.replace("\\n", "\n")
@@ -152,3 +172,28 @@ def step_impl(context: MinifiTestContext, max_increase: str, duration: str):
time.sleep(time_in_seconds)
final_memory_usage = context.get_default_minifi_container().get_memory_usage()
assert final_memory_usage - initial_memory_usage <= max_increase_in_bytes
+
+
+@then('after a wait of {duration}, at least {lower_bound:d} and at most {upper_bound:d} files are produced and placed in the "{directory}" directory')
+def step_impl(context, lower_bound, upper_bound, duration, directory):
+ duration_seconds = humanfriendly.parse_timespan(duration)
+ assert check_condition_after_wait(condition=lambda: context.get_default_minifi_container().get_number_of_files(directory) >= lower_bound
+ and context.get_default_minifi_container().get_number_of_files(directory) <= upper_bound,
+ context=context, wait_time=duration_seconds)
+
+
+@then('exactly these files are in the "{directory}" directory in less than {duration}: "{contents}"')
+def step_impl(context, directory, duration, contents):
+ if not contents:
+ context.execute_steps(f'then no files are placed in the "{directory}" directory in {duration} of running time')
+ return
+ contents_arr = contents.split(",")
+ timeout_in_seconds = humanfriendly.parse_timespan(duration)
+ assert wait_for_condition(condition=lambda: context.get_default_minifi_container().verify_file_contents(directory, contents_arr),
+ timeout_seconds=timeout_in_seconds, bail_condition=lambda: False,
+ context=context)
+
+
+@then('exactly these files are in the "{directory}" directory in less than {duration}: ""')
+def step_impl(context, directory, duration):
+ context.execute_steps(f'then no files are placed in the "{directory}" directory in {duration} of running time')
diff --git a/behave_framework/src/minifi_test_framework/steps/core_steps.py b/behave_framework/src/minifi_test_framework/steps/core_steps.py
index 343e622f7..c2369f5ad 100644
--- a/behave_framework/src/minifi_test_framework/steps/core_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/core_steps.py
@@ -38,7 +38,7 @@ def step_impl(context: MinifiTestContext):
@when("the MiNiFi instance starts up")
-def step_impl(context):
+def step_impl(context: MinifiTestContext):
assert context.get_or_create_default_minifi_container().deploy()
logging.debug("MiNiFi instance started up")
@@ -71,5 +71,15 @@ def step_impl(context: MinifiTestContext, filename: str, container_path: str):
@step("after {duration} have passed")
@step("after {duration} has passed")
-def step_impl(context, duration):
+def step_impl(context: MinifiTestContext, duration: str):
time.sleep(humanfriendly.parse_timespan(duration))
+
+
+@when("MiNiFi is stopped")
+def step_impl(context: MinifiTestContext):
+ context.get_or_create_default_minifi_container().stop()
+
+
+@when("MiNiFi is restarted")
+def step_impl(context: MinifiTestContext):
+ context.get_or_create_default_minifi_container().restart()
diff --git a/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py b/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py
index a41d096c7..22c3c604d 100644
--- a/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py
+++ b/behave_framework/src/minifi_test_framework/steps/flow_building_steps.py
@@ -15,6 +15,7 @@
# limitations under the License.
#
+import logging
from behave import given, step
from minifi_test_framework.containers.directory import Directory
@@ -77,7 +78,10 @@ def step_impl(context: MinifiTestContext, processor_type: str):
@step('the "{property_name}" property of the {processor_name} processor is set to "{property_value}"')
def step_impl(context: MinifiTestContext, property_name: str, processor_name: str, property_value: str):
processor = context.get_or_create_default_minifi_container().flow_definition.get_processor(processor_name)
- processor.add_property(property_name, property_value)
+ if property_value == "(not set)":
+ processor.remove_property(property_name)
+ else:
+ processor.add_property(property_name, property_value)
@step('a Funnel with the name "{funnel_name}" is set up')
@@ -220,3 +224,30 @@ def step_impl(context: MinifiTestContext, service_name: str, property_name: str,
controller_service = ControllerService(class_name=service_name, service_name=service_name)
controller_service.add_property(property_name, property_value)
context.get_or_create_default_minifi_container().flow_definition.controller_services.append(controller_service)
+
+
+@given("the \"{property_name}\" property of the {processor_name} processor is set to match the attribute \"{attribute_key}\" to \"{attribute_value}\"")
+def step_impl(context: MinifiTestContext, property_name: str, processor_name: str, attribute_key: str, attribute_value: str):
+ processor = context.get_or_create_default_minifi_container().flow_definition.get_processor(processor_name)
+ if attribute_value == "(not set)":
+ # Ignore filtering
+ processor.add_property(property_name, "true")
+ return
+ filtering = "${" + attribute_key + ":equals('" + attribute_value + "')}"
+ logging.info("Filter: \"%s\"", filtering)
+ logging.info("Key: \"%s\", value: \"%s\"", attribute_key, attribute_value)
+ processor.add_property(property_name, filtering)
+
+
+# TLS
+@given("an ssl context service is set up for {processor_name}")
+@given("an ssl context service with a manual CA cert file is set up for {processor_name}")
+def step_impl(context, processor_name):
+ controller_service = ControllerService(class_name="SSLContextService", service_name="SSLContextService")
+ controller_service.add_property("Client Certificate", "/tmp/resources/minifi_client.crt")
+ controller_service.add_property("Private Key", "/tmp/resources/minifi_client.key")
+ controller_service.add_property("CA Certificate", "/tmp/resources/root_ca.crt")
+ context.get_or_create_default_minifi_container().flow_definition.controller_services.append(controller_service)
+
+ processor = context.get_or_create_default_minifi_container().flow_definition.get_processor(processor_name)
+ processor.add_property('SSL Context Service', 'SSLContextService')
diff --git a/docker/RunBehaveTests.sh b/docker/RunBehaveTests.sh
index 89d59f076..786e9e6b3 100755
--- a/docker/RunBehaveTests.sh
+++ b/docker/RunBehaveTests.sh
@@ -199,4 +199,5 @@ exec \
"${docker_dir}/../extensions/azure/tests/features" \
"${docker_dir}/../extensions/sql/tests/features" \
"${docker_dir}/../extensions/llamacpp/tests/features" \
- "${docker_dir}/../extensions/opc/tests/features"
+ "${docker_dir}/../extensions/opc/tests/features" \
+ "${docker_dir}/../extensions/kafka/tests/features"
diff --git a/docker/test/integration/cluster/ContainerStore.py b/docker/test/integration/cluster/ContainerStore.py
index a78b0712c..d47f59ab1 100644
--- a/docker/test/integration/cluster/ContainerStore.py
+++ b/docker/test/integration/cluster/ContainerStore.py
@@ -18,7 +18,6 @@ from .containers.MinifiContainer import MinifiOptions
from .containers.MinifiContainer import MinifiContainer
from .containers.NifiContainer import NifiContainer
from .containers.NifiContainer import NiFiOptions
-from .containers.KafkaBrokerContainer import KafkaBrokerContainer
from .containers.KinesisServerContainer import KinesisServerContainer
from .containers.S3ServerContainer import S3ServerContainer
from .containers.AzureStorageServerContainer import AzureStorageServerContainer
@@ -119,14 +118,6 @@ class ContainerStore:
network=self.network,
image_store=self.image_store,
command=command))
- elif engine == 'kafka-broker':
- return self.containers.setdefault(container_name,
- KafkaBrokerContainer(feature_context=feature_context,
- name=container_name,
- vols=self.vols,
- network=self.network,
- image_store=self.image_store,
- command=command))
elif engine == 'http-proxy':
return self.containers.setdefault(container_name,
HttpProxyContainer(feature_context=feature_context,
diff --git a/docker/test/integration/cluster/DockerTestCluster.py b/docker/test/integration/cluster/DockerTestCluster.py
index 620f84267..58902f132 100644
--- a/docker/test/integration/cluster/DockerTestCluster.py
+++ b/docker/test/integration/cluster/DockerTestCluster.py
@@ -30,7 +30,6 @@ from .checkers.AwsChecker import AwsChecker
from .checkers.AzureChecker import AzureChecker
from .checkers.ElasticSearchChecker import ElasticSearchChecker
from .checkers.GcsChecker import GcsChecker
-from .checkers.KafkaHelper import KafkaHelper
from .checkers.PostgresChecker import PostgresChecker
from .checkers.PrometheusChecker import PrometheusChecker
from .checkers.SplunkChecker import SplunkChecker
@@ -58,7 +57,6 @@ class DockerTestCluster:
self.minifi_controller_executor = MinifiControllerExecutor(self.container_communicator)
self.modbus_checker = ModbusChecker(self.container_communicator)
self.couchbase_checker = CouchbaseChecker()
- self.kafka_checker = KafkaHelper(self.container_communicator, feature_id)
self.mqtt_helper = MqttHelper()
def cleanup(self):
@@ -293,14 +291,6 @@ class DockerTestCluster:
def segfault_happened(self):
return self.segfault
- def wait_for_kafka_consumer_to_be_registered(self, kafka_container_name: str, count: int):
- kafka_container_name = self.container_store.get_container_name_with_postfix(kafka_container_name)
- return self.wait_for_app_logs_regex(kafka_container_name, "Assignment received from leader.*for group docker_test_group", 60, count)
-
- def wait_for_kafka_consumer_to_be_reregistered(self, kafka_container_name):
- kafka_container_name = self.container_store.get_container_name_with_postfix(kafka_container_name)
- return self.wait_for_app_logs_regex(kafka_container_name, "Assignment received from leader.*for group docker_test_group.*Assignment received from leader.*for group docker_test_group.", 60)
-
def wait_for_metric_class_on_prometheus(self, metric_class, timeout_seconds):
return self.prometheus_checker.wait_for_metric_class_on_prometheus(metric_class, timeout_seconds)
diff --git a/docker/test/integration/cluster/checkers/KafkaHelper.py b/docker/test/integration/cluster/checkers/KafkaHelper.py
deleted file mode 100644
index 0307e497b..000000000
--- a/docker/test/integration/cluster/checkers/KafkaHelper.py
+++ /dev/null
@@ -1,58 +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.
-from __future__ import annotations
-
-import io
-import logging
-import docker
-
-
-class KafkaHelper:
- def __init__(self, container_communicator, feature_id):
- self.container_communicator = container_communicator
- self.feature_id = feature_id
-
- def create_topic(self, container_name: str, topic_name: str):
- logging.info(f"Creating topic {topic_name} in {container_name}")
- (code, output) = self.container_communicator.execute_command(container_name, ["/bin/bash", "-c", f"/opt/kafka/bin/kafka-topics.sh --create --topic {topic_name} --bootstrap-server {container_name}:9092"])
- logging.info(output)
- return code == 0
-
- def produce_message(self, container_name: str, topic_name: str, message: str):
- logging.info(f"Sending {message} to {container_name}:{topic_name}")
- (code, output) = self.container_communicator.execute_command(container_name, ["/bin/bash", "-c", f"/opt/kafka/bin/kafka-console-producer.sh --topic {topic_name} --bootstrap-server {container_name}:9092 <<< '{message}'"])
- logging.info(output)
- return code == 0
-
- def produce_message_with_key(self, container_name: str, topic_name: str, message: str, message_key: str):
- logging.info(f"Sending {message} to {container_name}:{topic_name}")
- (code, output) = self.container_communicator.execute_command(container_name, ["/bin/bash", "-c", f"/opt/kafka/bin/kafka-console-producer.sh --property 'key.separator=:' --property 'parse.key=true' --topic {topic_name} --bootstrap-server {container_name}:9092 <<< '{message_key}:{message}'"])
- logging.info(output)
- return code == 0
-
- def run_python_in_kafka_helper_docker(self, command: str):
- try:
- self.container_communicator.client.images.get("kafka-helper")
- except docker.errors.ImageNotFound:
- dockerfile_content = """
- FROM python:3.13-slim-bookworm
- RUN pip install confluent-kafka
- """
- dockerfile_stream = io.BytesIO(dockerfile_content.encode("utf-8"))
- self.container_communicator.client.images.build(fileobj=dockerfile_stream, tag="kafka-helper")
-
- output = self.container_communicator.client.containers.run("kafka-helper", ["python", "-c", command], remove=True, stdout=True, stderr=True, network=f"minifi_integration_test_network-{self.feature_id}")
- logging.info(output)
- return True
diff --git a/docker/test/integration/cluster/containers/KafkaBrokerContainer.py b/docker/test/integration/cluster/containers/KafkaBrokerContainer.py
deleted file mode 100644
index a597792d3..000000000
--- a/docker/test/integration/cluster/containers/KafkaBrokerContainer.py
+++ /dev/null
@@ -1,163 +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.
-
-
-import logging
-import tempfile
-import docker.types
-import jks
-import os
-from OpenSSL import crypto
-
-from .Container import Container
-from ssl_utils.SSL_cert_utils import make_server_cert
-
-
-class KafkaBrokerContainer(Container):
- def __init__(self, feature_context, name, vols, network, image_store, command=None):
- super().__init__(feature_context, name, 'kafka-broker', vols, network, image_store, command)
-
- kafka_cert, kafka_key = make_server_cert(f"kafka-broker-{feature_context.id}", feature_context.root_ca_cert, feature_context.root_ca_key)
-
- pke = jks.PrivateKeyEntry.new('kafka-broker-cert', [crypto.dump_certificate(crypto.FILETYPE_ASN1, kafka_cert)], crypto.dump_privatekey(crypto.FILETYPE_ASN1, kafka_key), 'rsa_raw')
- server_keystore = jks.KeyStore.new('jks', [pke])
-
- with tempfile.NamedTemporaryFile(delete=False, suffix=".jks") as server_keystore_file:
- server_keystore.save(server_keystore_file.name, 'abcdefgh')
- self.server_keystore_file_path = server_keystore_file.name
-
- os.chmod(self.server_keystore_file_path, 0o644)
-
- with tempfile.NamedTemporaryFile(delete=False) as credentials_file:
- credentials_file.write(b"abcdefgh")
- self.credentials_file_path = credentials_file.name
-
- os.chmod(self.credentials_file_path, 0o644)
-
- trusted_cert = jks.TrustedCertEntry.new(
- 'root-ca', # Alias for the certificate
- crypto.dump_certificate(crypto.FILETYPE_ASN1, feature_context.root_ca_cert)
- )
-
- # Create a JKS keystore that will serve as a truststore with the trusted cert entry.
- truststore = jks.KeyStore.new('jks', [trusted_cert])
-
- with tempfile.NamedTemporaryFile(delete=False, suffix=".jks") as server_truststore_file:
- truststore.save(server_truststore_file.name, 'abcdefgh')
- self.server_truststore_file_path = server_truststore_file.name
-
- os.chmod(self.server_truststore_file_path, 0o644)
-
- with tempfile.NamedTemporaryFile(delete=False, mode="w", encoding="utf-8") as jaas_config_file:
- jaas_config_file.write("""
-KafkaServer {
- org.apache.kafka.common.security.plain.PlainLoginModule required
- username="admin"
- password="admin-secret"
- user_admin="admin-secret"
- user_alice="alice-secret";
-};
-
-Client {
- org.apache.kafka.common.security.plain.PlainLoginModule required
- username="admin"
- password="admin-secret";
-};
-""")
- self.jaas_config_file_path = jaas_config_file.name
- os.chmod(self.jaas_config_file_path, 0o644)
-
- def get_startup_finished_log_entry(self):
- return "Kafka Server started"
-
- def deploy(self):
- if not self.set_deployed():
- return
-
- logging.info('Creating and running kafka broker docker container...')
- self.client.containers.run(
- image="apache/kafka:4.1.0",
- detach=True,
- name=self.name,
- network=self.network.name,
- environment=[
- "KAFKA_NODE_ID=1",
- "KAFKA_PROCESS_ROLES=controller,broker",
- "KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT",
- "KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER",
- "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1",
- "KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1",
- "KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1",
-
- f"KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka-broker-{self.feature_context.id}:9096",
-
- f"KAFKA_LISTENERS=PLAINTEXT://kafka-broker-{self.feature_context.id}:9092,"
- f"SASL_PLAINTEXT://kafka-broker-{self.feature_context.id}:9094,"
- f"SSL://kafka-broker-{self.feature_context.id}:9093,"
- f"SASL_SSL://kafka-broker-{self.feature_context.id}:9095,"
- f"CONTROLLER://kafka-broker-{self.feature_context.id}:9096",
-
- f"KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka-broker-{self.feature_context.id}:9092,"
- f"SASL_PLAINTEXT://kafka-broker-{self.feature_context.id}:9094,"
- f"SSL://kafka-broker-{self.feature_context.id}:9093,"
- f"SASL_SSL://kafka-broker-{self.feature_context.id}:9095,"
- f"CONTROLLER://kafka-broker-{self.feature_context.id}:9096",
-
- "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,"
- "SASL_PLAINTEXT:SASL_PLAINTEXT,"
- "SSL:SSL,"
- "SASL_SSL:SASL_SSL,"
- "CONTROLLER:PLAINTEXT",
-
- # **If using SASL_PLAINTEXT, provide JAAS config**
- 'KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN',
- 'KAFKA_SASL_ENABLED_MECHANISMS=PLAIN',
- 'KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kafka_jaas.conf -Dlog4j2.rootLogger.level=DEBUG -Dlog4j2.logger.org.apache.kafka.controller.level=DEBUG',
-
- "KAFKA_SSL_PROTOCOL=TLS",
- "KAFKA_SSL_ENABLED_PROTOCOLS=TLSv1.2",
- "KAFKA_SSL_KEYSTORE_TYPE=JKS",
- "KAFKA_SSL_KEYSTORE_FILENAME=kafka.keystore.jks",
- "KAFKA_SSL_KEYSTORE_CREDENTIALS=credentials.conf",
- "KAFKA_SSL_KEY_CREDENTIALS=credentials.conf",
- "KAFKA_SSL_TRUSTSTORE_CREDENTIALS=credentials.conf",
- "KAFKA_SSL_TRUSTSTORE_TYPE=JKS",
- "KAFKA_SSL_TRUSTSTORE_FILENAME=kafka.truststore.jks",
- "KAFKA_SSL_CLIENT_AUTH=none"
- ],
- mounts=[
- docker.types.Mount(
- type='bind',
- source=self.server_keystore_file_path,
- target='/etc/kafka/secrets/kafka.keystore.jks'
- ),
- docker.types.Mount(
- type='bind',
- source=self.server_truststore_file_path,
- target='/etc/kafka/secrets/kafka.truststore.jks'
- ),
- docker.types.Mount(
- type='bind',
- source=self.jaas_config_file_path,
- target='/opt/kafka/config/kafka_jaas.conf'
- ),
- docker.types.Mount(
- type='bind',
- source=self.credentials_file_path,
- target='/etc/kafka/secrets/credentials.conf'
- )
- ],
- entrypoint=self.command)
- logging.info('Added container \'%s\'', self.name)
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index e74e1772f..9e07421c6 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -64,11 +64,6 @@ class MiNiFi_integration_test:
def acquire_transient_minifi(self, context, name, engine='minifi-cpp'):
return self.cluster.acquire_transient_minifi(context=context, name=name, engine=engine)
- def start_kafka_broker(self, context):
- self.cluster.acquire_container(context=context, name='kafka-broker', engine='kafka-broker')
- self.cluster.deploy_container(name='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')
@@ -323,9 +318,6 @@ class MiNiFi_integration_test:
def check_azure_storage_server_data(self, azure_container_name, object_data):
assert self.cluster.check_azure_storage_server_data(azure_container_name, object_data) or self.cluster.log_app_output()
- def wait_for_kafka_consumer_to_be_registered(self, kafka_container_name, count):
- assert self.cluster.wait_for_kafka_consumer_to_be_registered(kafka_container_name, count) or self.cluster.log_app_output()
-
def check_splunk_event(self, splunk_container_name, query):
assert self.cluster.check_splunk_event(splunk_container_name, query) or self.cluster.log_app_output()
diff --git a/docker/test/integration/features/steps/kafka_steps.py b/docker/test/integration/features/steps/kafka_steps.py
deleted file mode 100644
index 6f7d21325..000000000
--- a/docker/test/integration/features/steps/kafka_steps.py
+++ /dev/null
@@ -1,188 +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.
-
-
-from behave import given, when
-import logging
-import binascii
-import time
-
-
-@given("a kafka producer workflow publishing files placed in \"{directory}\" to a broker exactly once")
-def publish_producer_workflow(context, directory):
- context.execute_steps("""
- given a GetFile processor with the \"Input Directory\" property set to \"{directory}\"
- and the \"Keep Source File\" property of the GetFile processor is set to \"false\"
- and a PublishKafka processor set up to communicate with a kafka broker instance
- and the "success" relationship of the GetFile processor is connected to the PublishKafka""".format(directory=directory))
-
-
-@given("the \"{property_name}\" property of the {processor_name} processor is set to match {key_attribute_encoding} encoded kafka message key \"{message_key}\"")
-def set_property_to_match_message_key(context, property_name, processor_name, key_attribute_encoding, message_key):
- if key_attribute_encoding.lower() == "hex":
- # Hex is presented upper-case to be in sync with NiFi
- encoded_key = binascii.hexlify(message_key.encode("utf-8")).upper()
- elif key_attribute_encoding.lower() == "(not set)":
- encoded_key = message_key.encode("utf-8")
- else:
- encoded_key = message_key.encode(key_attribute_encoding)
- logging.info("%s processor is set up to match encoded key \"%s\"", processor_name, encoded_key)
- filtering = "${kafka.key:equals('" + encoded_key.decode("utf-8") + "')}"
- logging.info("Filter: \"%s\"", filtering)
- processor = context.test.get_node_by_name(processor_name)
- processor.set_property(property_name, filtering)
-
- # Kafka setup
-
-
-@given("a kafka broker is set up in correspondence with the PublishKafka")
-@given("a kafka broker is set up in correspondence with the third-party kafka publisher")
-@given("a kafka broker is set up in correspondence with the publisher flow")
-def kafka_broker_setup(context):
- context.test.acquire_container(context=context, name="kafka-broker", engine="kafka-broker")
-
-
-@given("the kafka broker is started")
-def kafka_broker_start(context):
- context.test.start_kafka_broker(context)
-
-
-@given("the topic \"{topic_name}\" is initialized on the kafka broker")
-def init_topic(context, topic_name):
- container_name = context.test.get_container_name_with_postfix("kafka-broker")
- assert context.test.cluster.kafka_checker.create_topic(container_name, topic_name) or context.test.cluster.log_app_output()
-
-
-@when("a message with content \"{content}\" is published to the \"{topic_name}\" topic")
-def publish_to_topic(context, content, topic_name):
- container_name = context.test.get_container_name_with_postfix("kafka-broker")
- assert context.test.cluster.kafka_checker.produce_message(container_name, topic_name, content) or context.test.cluster.log_app_output()
-
-
-@when("two messages with content \"{content_one}\" and \"{content_two}\" is published to the \"{topic_name}\" topic")
-def publish_two_messages_to_topic(context, content_one, content_two, topic_name):
- python_code = f"""
-from confluent_kafka import Producer
-import uuid
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092"}})
-producer.produce("{topic_name}", "{content_one}")
-producer.produce("{topic_name}", "{content_two}")
-producer.flush(10)
- """
- assert context.test.cluster.kafka_checker.run_python_in_kafka_helper_docker(python_code) or context.test.cluster.log_app_output()
-
-
-@when("the publisher performs a {transaction_type} transaction publishing to the \"{topic_name}\" topic these messages: {messages}")
-def publish_to_topic_transaction_style(context, transaction_type, topic_name, messages):
- if transaction_type == "SINGLE_COMMITTED_TRANSACTION":
- python_code = f"""
-from confluent_kafka import Producer
-
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092", "transactional.id": "1001"}})
-producer.init_transactions()
-producer.begin_transaction()
-for content in "{messages}".split(", "):
- producer.produce("{topic_name}", content.encode("utf-8"))
-producer.commit_transaction()
-producer.flush(10)
- """
- elif transaction_type == "TWO_SEPARATE_TRANSACTIONS":
- python_code = f"""
-from confluent_kafka import Producer
-
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092", "transactional.id": "1001"}})
-producer.init_transactions()
-for content in "{messages}".split(", "):
- producer.begin_transaction()
- producer.produce("{topic_name}", content.encode("utf-8"))
- producer.commit_transaction()
-producer.flush(10)
- """
- elif transaction_type == "NON_COMMITTED_TRANSACTION":
- python_code = f"""
-from confluent_kafka import Producer
-
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092", "transactional.id": "1001"}})
-producer.init_transactions()
-producer.begin_transaction()
-for content in "{messages}".split(", "):
- producer.produce("{topic_name}", content.encode("utf-8"))
-producer.flush(10)
- """
- elif transaction_type == "CANCELLED_TRANSACTION":
- python_code = f"""
-from confluent_kafka import Producer
-
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092", "transactional.id": "1001"}})
-producer.init_transactions()
-producer.begin_transaction()
-for content in "{messages}".split(", "):
- producer.produce("{topic_name}", content.encode("utf-8"))
-producer.flush(10)
-producer.abort_transaction()
- """
- else:
- raise Exception("Unknown transaction type.")
- assert context.test.cluster.kafka_checker.run_python_in_kafka_helper_docker(python_code) or context.test.cluster.log_app_output()
-
-
-@when("a message with content \"{content}\" is published to the \"{topic_name}\" topic with key \"{message_key}\"")
-def publish_to_topic_with_key(context, content, topic_name, message_key):
- container_name = context.test.get_container_name_with_postfix("kafka-broker")
- assert context.test.cluster.kafka_checker.produce_message_with_key(container_name, topic_name, content, message_key) or context.test.cluster.log_app_output()
-
-
-@when("{number_of_messages} kafka messages are sent to the topic \"{topic_name}\"")
-def publish_batch_to_topic(context, number_of_messages, topic_name):
- python_code = f"""
-from confluent_kafka import Producer
-import uuid
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092"}})
-for i in range(0, int({number_of_messages})):
- producer.produce("{topic_name}", str(uuid.uuid4()).encode("utf-8"))
-producer.flush(10)
- """
- assert context.test.cluster.kafka_checker.run_python_in_kafka_helper_docker(python_code) or context.test.cluster.log_app_output()
-
-
-@when("a message with content \"{content}\" is published to the \"{topic_name}\" topic with headers \"{semicolon_separated_headers}\"")
-def publish_with_headers_to_topic(context, content, topic_name, semicolon_separated_headers):
- python_code = f"""
-from confluent_kafka import Producer
-headers = []
-for header in "{semicolon_separated_headers}".split(";"):
- kv = header.split(":")
- headers.append((kv[0].strip(), kv[1].strip().encode("utf-8")))
-producer = Producer({{"bootstrap.servers": "kafka-broker-{context.feature_id}:9092"}})
-producer.produce("{topic_name}", "{content}".encode("utf-8"), headers=headers)
-producer.flush(10)
- """
- assert context.test.cluster.kafka_checker.run_python_in_kafka_helper_docker(python_code) or context.test.cluster.log_app_output()
-
-
-@when("the Kafka consumer is reregistered in kafka broker")
-def wait_for_consumer_reregistration(context):
- context.test.wait_for_kafka_consumer_to_be_registered("kafka-broker", 2)
- # After the consumer is registered there is still some time needed for consumer-broker synchronization
- # Unfortunately there are no additional log messages that could be checked for this
- time.sleep(2)
-
-
-@when("the Kafka consumer is registered in kafka broker")
-def wait_for_consumer_registration(context):
- context.test.wait_for_kafka_consumer_to_be_registered("kafka-broker", 1)
- # After the consumer is registered there is still some time needed for consumer-broker synchronization
- # Unfortunately there are no additional log messages that could be checked for this
- time.sleep(2)
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 87285e409..fbf62251b 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -128,7 +128,6 @@ def step_impl(context, processor_type, minifi_container_name):
@given("a {processor_type} processor set up to communicate with an s3 server")
@given("a {processor_type} processor set up to communicate with the same s3 server")
@given("a {processor_type} processor set up to communicate with an Azure blob storage")
-@given("a {processor_type} processor set up to communicate with a kafka broker instance")
@given("a {processor_type} processor set up to communicate with an MQTT broker instance")
@given("a {processor_type} processor set up to communicate with the Splunk HEC instance")
@given("a {processor_type} processor set up to communicate with the kinesis server")
diff --git a/docker/test/integration/minifi/processors/ConsumeKafka.py b/docker/test/integration/minifi/processors/ConsumeKafka.py
deleted file mode 100644
index cbc1a482a..000000000
--- a/docker/test/integration/minifi/processors/ConsumeKafka.py
+++ /dev/null
@@ -1,37 +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.
-
-
-from ..core.Processor import Processor
-
-
-class ConsumeKafka(Processor):
- def __init__(self, context, schedule=None):
- super(ConsumeKafka, self).__init__(
- context=context,
- clazz="ConsumeKafka",
- properties={
- "Kafka Brokers": f"kafka-broker-{context.feature_id}:9092",
- "Topic Names": "ConsumeKafkaTest",
- "Topic Name Format": "Names",
- "Honor Transactions": "true",
- "Group ID": "docker_test_group",
- "Offset Reset": "earliest",
- "Key Attribute Encoding": "UTF-8",
- "Message Header Encoding": "UTF-8",
- "Max Poll Time": "4 sec",
- "Session Timeout": "60 sec"},
- auto_terminate=["success"],
- schedule=schedule)
diff --git a/extensions/aws/tests/features/s3.feature b/extensions/aws/tests/features/s3.feature
index e53817158..b16ea1f25 100644
--- a/extensions/aws/tests/features/s3.feature
+++ b/extensions/aws/tests/features/s3.feature
@@ -33,7 +33,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When both instances start up
- Then there is a single file with "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" is placed in the "/tmp/output" directory in less than 20 seconds
And the object on the s3 server is "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{"
And the object content type on the s3 server is "application/octet-stream" and the object metadata matches use metadata
And the Minifi logs contain the following message: "in a single upload" in less than 10 seconds
@@ -58,7 +58,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
And the http proxy server is set up
When all instances start up
- Then there is a single file with "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" is placed in the "/tmp/output" directory in less than 20 seconds
And the object on the s3 server is "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{"
And the object content type on the s3 server is "application/octet-stream" and the object metadata matches use metadata
And no errors were generated on the http-proxy regarding "http://s3-server-s3-1:9090/test_bucket/test_object_key"
@@ -81,7 +81,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When both instances start up
- Then there is a single file with "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" is placed in the "/tmp/output" directory in less than 20 seconds
And the object bucket on the s3 server is empty in less than 10 seconds
Scenario: Deletion of a non-existent s3 object succeeds
@@ -96,7 +96,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When both instances start up
- Then there is a single file with "test" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 20 seconds
And the object bucket on the s3 server is empty in less than 10 seconds
Scenario: Deletion of a s3 object through a proxy-server succeeds
@@ -124,7 +124,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When all instances start up
- Then there is a single file with "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "LH_O#L|FD<FASD{FO#@$#$%^ \"#\"$L%:\"@#$L\":test_data#$#%#$%?{\"F{" is placed in the "/tmp/output" directory in less than 20 seconds
And the object bucket on the s3 server is empty in less than 10 seconds
And no errors were generated on the http-proxy regarding "http://s3-server-s3-4:9090/test_bucket/test_object_key"
@@ -149,7 +149,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When all instances start up
- Then there is a single file with "test" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 20 seconds
Scenario: A MiNiFi instance can download s3 bucket objects via a http-proxy
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
@@ -176,7 +176,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When all instances start up
- Then there is a single file with "test" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 20 seconds
And no errors were generated on the http-proxy regarding "http://s3-server-s3-6:9090/test_bucket/test_object_key"
Scenario: A MiNiFi instance can list an S3 bucket directly
@@ -197,7 +197,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When all instances start up
- Then there are 2 files in the "/tmp/output" directory in less than 20 seconds
+ Then 2 files are placed in the "/tmp/output" directory in less than 20 seconds
Scenario: A MiNiFi instance can list an S3 bucket objects via a http-proxy
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
@@ -220,7 +220,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When all instances start up
- Then there is 1 file in the "/tmp/output" directory in less than 20 seconds
+ Then 1 file is placed in the "/tmp/output" directory in less than 20 seconds
And no errors were generated on the http-proxy regarding "http://s3-server-s3-8:9090/test_bucket"
Scenario: A MiNiFi instance transfers data in multiple parts to s3
@@ -236,7 +236,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
When all instances start up
- Then there is 1 file in the "/tmp/output" directory in less than 20 seconds
+ Then 1 file is placed in the "/tmp/output" directory in less than 20 seconds
And the object on the s3 server is present and matches the original hash
And the Minifi logs contain the following message: "passes the multipart threshold, uploading it in multiple parts" in less than 10 seconds
@@ -260,7 +260,7 @@ Feature: Sending data from MiNiFi-C++ to an AWS server
And the http proxy server is set up
When all instances start up
- Then there is 1 file in the "/tmp/output" directory in less than 20 seconds
+ Then 1 file is placed in the "/tmp/output" directory in less than 20 seconds
And the object on the s3 server is present and matches the original hash
And the Minifi logs contain the following message: "passes the multipart threshold, uploading it in multiple parts" in less than 10 seconds
And no errors were generated on the http-proxy regarding "http://s3-server-s3-10:9090/test_bucket/test_object_key"
diff --git a/extensions/azure/tests/features/azure_storage.feature b/extensions/azure/tests/features/azure_storage.feature
index 520b07274..26fb71dc1 100644
--- a/extensions/azure/tests/features/azure_storage.feature
+++ b/extensions/azure/tests/features/azure_storage.feature
@@ -32,7 +32,7 @@ Feature: Sending data from MiNiFi-C++ to an Azure storage server
And an Azure storage server is set up
When the MiNiFi instance starts up
- Then there is a single file with "#test_data$123$#" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "#test_data$123$#" is placed in the "/tmp/output" directory in less than 20 seconds
And the object on the Azure storage server is "#test_data$123$#"
Scenario: A MiNiFi instance can delete blob from Azure blob storage
@@ -90,7 +90,7 @@ Feature: Sending data from MiNiFi-C++ to an Azure storage server
When the MiNiFi instance starts up
And test blob "test" with the content "#test_data_123$#" is created on Azure blob storage
- Then there is a single file with "data_" content in the "/tmp/output" directory in less than 20 seconds
+ Then a single file with the content "data_" is placed in the "/tmp/output" directory in less than 20 seconds
Scenario: A MiNiFi instance can list a container on Azure blob storage
Given a ListAzureBlobStorage processor set up to communicate with an Azure blob storage
diff --git a/docker/test/integration/features/consumekafka.feature b/extensions/kafka/tests/features/consumekafka.feature
similarity index 71%
rename from docker/test/integration/features/consumekafka.feature
rename to extensions/kafka/tests/features/consumekafka.feature
index 5a2171335..b23133cd2 100644
--- a/docker/test/integration/features/consumekafka.feature
+++ b/extensions/kafka/tests/features/consumekafka.feature
@@ -19,26 +19,25 @@ Feature: Receiving data from using Kafka streaming platform using ConsumeKafka
As a user of MiNiFi
I need to have ConsumeKafka processor
- Background:
- Given the content of "/tmp/output" is monitored
-
Scenario Outline: ConsumeKafka parses and uses kafka topics and topic name formats
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Topic Names" property of the ConsumeKafka processor is set to "<topic names>"
And the "Topic Name Format" property of the ConsumeKafka processor is set to "<topic name format>"
And the "Offset Reset" property of the ConsumeKafka processor is set to "earliest"
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And a kafka broker is set up in correspondence with the third-party kafka publisher
- And the kafka broker is started
+ And the Kafka server is started
And the topic "ConsumeKafkaTest" is initialized on the kafka broker
When a message with content "<message 1>" is published to the "ConsumeKafkaTest" topic
- And all other processes start up
+ And the MiNiFi instance starts up
And a message with content "<message 2>" is published to the "ConsumeKafkaTest" topic
- Then two flowfiles with the contents "<message 1>" and "<message 2>" are placed in the monitored directory in less than 90 seconds
+ Then files with contents "<message 1>" and "<message 2>" are placed in the "/tmp/output" directory in less than 30 seconds
Examples: Topic names and formats to test
| message 1 | message 2 | topic names | topic name format |
@@ -49,24 +48,27 @@ Feature: Receiving data from using Kafka streaming platform using ConsumeKafka
| Hamlet | William Shakespeare | Cons[emu]*KafkaTest | Patterns |
Scenario Outline: ConsumeKafka key attribute is encoded according to the "Key Attribute Encoding" property
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Key Attribute Encoding" property of the ConsumeKafka processor is set to "<key attribute encoding>"
- And a RouteOnAttribute processor in the "kafka-consumer-flow" flow
- And a LogAttribute processor in the "kafka-consumer-flow" flow
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ And a RouteOnAttribute processor
+ And RouteOnAttribute is EVENT_DRIVEN
+ And a LogAttribute processor
+ And LogAttribute is EVENT_DRIVEN
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" property of the RouteOnAttribute processor is set to match <key attribute encoding> encoded kafka message key "consume_kafka_test_key"
And the "success" relationship of the ConsumeKafka processor is connected to the LogAttribute
And the "success" relationship of the LogAttribute processor is connected to the RouteOnAttribute
And the "success" relationship of the RouteOnAttribute processor is connected to the PutFile
-
- And a kafka broker is set up in correspondence with the third-party kafka publisher
+ And PutFile's success relationship is auto-terminated
When all instances start up
And a message with content "<message 1>" is published to the "ConsumeKafkaTest" topic with key "consume_kafka_test_key"
And a message with content "<message 2>" is published to the "ConsumeKafkaTest" topic with key "consume_kafka_test_key"
- Then two flowfiles with the contents "<message 1>" and "<message 2>" are placed in the monitored directory in less than 45 seconds
+ Then files with contents "<message 1>" and "<message 2>" are placed in the "/tmp/output" directory in less than 30 seconds
Examples: Key attribute encoding values
| message 1 | message 2 | key attribute encoding |
@@ -75,18 +77,19 @@ Feature: Receiving data from using Kafka streaming platform using ConsumeKafka
| Crime and Punishment | Фёдор Михайлович Достоевский | Hex |
Scenario Outline: ConsumeKafka transactional behaviour is supported
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Topic Names" property of the ConsumeKafka processor is set to "ConsumeKafkaTest"
And the "Honor Transactions" property of the ConsumeKafka processor is set to "<honor transactions>"
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
-
- And a kafka broker is set up in correspondence with the third-party kafka publisher
+ And PutFile's success relationship is auto-terminated
When all instances start up
And the publisher performs a <transaction type> transaction publishing to the "ConsumeKafkaTest" topic these messages: <messages sent>
- Then <number of flowfiles expected> flowfiles are placed in the monitored directory in less than 15 seconds
+ Then <number of flowfiles expected> files are placed in the "/tmp/output" directory in less than 15 seconds
Examples: Transaction descriptions
| messages sent | transaction type | honor transactions | number of flowfiles expected |
@@ -97,27 +100,29 @@ Feature: Receiving data from using Kafka streaming platform using ConsumeKafka
| Operation Dark Heart | CANCELLED_TRANSACTION | true | 0 |
| Brexit | CANCELLED_TRANSACTION | false | 1 |
-
Scenario Outline: Headers on consumed kafka messages are extracted into attributes if requested on ConsumeKafka
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Headers To Add As Attributes" property of the ConsumeKafka processor is set to "<headers to add as attributes>"
And the "Duplicate Header Handling" property of the ConsumeKafka processor is set to "<duplicate header handling>"
- And a RouteOnAttribute processor in the "kafka-consumer-flow" flow
- And a LogAttribute processor in the "kafka-consumer-flow" flow
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ And a RouteOnAttribute processor
+ And RouteOnAttribute is EVENT_DRIVEN
+ And a LogAttribute processor
+ And LogAttribute is EVENT_DRIVEN
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" property of the RouteOnAttribute processor is set to match the attribute "<headers to add as attributes>" to "<expected value>"
And the "success" relationship of the ConsumeKafka processor is connected to the LogAttribute
And the "success" relationship of the LogAttribute processor is connected to the RouteOnAttribute
And the "success" relationship of the RouteOnAttribute processor is connected to the PutFile
-
- And a kafka broker is set up in correspondence with the third-party kafka publisher
+ And PutFile's success relationship is auto-terminated
When all instances start up
And a message with content "<message 1>" is published to the "ConsumeKafkaTest" topic with headers "<message headers sent>"
And a message with content "<message 2>" is published to the "ConsumeKafkaTest" topic with headers "<message headers sent>"
- Then two flowfiles with the contents "<message 1>" and "<message 2>" are placed in the monitored directory in less than 45 seconds
+ Then files with contents "<message 1>" and "<message 2>" are placed in the "/tmp/output" directory in less than 45 seconds
Examples: Messages with headers
| message 1 | message 2 | message headers sent | headers to add as attributes | expected value | duplicate header handling |
@@ -129,137 +134,144 @@ Feature: Receiving data from using Kafka streaming platform using ConsumeKafka
| The Lord of the Rings | J. R. R. Tolkien | Parts: First, second, third | Parts | First, second, third | (not set) |
Scenario: Messages are merged if the message demarcator is present in the message
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Message Demarcator" property of the ConsumeKafka processor is set to ","
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And a kafka broker is set up in correspondence with the third-party kafka publisher
- And the kafka broker is started
+ And the Kafka server is started
And the topic "ConsumeKafkaTest" is initialized on the kafka broker
- When all instances start up
+ When the MiNiFi instance starts up
And two messages with content "Barbapapa" and "Anette Tison and Talus Taylor" is published to the "ConsumeKafkaTest" topic
- Then a flowfile with the content "Barbapapa,Anette Tison and Talus Taylor" is placed in the monitored directory in less than 45 seconds
+ Then a single file with the content "Barbapapa,Anette Tison and Talus Taylor" is placed in the "/tmp/output" directory in less than 45 seconds
Scenario Outline: The ConsumeKafka "Maximum Poll Records" property sets a limit on the messages processed in a single batch
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
- And a LogAttribute processor in the "kafka-consumer-flow" flow
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
+ And a LogAttribute processor
+ And LogAttribute is EVENT_DRIVEN
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "Max Poll Records" property of the ConsumeKafka processor is set to "<max poll records>"
- And the scheduling period of the ConsumeKafka processor is set to "<scheduling period>"
+ And the scheduling period of the ConsumeKafka processor is set to "5 sec"
And the scheduling period of the LogAttribute processor is set to "1 sec"
And the "FlowFiles To Log" property of the LogAttribute processor is set to "<max poll records>"
And the "success" relationship of the ConsumeKafka processor is connected to the LogAttribute
And the "success" relationship of the LogAttribute processor is connected to the PutFile
-
- And a kafka broker is set up in correspondence with the third-party kafka publisher
+ And PutFile's success relationship is auto-terminated
When all instances start up
And 1000 kafka messages are sent to the topic "ConsumeKafkaTest"
- Then after a wait of <polling time>, at least <min expected messages> and at most <max expected messages> flowfiles are produced and placed in the monitored directory
+ Then after a wait of 15 seconds, at least <min expected messages> and at most <max expected messages> files are produced and placed in the "/tmp/output" directory
Examples: Message batching
- | max poll records | scheduling period | polling time | min expected messages | max expected messages |
- | 3 | 5 sec | 15 seconds | 6 | 12 |
- | 6 | 5 sec | 15 seconds | 12 | 24 |
+ | max poll records | min expected messages | max expected messages |
+ | 3 | 6 | 12 |
+ | 6 | 12 | 24 |
Scenario: ConsumeKafka receives data via SSL
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
- And these processor properties are set:
- | processor name | property name | property value |
- | ConsumeKafka | Kafka Brokers | kafka-broker-${feature_id}:9093 |
- | ConsumeKafka | Security Protocol | ssl |
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | ConsumeKafka | Kafka Brokers | kafka-server-${scenario_id}:9093 |
+ | ConsumeKafka | Security Protocol | ssl |
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And an ssl context service is set up for ConsumeKafka
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
-
- And a kafka broker is set up in correspondence with the publisher flow
+ And PutFile's success relationship is auto-terminated
When all instances start up
And a message with content "Through the Looking-Glass" is published to the "ConsumeKafkaTest" topic
And a message with content "Lewis Carroll" is published to the "ConsumeKafkaTest" topic
- Then two flowfiles with the contents "Through the Looking-Glass" and "Lewis Carroll" are placed in the monitored directory in less than 60 seconds
+ Then files with contents "Through the Looking-Glass" and "Lewis Carroll" are placed in the "/tmp/output" directory in less than 60 seconds
Scenario: ConsumeKafka receives data via SASL SSL
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
- And these processor properties are set:
- | processor name | property name | property value |
- | ConsumeKafka | Kafka Brokers | kafka-broker-${feature_id}:9095 |
- | ConsumeKafka | Security Protocol | sasl_ssl |
- | ConsumeKafka | SASL Mechanism | PLAIN |
- | ConsumeKafka | Username | alice |
- | ConsumeKafka | Password | alice-secret |
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | ConsumeKafka | Kafka Brokers | kafka-server-${scenario_id}:9095 |
+ | ConsumeKafka | Security Protocol | sasl_ssl |
+ | ConsumeKafka | SASL Mechanism | PLAIN |
+ | ConsumeKafka | Username | alice |
+ | ConsumeKafka | Password | alice-secret |
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And an ssl context service is set up for ConsumeKafka
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
-
- And a kafka broker is set up in correspondence with the publisher flow
+ And PutFile's success relationship is auto-terminated
When all instances start up
And a message with content "Through the Looking-Glass" is published to the "ConsumeKafkaTest" topic
And a message with content "Lewis Carroll" is published to the "ConsumeKafkaTest" topic
- Then two flowfiles with the contents "Through the Looking-Glass" and "Lewis Carroll" are placed in the monitored directory in less than 60 seconds
+ Then files with contents "Through the Looking-Glass" and "Lewis Carroll" are placed in the "/tmp/output" directory in less than 45 seconds
Scenario: MiNiFi consumes data from a kafka topic via SASL PLAIN connection
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
- And these processor properties are set:
- | processor name | property name | property value |
- | ConsumeKafka | Kafka Brokers | kafka-broker-${feature_id}:9094 |
- | ConsumeKafka | Security Protocol | sasl_plaintext |
- | ConsumeKafka | SASL Mechanism | PLAIN |
- | ConsumeKafka | Username | alice |
- | ConsumeKafka | Password | alice-secret |
-
- And a kafka broker is set up in correspondence with the third-party kafka publisher
+ And these processor properties are set
+ | processor name | property name | property value |
+ | ConsumeKafka | Kafka Brokers | kafka-server-${scenario_id}:9094 |
+ | ConsumeKafka | Security Protocol | sasl_plaintext |
+ | ConsumeKafka | SASL Mechanism | PLAIN |
+ | ConsumeKafka | Username | alice |
+ | ConsumeKafka | Password | alice-secret |
When all instances start up
And a message with content "some test message" is published to the "ConsumeKafkaTest" topic
- Then at least one flowfile with the content "some test message" is placed in the monitored directory in less than 60 seconds
+ Then at least one file with the content "some test message" is placed in the "/tmp/output" directory in less than 60 seconds
Scenario Outline: MiNiFi commit policy tests without incoming flowfiles
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Topic Names" property of the ConsumeKafka processor is set to "ConsumeKafkaTest"
And the "Commit Offsets Policy" property of the ConsumeKafka processor is set to "<commit_policy>"
And the "Offset Reset" property of the ConsumeKafka processor is set to "<offset_reset>"
And the "Security Protocol" property of the ConsumeKafka processor is set to "plaintext"
And the "SASL Mechanism" property of the ConsumeKafka processor is set to "PLAIN"
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
- And a kafka broker is set up in correspondence with the third-party kafka publisher
- And the kafka broker is started
+ And the Kafka server is started
And the topic "ConsumeKafkaTest" is initialized on the kafka broker
When a message with content "Augustus" is published to the "ConsumeKafkaTest" topic
- And all other processes start up
+ And the MiNiFi instance starts up
And the Kafka consumer is registered in kafka broker
- Then exactly these flowfiles are in the monitored directory in less than 10 seconds: "<contents_after_augustus>"
+ Then exactly these files are in the "/tmp/output" directory in less than 10 seconds: "<contents_after_augustus>"
When a message with content "Tiberius" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory in less than 10 seconds: "<contents_after_tiberius>"
+ Then exactly these files are in the "/tmp/output" directory in less than 10 seconds: "<contents_after_tiberius>"
- When "kafka-consumer-flow" flow is stopped
+ When MiNiFi is stopped
And a message with content "Caligula" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory in less than 10 seconds: "<contents_after_caligula>"
+ Then exactly these files are in the "/tmp/output" directory in less than 10 seconds: "<contents_after_caligula>"
When a message with content "Claudius" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory in less than 10 seconds: "<contents_after_cladius>"
+ Then exactly these files are in the "/tmp/output" directory in less than 10 seconds: "<contents_after_cladius>"
- When "kafka-consumer-flow" flow is restarted
+ When MiNiFi is restarted
And the Kafka consumer is reregistered in kafka broker
And a message with content "Nero" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory in less than 10 seconds: "<contents_after_nero>"
+ Then exactly these files are in the "/tmp/output" directory in less than 10 seconds: "<contents_after_nero>"
Examples: No Commit
@@ -275,48 +287,50 @@ Feature: Receiving data from using Kafka streaming platform using ConsumeKafka
| Commit from incoming flowfiles | latest | | Tiberius | Tiberius | Tiberius | Tiberius,Nero |
| Commit from incoming flowfiles | earliest | Augustus | Augustus,Tiberius | Augustus,Tiberius | Augustus,Tiberius | Augustus,Tiberius,Augustus,Tiberius,Caligula,Claudius,Nero |
-
Scenario Outline: MiNiFi commit policy tests with incoming flowfiles
- Given a ConsumeKafka processor set up in a "kafka-consumer-flow" flow
- And there is a "processed" subdirectory in the monitored directory
- And there is a "committed" subdirectory in the monitored directory
+ Given a Kafka server is set up
+ And ConsumeKafka processor is set up to communicate with that server
And the "Topic Names" property of the ConsumeKafka processor is set to "ConsumeKafkaTest"
And the "Commit Offsets Policy" property of the ConsumeKafka processor is set to "<commit_policy>"
And the "Offset Reset" property of the ConsumeKafka processor is set to "<offset_reset>"
And the "Security Protocol" property of the ConsumeKafka processor is set to "plaintext"
And the "SASL Mechanism" property of the ConsumeKafka processor is set to "PLAIN"
- And a PutFile processor with the name "Consumed" and the "Directory" property set to "/tmp/output/processed" in the "kafka-consumer-flow" flow
- And a PutFile processor with the name "Committed" and the "Directory" property set to "/tmp/output/committed" in the "kafka-consumer-flow" flow
+ And a PutFile processor with the name "Consumed" and the "Directory" property set to "/tmp/output/processed"
+ And a PutFile processor with the name "Committed" and the "Directory" property set to "/tmp/output/committed"
+ And Consumed is EVENT_DRIVEN
+ And Committed is EVENT_DRIVEN
And the "success" relationship of the ConsumeKafka processor is connected to the Consumed
And the "success" relationship of the Consumed processor is connected to the ConsumeKafka
And the "committed" relationship of the ConsumeKafka processor is connected to the Committed
+ And Consumed's success relationship is auto-terminated
+ And Committed's success relationship is auto-terminated
- And a kafka broker is set up in correspondence with the third-party kafka publisher
- And the kafka broker is started
+ And the Kafka server is started
And the topic "ConsumeKafkaTest" is initialized on the kafka broker
When a message with content "Augustus" is published to the "ConsumeKafkaTest" topic
- And all other processes start up
+ And the MiNiFi instance starts up
And the Kafka consumer is registered in kafka broker
- Then exactly these flowfiles are in the monitored directory's "processed" subdirectory in less than 15 seconds: "<contents_after_augustus>"
+ Then exactly these files are in the "/tmp/output/processed" directory in less than 15 seconds: "<contents_after_augustus>"
When a message with content "Tiberius" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory's "processed" subdirectory in less than 15 seconds: "<contents_after_tiberius>"
- And exactly these flowfiles are in the monitored directory's "committed" subdirectory in less than 15 seconds: "<contents_after_tiberius>"
+ Then exactly these files are in the "/tmp/output/processed" directory in less than 15 seconds: "<contents_after_tiberius>"
+ And exactly these files are in the "/tmp/output/committed" directory in less than 15 seconds: "<contents_after_tiberius>"
- When "kafka-consumer-flow" flow is stopped
+ When MiNiFi is stopped
And a message with content "Caligula" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory's "processed" subdirectory in less than 15 seconds: "<contents_after_caligula>"
+ Then exactly these files are in the "/tmp/output/processed" directory in less than 15 seconds: "<contents_after_caligula>"
When a message with content "Claudius" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory's "processed" subdirectory in less than 15 seconds: "<contents_after_cladius>"
+ Then exactly these files are in the "/tmp/output/processed" directory in less than 15 seconds: "<contents_after_cladius>"
- When "kafka-consumer-flow" flow is restarted
+ When MiNiFi is restarted
And the Kafka consumer is reregistered in kafka broker
And a message with content "Nero" is published to the "ConsumeKafkaTest" topic
- Then exactly these flowfiles are in the monitored directory's "processed" subdirectory in less than 15 seconds: "<contents_after_nero>"
+ Then exactly these files are in the "/tmp/output/processed" directory in less than 15 seconds: "<contents_after_nero>"
Examples: Commit from FlowFiles (with incoming flowfiles)
| commit_policy | offset_reset | contents_after_augustus | contents_after_tiberius | contents_after_caligula | contents_after_cladius | contents_after_nero |
| Commit from incoming flowfiles | latest | | Tiberius | Tiberius | Tiberius | Tiberius,Caligula,Claudius,Nero |
| Commit from incoming flowfiles | earliest | Augustus | Augustus,Tiberius | Augustus,Tiberius | Augustus,Tiberius | Augustus,Tiberius,Caligula,Claudius,Nero |
+
diff --git a/docker/test/integration/minifi/processors/PublishKafka.py b/extensions/kafka/tests/features/environment.py
similarity index 51%
rename from docker/test/integration/minifi/processors/PublishKafka.py
rename to extensions/kafka/tests/features/environment.py
index a79e50728..99986cd33 100644
--- a/docker/test/integration/minifi/processors/PublishKafka.py
+++ b/extensions/kafka/tests/features/environment.py
@@ -12,24 +12,25 @@
# 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 minifi_test_framework.containers.docker_image_builder import DockerImageBuilder
+from minifi_test_framework.core.hooks import common_before_scenario
+from minifi_test_framework.core.hooks import common_after_scenario
-from ..core.Processor import Processor
+def before_all(context):
+ dockerfile = """
+ FROM python:3.13-slim-bookworm
+ RUN pip install confluent-kafka"""
+ builder = DockerImageBuilder(
+ image_tag="minifi-kafka-helper:latest",
+ dockerfile_content=dockerfile
+ )
+ builder.build()
-class PublishKafka(Processor):
- def __init__(self, context, schedule={'scheduling strategy': 'EVENT_DRIVEN'}):
- super(PublishKafka, self).__init__(
- context=context,
- clazz='PublishKafka',
- properties={
- 'Client Name': 'nghiaxlee',
- 'Known Brokers': f'kafka-broker-{context.feature_id}:9092',
- 'Topic Name': 'test',
- 'Batch Size': '10',
- 'Compress Codec': 'none',
- 'Delivery Guarantee': '1',
- 'Request Timeout': '10 sec',
- 'Message Timeout': '12 sec'},
- auto_terminate=['success'],
- schedule=schedule)
+def before_scenario(context, scenario):
+ common_before_scenario(context, scenario)
+
+
+def after_scenario(context, scenario):
+ common_after_scenario(context, scenario)
diff --git a/docker/test/integration/features/publish_kafka.feature b/extensions/kafka/tests/features/publishkafka.feature
similarity index 66%
rename from docker/test/integration/features/publish_kafka.feature
rename to extensions/kafka/tests/features/publishkafka.feature
index 33a1937ea..aac5439a3 100644
--- a/docker/test/integration/features/publish_kafka.feature
+++ b/extensions/kafka/tests/features/publishkafka.feature
@@ -19,45 +19,45 @@ Feature: Sending data to using Kafka streaming platform using PublishKafka
As a user of MiNiFi
I need to have PublishKafka processor
- Background:
- Given the content of "/tmp/output" is monitored
-
Scenario Outline: A MiNiFi instance transfers data to a kafka broker
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "test" is present in "/tmp/input"
+ Given a Kafka server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a directory at "/tmp/input" has a file with the content "test"
And a UpdateAttribute processor
- And these processor properties are set:
+ And UpdateAttribute is EVENT_DRIVEN
+ And these processor properties are set
| processor name | property name | property value |
| UpdateAttribute | kafka_require_num_acks | 1 |
| UpdateAttribute | kafka_message_key | unique_message_key_123 |
- And a PublishKafka processor set up to communicate with a kafka broker instance
- And these processor properties are set:
- | processor name | property name | property value |
- | PublishKafka | Topic Name | test |
- | PublishKafka | Delivery Guarantee | ${kafka_require_num_acks} |
- | PublishKafka | Request Timeout | 12 s |
- | PublishKafka | Message Timeout | 13 s |
- | PublishKafka | Known Brokers | kafka-broker-${feature_id}:${literal(9000):plus(92)} |
- | PublishKafka | Client Name | client_no_${literal(6):multiply(7)} |
- | PublishKafka | Kafka Key | ${kafka_message_key} |
- | PublishKafka | retry.backoff.ms | ${literal(2):multiply(25):multiply(3)} |
- | PublishKafka | Message Key Field | kafka.key |
- | PublishKafka | Compress Codec | <compress_codec> |
+ And a PublishKafka processor
+ And PublishKafka is EVENT_DRIVEN
+ And these processor properties are set
+ | processor name | property name | property value |
+ | PublishKafka | Topic Name | test |
+ | PublishKafka | Delivery Guarantee | ${kafka_require_num_acks} |
+ | PublishKafka | Request Timeout | 12 s |
+ | PublishKafka | Message Timeout | 13 s |
+ | PublishKafka | Known Brokers | kafka-server-${scenario_id}:${literal(9000):plus(92)} |
+ | PublishKafka | Client Name | client_no_${literal(6):multiply(7)} |
+ | PublishKafka | Kafka Key | ${kafka_message_key} |
+ | PublishKafka | retry.backoff.ms | ${literal(2):multiply(25):multiply(3)} |
+ | PublishKafka | Message Key Field | kafka.key |
+ | PublishKafka | Compress Codec | <compress_codec> |
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the UpdateAttribute
And the "success" relationship of the UpdateAttribute processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
And the "failure" relationship of the PublishKafka processor is connected to the PublishKafka
-
- And a kafka broker is set up in correspondence with the PublishKafka
+ And PutFile's success relationship is auto-terminated
When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
And the Minifi logs contain the following message: " is 'test'" in less than 60 seconds
And the Minifi logs contain the following message: "PublishKafka: request.required.acks [1]" in less than 10 seconds
And the Minifi logs contain the following message: "PublishKafka: request.timeout.ms [12000]" in less than 10 seconds
And the Minifi logs contain the following message: "PublishKafka: message.timeout.ms [13000]" in less than 10 seconds
- And the Minifi logs contain the following message: "PublishKafka: bootstrap.servers [kafka-broker-${feature_id}:9092]" in less than 10 seconds
+ And the Minifi logs contain the following message: "PublishKafka: bootstrap.servers [kafka-server-${scenario_id}:9092]" in less than 10 seconds
And the Minifi logs contain the following message: "PublishKafka: client.id [client_no_42]" in less than 10 seconds
And the Minifi logs contain the following message: "PublishKafka: Message Key [unique_message_key_123]" in less than 10 seconds
And the Minifi logs contain the following message: "PublishKafka: DynamicProperty: [retry.backoff.ms] -> [150]" in less than 10 seconds
@@ -73,156 +73,169 @@ Feature: Sending data to using Kafka streaming platform using PublishKafka
Scenario: PublishKafka sends flowfiles to failure when the broker is not available
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "no broker" is present in "/tmp/input"
- And a PublishKafka processor set up to communicate with a kafka broker instance
+ And a directory at "/tmp/input" has a file with the content "no broker"
+ And PublishKafka processor is set up to communicate with that server
+ And PublishKafka is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "failure" relationship of the PublishKafka processor is connected to the PutFile
+ And PutFile's success relationship is auto-terminated
When the MiNiFi instance starts up
- Then a flowfile with the content "no broker" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "no broker" is placed in the "/tmp/output" directory in less than 60 seconds
Scenario: PublishKafka sends can use SSL connect with security properties
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "test" is present in "/tmp/input"
- And a PublishKafka processor set up to communicate with a kafka broker instance
- And these processor properties are set:
- | processor name | property name | property value |
- | PublishKafka | Client Name | LMN |
- | PublishKafka | Known Brokers | kafka-broker-${feature_id}:9093 |
- | PublishKafka | Topic Name | test |
- | PublishKafka | Batch Size | 10 |
- | PublishKafka | Compress Codec | none |
- | PublishKafka | Delivery Guarantee | 1 |
- | PublishKafka | Request Timeout | 10 sec |
- | PublishKafka | Message Timeout | 12 sec |
- | PublishKafka | Security CA | /tmp/resources/root_ca.crt |
- | PublishKafka | Security Cert | /tmp/resources/minifi_client.crt |
- | PublishKafka | Security Private Key | /tmp/resources/minifi_client.key |
- | PublishKafka | Security Protocol | ssl |
+ Given a Kafka server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a directory at "/tmp/input" has a file with the content "test"
+ And PublishKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | PublishKafka | Client Name | LMN |
+ | PublishKafka | Known Brokers | kafka-server-${scenario_id}:9093 |
+ | PublishKafka | Topic Name | test |
+ | PublishKafka | Batch Size | 10 |
+ | PublishKafka | Compress Codec | none |
+ | PublishKafka | Delivery Guarantee | 1 |
+ | PublishKafka | Request Timeout | 10 sec |
+ | PublishKafka | Message Timeout | 12 sec |
+ | PublishKafka | Security CA | /tmp/resources/root_ca.crt |
+ | PublishKafka | Security Cert | /tmp/resources/minifi_client.crt |
+ | PublishKafka | Security Private Key | /tmp/resources/minifi_client.key |
+ | PublishKafka | Security Protocol | ssl |
+ And PublishKafka is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
And the "failure" relationship of the PublishKafka processor is connected to the PublishKafka
-
- And a kafka broker is set up in correspondence with the PublishKafka
+ And PutFile's success relationship is auto-terminated
When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
# We fallback to the flowfile's uuid as message key if the Kafka Key property is not set
And the Minifi logs match the following regex: "PublishKafka: Message Key \[[a-z0-9]{8}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{12}\]" in less than 10 seconds
Scenario: A MiNiFi instance transfers data to a kafka broker through SASL Plain security protocol
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "test" is present in "/tmp/input"
- And a PublishKafka processor set up to communicate with a kafka broker instance
- And these processor properties are set:
- | processor name | property name | property value |
- | PublishKafka | Topic Name | test |
- | PublishKafka | Request Timeout | 10 sec |
- | PublishKafka | Message Timeout | 12 sec |
- | PublishKafka | Known Brokers | kafka-broker-${feature_id}:9094 |
- | PublishKafka | Client Name | LMN |
- | PublishKafka | Security Protocol | sasl_plaintext |
- | PublishKafka | SASL Mechanism | PLAIN |
- | PublishKafka | Username | alice |
- | PublishKafka | Password | alice-secret |
+ Given a Kafka server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a directory at "/tmp/input" has a file with the content "test"
+ And PublishKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | PublishKafka | Topic Name | test |
+ | PublishKafka | Request Timeout | 10 sec |
+ | PublishKafka | Message Timeout | 12 sec |
+ | PublishKafka | Known Brokers | kafka-server-${scenario_id}:9094 |
+ | PublishKafka | Client Name | LMN |
+ | PublishKafka | Security Protocol | sasl_plaintext |
+ | PublishKafka | SASL Mechanism | PLAIN |
+ | PublishKafka | Username | alice |
+ | PublishKafka | Password | alice-secret |
+ And PublishKafka is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
And the "failure" relationship of the PublishKafka processor is connected to the PublishKafka
-
- And a kafka broker is set up in correspondence with the PublishKafka
+ And PutFile's success relationship is auto-terminated
When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
Scenario: PublishKafka sends can use SASL SSL connect with security properties
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "test" is present in "/tmp/input"
- And a PublishKafka processor set up to communicate with a kafka broker instance
- And these processor properties are set:
- | processor name | property name | property value |
- | PublishKafka | Client Name | LMN |
- | PublishKafka | Known Brokers | kafka-broker-${feature_id}:9095 |
- | PublishKafka | Topic Name | test |
- | PublishKafka | Batch Size | 10 |
- | PublishKafka | Compress Codec | none |
- | PublishKafka | Delivery Guarantee | 1 |
- | PublishKafka | Request Timeout | 10 sec |
- | PublishKafka | Message Timeout | 12 sec |
- | PublishKafka | Security CA | /tmp/resources/root_ca.crt |
- | PublishKafka | Security Cert | /tmp/resources/minifi_client.crt |
- | PublishKafka | Security Private Key | /tmp/resources/minifi_client.key |
- | PublishKafka | Security Protocol | sasl_ssl |
- | PublishKafka | SASL Mechanism | PLAIN |
- | PublishKafka | Username | alice |
- | PublishKafka | Password | alice-secret |
+ Given a Kafka server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a directory at "/tmp/input" has a file with the content "test"
+ And PublishKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | PublishKafka | Client Name | LMN |
+ | PublishKafka | Known Brokers | kafka-server-${scenario_id}:9095 |
+ | PublishKafka | Topic Name | test |
+ | PublishKafka | Batch Size | 10 |
+ | PublishKafka | Compress Codec | none |
+ | PublishKafka | Delivery Guarantee | 1 |
+ | PublishKafka | Request Timeout | 10 sec |
+ | PublishKafka | Message Timeout | 12 sec |
+ | PublishKafka | Security CA | /tmp/resources/root_ca.crt |
+ | PublishKafka | Security Cert | /tmp/resources/minifi_client.crt |
+ | PublishKafka | Security Private Key | /tmp/resources/minifi_client.key |
+ | PublishKafka | Security Protocol | sasl_ssl |
+ | PublishKafka | SASL Mechanism | PLAIN |
+ | PublishKafka | Username | alice |
+ | PublishKafka | Password | alice-secret |
+ And PublishKafka is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
And the "failure" relationship of the PublishKafka processor is connected to the PublishKafka
-
- And a kafka broker is set up in correspondence with the PublishKafka
+ And PutFile's success relationship is auto-terminated
When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
Scenario: PublishKafka sends can use SASL SSL connect with SSL Context
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "test" is present in "/tmp/input"
- And a PublishKafka processor set up to communicate with a kafka broker instance
- And these processor properties are set:
- | processor name | property name | property value |
- | PublishKafka | Client Name | LMN |
- | PublishKafka | Known Brokers | kafka-broker-${feature_id}:9095 |
- | PublishKafka | Topic Name | test |
- | PublishKafka | Batch Size | 10 |
- | PublishKafka | Compress Codec | none |
- | PublishKafka | Delivery Guarantee | 1 |
- | PublishKafka | Request Timeout | 10 sec |
- | PublishKafka | Message Timeout | 12 sec |
- | PublishKafka | Security Protocol | sasl_ssl |
- | PublishKafka | SASL Mechanism | PLAIN |
- | PublishKafka | Username | alice |
- | PublishKafka | Password | alice-secret |
+ Given a Kafka server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a directory at "/tmp/input" has a file with the content "test"
+ And PublishKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | PublishKafka | Client Name | LMN |
+ | PublishKafka | Known Brokers | kafka-server-${scenario_id}:9095 |
+ | PublishKafka | Topic Name | test |
+ | PublishKafka | Batch Size | 10 |
+ | PublishKafka | Compress Codec | none |
+ | PublishKafka | Delivery Guarantee | 1 |
+ | PublishKafka | Request Timeout | 10 sec |
+ | PublishKafka | Message Timeout | 12 sec |
+ | PublishKafka | Security Protocol | sasl_ssl |
+ | PublishKafka | SASL Mechanism | PLAIN |
+ | PublishKafka | Username | alice |
+ | PublishKafka | Password | alice-secret |
+ And PublishKafka is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And an ssl context service is set up for PublishKafka
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
And the "failure" relationship of the PublishKafka processor is connected to the PublishKafka
-
- And a kafka broker is set up in correspondence with the PublishKafka
+ And PutFile's success relationship is auto-terminated
When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
Scenario: PublishKafka sends can use SSL connect with SSL Context Service
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And a file with the content "test" is present in "/tmp/input"
- And a PublishKafka processor set up to communicate with a kafka broker instance
- And these processor properties are set:
- | processor name | property name | property value |
- | PublishKafka | Client Name | LMN |
- | PublishKafka | Known Brokers | kafka-broker-${feature_id}:9093 |
- | PublishKafka | Topic Name | test |
- | PublishKafka | Batch Size | 10 |
- | PublishKafka | Compress Codec | none |
- | PublishKafka | Delivery Guarantee | 1 |
- | PublishKafka | Request Timeout | 10 sec |
- | PublishKafka | Message Timeout | 12 sec |
- | PublishKafka | Security Protocol | ssl |
+ Given a Kafka server is set up
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a directory at "/tmp/input" has a file with the content "test"
+ And PublishKafka processor is set up to communicate with that server
+ And these processor properties are set
+ | processor name | property name | property value |
+ | PublishKafka | Client Name | LMN |
+ | PublishKafka | Known Brokers | kafka-server-${scenario_id}:9093 |
+ | PublishKafka | Topic Name | test |
+ | PublishKafka | Batch Size | 10 |
+ | PublishKafka | Compress Codec | none |
+ | PublishKafka | Delivery Guarantee | 1 |
+ | PublishKafka | Request Timeout | 10 sec |
+ | PublishKafka | Message Timeout | 12 sec |
+ | PublishKafka | Security Protocol | ssl |
+ And PublishKafka is EVENT_DRIVEN
And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And PutFile is EVENT_DRIVEN
And an ssl context service is set up for PublishKafka
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
And the "failure" relationship of the PublishKafka processor is connected to the PublishKafka
-
- And a kafka broker is set up in correspondence with the PublishKafka
+ And PutFile's success relationship is auto-terminated
When both instances start up
- Then a flowfile with the content "test" is placed in the monitored directory in less than 60 seconds
+ Then a single file with the content "test" is placed in the "/tmp/output" directory in less than 60 seconds
# We fallback to the flowfile's uuid as message key if the Kafka Key property is not set
And the Minifi logs match the following regex: "PublishKafka: Message Key \[[a-z0-9]{8}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{4}-[a-z0-9]{12}\]" in less than 10 seconds
diff --git a/extensions/kafka/tests/features/steps/kafka_server_container.py b/extensions/kafka/tests/features/steps/kafka_server_container.py
new file mode 100644
index 000000000..49c15a429
--- /dev/null
+++ b/extensions/kafka/tests/features/steps/kafka_server_container.py
@@ -0,0 +1,126 @@
+# 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 re
+import jks
+
+from OpenSSL import crypto
+from minifi_test_framework.core.helpers import wait_for_condition
+from minifi_test_framework.core.ssl_utils import make_server_cert
+from minifi_test_framework.containers.container import Container
+from minifi_test_framework.containers.file import File
+from minifi_test_framework.core.minifi_test_context import MinifiTestContext
+
+
+class KafkaServer(Container):
+ def __init__(self, test_context: MinifiTestContext):
+ super().__init__("apache/kafka:4.1.0", f"kafka-server-{test_context.scenario_id}", test_context.network)
+ self.environment.append("KAFKA_NODE_ID=1")
+ self.environment.append("KAFKA_PROCESS_ROLES=controller,broker")
+ self.environment.append("KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT")
+ self.environment.append("KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER")
+ self.environment.append("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1")
+ self.environment.append("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1")
+ self.environment.append("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1")
+ self.environment.append(f"KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka-server-{test_context.scenario_id}:9096")
+ self.environment.append(f"KAFKA_LISTENERS=PLAINTEXT://kafka-server-{test_context.scenario_id}:9092,SASL_PLAINTEXT://kafka-server-{test_context.scenario_id}:9094,SSL://kafka-server-{test_context.scenario_id}:9093,SASL_SSL://kafka-server-{test_context.scenario_id}:9095,CONTROLLER://kafka-server-{test_context.scenario_id}:9096")
+ self.environment.append(f"KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka-server-{test_context.scenario_id}:9092,SASL_PLAINTEXT://kafka-server-{test_context.scenario_id}:9094,SSL://kafka-server-{test_context.scenario_id}:9093,SASL_SSL://kafka-server-{test_context.scenario_id}:9095,CONTROLLER://kafka-server-{test_context.scenario_id}:9096")
+ self.environment.append("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,SSL:SSL,SASL_SSL:SASL_SSL,CONTROLLER:PLAINTEXT")
+ self.environment.append("KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN")
+ self.environment.append("KAFKA_SASL_ENABLED_MECHANISMS=PLAIN")
+ self.environment.append("KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kafka_jaas.conf -Dlog4j2.rootLogger.level=DEBUG -Dlog4j2.logger.org.apache.kafka.controller.level=DEBUG")
+ self.environment.append("KAFKA_SSL_PROTOCOL=TLS")
+ self.environment.append("KAFKA_SSL_ENABLED_PROTOCOLS=TLSv1.2")
+ self.environment.append("KAFKA_SSL_KEYSTORE_TYPE=JKS")
+ self.environment.append("KAFKA_SSL_KEYSTORE_FILENAME=kafka.keystore.jks")
+ self.environment.append("KAFKA_SSL_KEYSTORE_CREDENTIALS=credentials.conf")
+ self.environment.append("KAFKA_SSL_KEY_CREDENTIALS=credentials.conf")
+ self.environment.append("KAFKA_SSL_TRUSTSTORE_CREDENTIALS=credentials.conf")
+ self.environment.append("KAFKA_SSL_TRUSTSTORE_TYPE=JKS")
+ self.environment.append("KAFKA_SSL_TRUSTSTORE_FILENAME=kafka.truststore.jks")
+ self.environment.append("KAFKA_SSL_CLIENT_AUTH=none")
+
+ kafka_cert, kafka_key = make_server_cert(self.container_name, test_context.root_ca_cert, test_context.root_ca_key)
+
+ pke = jks.PrivateKeyEntry.new('kafka-broker-cert', [crypto.dump_certificate(crypto.FILETYPE_ASN1, kafka_cert)], crypto.dump_privatekey(crypto.FILETYPE_ASN1, kafka_key), 'rsa_raw')
+ server_keystore = jks.KeyStore.new('jks', [pke])
+ server_keystore_content = server_keystore.saves('abcdefgh')
+ self.files.append(File("/etc/kafka/secrets/kafka.keystore.jks", server_keystore_content, permissions=0o644))
+ self.files.append(File("/etc/kafka/secrets/credentials.conf", b'abcdefgh', permissions=0o644))
+
+ trusted_cert = jks.TrustedCertEntry.new(
+ 'root-ca', # Alias for the certificate
+ crypto.dump_certificate(crypto.FILETYPE_ASN1, test_context.root_ca_cert)
+ )
+
+ # Create a JKS keystore that will serve as a truststore with the trusted cert entry.
+ truststore = jks.KeyStore.new('jks', [trusted_cert])
+ truststore_content = truststore.saves('abcdefgh')
+ self.files.append(File("/etc/kafka/secrets/kafka.truststore.jks", truststore_content, permissions=0o644))
+
+ jaas_config_file_content = """
+KafkaServer {
+ org.apache.kafka.common.security.plain.PlainLoginModule required
+ username="admin"
+ password="admin-secret"
+ user_admin="admin-secret"
+ user_alice="alice-secret";
+};
+
+Client {
+ org.apache.kafka.common.security.plain.PlainLoginModule required
+ username="admin"
+ password="admin-secret";
+};
+"""
+ self.files.append(File("/opt/kafka/config/kafka_jaas.conf", jaas_config_file_content, permissions=0o644))
+
+ def deploy(self):
+ super().deploy()
+ finished_str = "Kafka Server started"
+ return wait_for_condition(
+ condition=lambda: finished_str in self.get_logs(),
+ timeout_seconds=60,
+ bail_condition=lambda: self.exited,
+ context=None)
+
+ def create_topic(self, topic_name: str):
+ (code, output) = self.exec_run(["/bin/sh", "-c", f"/opt/kafka/bin/kafka-topics.sh --create --topic '{topic_name}' --bootstrap-server '{self.container_name}':9092"])
+ logging.info(f"Create topic output: '{output}' with code {code}")
+ return code == 0
+
+ def produce_message(self, topic_name: str, message: str):
+ (code, output) = self.exec_run(["/bin/sh", "-c", f"echo '{message}' | /opt/kafka/bin/kafka-console-producer.sh --topic '{topic_name}' --bootstrap-server '{self.container_name}':9092"])
+ logging.info(f"Produce message output: '{output}' with code {code}")
+ return code == 0
+
+ def produce_message_with_key(self, topic_name: str, message: str, message_key: str):
+ (code, output) = self.exec_run(["/bin/sh", "-c", f" echo '{message_key}:{message}' | /opt/kafka/bin/kafka-console-producer.sh --property 'key.separator=:' --property 'parse.key=true' --topic '{topic_name}' --bootstrap-server '{self.container_name}':9092"])
+ logging.info(f"Produce message with key output: '{output}' with code {code}")
+ return code == 0
+
+ def run_python_in_kafka_helper_docker(self, command: str):
+ output = self.client.containers.run("minifi-kafka-helper:latest", ["python", "-c", command], remove=True, stdout=True, stderr=True, network=self.network.name)
+ logging.info(f"Run python in kafka helper docker output: '{output.decode('utf-8')}'")
+ return True
+
+ def wait_for_kafka_consumer_to_be_registered(self, expected_consumer_count: int):
+ message_regex = "Assignment received from leader.*for group docker_test_group"
+ return wait_for_condition(
+ condition=lambda: len(re.findall(message_regex, self.get_logs())) >= expected_consumer_count,
+ timeout_seconds=60,
+ bail_condition=lambda: self.exited,
+ context=None)
diff --git a/extensions/kafka/tests/features/steps/steps.py b/extensions/kafka/tests/features/steps/steps.py
new file mode 100644
index 000000000..474cc7bf1
--- /dev/null
+++ b/extensions/kafka/tests/features/steps/steps.py
@@ -0,0 +1,214 @@
+# 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 binascii
+import time
+
+from behave import step, when, given
+
+from minifi_test_framework.steps import checking_steps # noqa: F401
+from minifi_test_framework.steps import configuration_steps # noqa: F401
+from minifi_test_framework.steps import core_steps # noqa: F401
+from minifi_test_framework.steps import flow_building_steps # noqa: F401
+from minifi_test_framework.core.minifi_test_context import MinifiTestContext
+from minifi_test_framework.minifi.processor import Processor
+from kafka_server_container import KafkaServer
+
+
+@step("a Kafka server is set up")
+def step_impl(context):
+ context.containers["kafka-server"] = KafkaServer(context)
+
+
+@step("ConsumeKafka processor is set up to communicate with that server")
+def step_impl(context):
+ consume_kafka = Processor("ConsumeKafka", "ConsumeKafka")
+ consume_kafka.add_property("Kafka Brokers", f"kafka-server-{context.scenario_id}:9092")
+ consume_kafka.add_property("Topic Names", "ConsumeKafkaTest")
+ consume_kafka.add_property("Topic Name Format", "Names")
+ consume_kafka.add_property("Honor Transactions", "true")
+ consume_kafka.add_property("Group ID", "docker_test_group")
+ consume_kafka.add_property("Offset Reset", "earliest")
+ consume_kafka.add_property("Key Attribute Encoding", "UTF-8")
+ consume_kafka.add_property("Message Header Encoding", "UTF-8")
+ consume_kafka.add_property("Max Poll Time", "4 sec")
+ consume_kafka.add_property("Session Timeout", "6 sec")
+ context.get_or_create_default_minifi_container().flow_definition.add_processor(consume_kafka)
+
+
+@step("PublishKafka processor is set up to communicate with that server")
+def step_impl(context):
+ publish_kafka = Processor("PublishKafka", "PublishKafka")
+ publish_kafka.add_property("Known Brokers", f"kafka-server-{context.scenario_id}:9092")
+ publish_kafka.add_property("Client Name", "minifi-client")
+ publish_kafka.add_property("Topic Name", "test")
+ publish_kafka.add_property("Batch Size", "10")
+ publish_kafka.add_property("Compress Codec", "none")
+ publish_kafka.add_property("Delivery Guarantee", "1")
+ publish_kafka.add_property("Request Timeout", "10 sec")
+ publish_kafka.add_property("Message Timeout", "12 sec")
+ context.get_or_create_default_minifi_container().flow_definition.add_processor(publish_kafka)
+
+
+@step("the Kafka server is started")
+def step_impl(context: MinifiTestContext):
+ kafka_server_container = context.containers["kafka-server"]
+ assert isinstance(kafka_server_container, KafkaServer)
+ assert kafka_server_container.deploy()
+
+
+@step('the topic "{topic_name}" is initialized on the kafka broker')
+def step_impl(context: MinifiTestContext, topic_name: str):
+ kafka_server_container = context.containers["kafka-server"]
+ assert isinstance(kafka_server_container, KafkaServer)
+ assert kafka_server_container.create_topic(topic_name=topic_name) or kafka_server_container.log_app_output()
+
+
+@when('a message with content "{message}" is published to the "{topic_name}" topic')
+def step_impl(context: MinifiTestContext, message: str, topic_name: str):
+ kafka_server_container = context.containers["kafka-server"]
+ assert isinstance(kafka_server_container, KafkaServer)
+ assert kafka_server_container.produce_message(topic_name=topic_name, message=message) or kafka_server_container.log_app_output()
+
+
+@step('a message with content "{message}" is published to the "{topic_name}" topic with key "{key}"')
+def step_impl(context: MinifiTestContext, message: str, topic_name: str, key: str):
+ kafka_server_container = context.containers["kafka-server"]
+ assert isinstance(kafka_server_container, KafkaServer)
+ assert kafka_server_container.produce_message_with_key(topic_name=topic_name, message=message, message_key=key) or kafka_server_container.log_app_output()
+
+
+@given("the \"{property_name}\" property of the {processor_name} processor is set to match {key_attribute_encoding} encoded kafka message key \"{message_key}\"")
+def set_property_to_match_message_key(context, property_name, processor_name, key_attribute_encoding, message_key):
+ if key_attribute_encoding.lower() == "hex":
+ encoded_key = binascii.hexlify(message_key.encode("utf-8")).upper()
+ elif key_attribute_encoding.lower() == "(not set)":
+ encoded_key = message_key.encode("utf-8")
+ else:
+ encoded_key = message_key.encode(key_attribute_encoding)
+ filtering = "${kafka.key:equals('" + encoded_key.decode("utf-8") + "')}"
+ processor = context.get_or_create_default_minifi_container().flow_definition.get_processor(processor_name)
+ processor.add_property(property_name, filtering)
+
+
+@when("the publisher performs a {transaction_type} transaction publishing to the \"{topic_name}\" topic these messages: {messages}")
+def publish_to_topic_transaction_style(context, transaction_type, topic_name, messages):
+ if transaction_type == "SINGLE_COMMITTED_TRANSACTION":
+ python_code = f"""
+from confluent_kafka import Producer
+
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092", "transactional.id": "1001"}})
+producer.init_transactions()
+producer.begin_transaction()
+for content in "{messages}".split(", "):
+ producer.produce("{topic_name}", content.encode("utf-8"))
+producer.commit_transaction()
+producer.flush(10)
+ """
+ elif transaction_type == "TWO_SEPARATE_TRANSACTIONS":
+ python_code = f"""
+from confluent_kafka import Producer
+
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092", "transactional.id": "1001"}})
+producer.init_transactions()
+for content in "{messages}".split(", "):
+ producer.begin_transaction()
+ producer.produce("{topic_name}", content.encode("utf-8"))
+ producer.commit_transaction()
+producer.flush(10)
+ """
+ elif transaction_type == "NON_COMMITTED_TRANSACTION":
+ python_code = f"""
+from confluent_kafka import Producer
+
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092", "transactional.id": "1001"}})
+producer.init_transactions()
+producer.begin_transaction()
+for content in "{messages}".split(", "):
+ producer.produce("{topic_name}", content.encode("utf-8"))
+producer.flush(10)
+ """
+ elif transaction_type == "CANCELLED_TRANSACTION":
+ python_code = f"""
+from confluent_kafka import Producer
+
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092", "transactional.id": "1001"}})
+producer.init_transactions()
+producer.begin_transaction()
+for content in "{messages}".split(", "):
+ producer.produce("{topic_name}", content.encode("utf-8"))
+producer.flush(10)
+producer.abort_transaction()
+ """
+ else:
+ raise Exception("Unknown transaction type.")
+ assert context.containers["kafka-server"].run_python_in_kafka_helper_docker(python_code) or context.containers["kafka-server"].log_app_output()
+
+
+@when("a message with content \"{content}\" is published to the \"{topic_name}\" topic with headers \"{semicolon_separated_headers}\"")
+def publish_with_headers_to_topic(context, content, topic_name, semicolon_separated_headers):
+ python_code = f"""
+from confluent_kafka import Producer
+headers = []
+for header in "{semicolon_separated_headers}".split(";"):
+ kv = header.split(":")
+ headers.append((kv[0].strip(), kv[1].strip().encode("utf-8")))
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092"}})
+producer.produce("{topic_name}", "{content}".encode("utf-8"), headers=headers)
+producer.flush(10)
+ """
+ assert context.containers["kafka-server"].run_python_in_kafka_helper_docker(python_code) or context.containers["kafka-server"].log_app_output()
+
+
+@when("two messages with content \"{content_one}\" and \"{content_two}\" is published to the \"{topic_name}\" topic")
+def publish_two_messages_to_topic(context, content_one, content_two, topic_name):
+ python_code = f"""
+from confluent_kafka import Producer
+import uuid
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092"}})
+producer.produce("{topic_name}", "{content_one}")
+producer.produce("{topic_name}", "{content_two}")
+producer.flush(10)
+ """
+ assert context.containers["kafka-server"].run_python_in_kafka_helper_docker(python_code) or context.containers["kafka-server"].log_app_output()
+
+
+@when("{number_of_messages} kafka messages are sent to the topic \"{topic_name}\"")
+def publish_batch_to_topic(context, number_of_messages, topic_name):
+ python_code = f"""
+from confluent_kafka import Producer
+import uuid
+producer = Producer({{"bootstrap.servers": "kafka-server-{context.scenario_id}:9092"}})
+for i in range(0, int({number_of_messages})):
+ producer.produce("{topic_name}", str(uuid.uuid4()).encode("utf-8"))
+producer.flush(10)
+ """
+ assert context.containers["kafka-server"].run_python_in_kafka_helper_docker(python_code) or context.containers["kafka-server"].log_app_output()
+
+
+@when("the Kafka consumer is registered in kafka broker")
+def wait_for_consumer_registration(context):
+ assert context.containers["kafka-server"].wait_for_kafka_consumer_to_be_registered(1) or context.containers["kafka-server"].log_app_output()
+ # After the consumer is registered there is still some time needed for consumer-broker synchronization
+ # Unfortunately there are no additional log messages that could be checked for this
+ time.sleep(2)
+
+
+@when("the Kafka consumer is reregistered in kafka broker")
+def wait_for_consumer_registration(context):
+ assert context.containers["kafka-server"].wait_for_kafka_consumer_to_be_registered(2) or context.containers["kafka-server"].log_app_output()
+ # After the consumer is registered there is still some time needed for consumer-broker synchronization
+ # Unfortunately there are no additional log messages that could be checked for this
+ time.sleep(2)
diff --git a/extensions/standard-processors/tests/features/replace_text.feature b/extensions/standard-processors/tests/features/replace_text.feature
index 8a440a9b6..f8a3fdb05 100644
--- a/extensions/standard-processors/tests/features/replace_text.feature
+++ b/extensions/standard-processors/tests/features/replace_text.feature
@@ -29,7 +29,7 @@ Feature: Changing flowfile contents using the ReplaceText processor
And the "success" relationship of the GenerateFlowFile processor is connected to the ReplaceText
And the "success" relationship of the ReplaceText processor is connected to the PutFile
When the MiNiFi instance starts up
- Then there is a single file with "<output>" content in the "/tmp/output" directory in less than 10 seconds
+ Then a single file with the content "<output>" is placed in the "/tmp/output" directory in less than 10 seconds
Examples:
| input | replacement_strategy | search_value | replacement_value | output |
@@ -51,7 +51,7 @@ Feature: Changing flowfile contents using the ReplaceText processor
And the "success" relationship of the GetFile processor is connected to the ReplaceText
And the "success" relationship of the ReplaceText processor is connected to the PutFile
When the MiNiFi instance starts up
- Then there is a single file with "<output>" content in the "/tmp/output" directory in less than 10 seconds
+ Then a single file with the content "<output>" is placed in the "/tmp/output" directory in less than 10 seconds
Examples:
| input | evaluation_mode | output |
diff --git a/extensions/standard-processors/tests/features/routetext.feature b/extensions/standard-processors/tests/features/routetext.feature
index cfed4b9b1..5a27f1b19 100644
--- a/extensions/standard-processors/tests/features/routetext.feature
+++ b/extensions/standard-processors/tests/features/routetext.feature
@@ -34,4 +34,4 @@ Feature: Processing log files line-by-line using RouteText
And the "success" relationship of the UpdateInfo processor is connected to the PutFile
And the "success" relationship of the UpdateWarning processor is connected to the PutFile
When the MiNiFi instance starts up
- Then the contents of "/tmp/output" in less than 15 seconds are: "[INFO] one\n[INFO] three\n" and "[WARNING] two\n[WARNING] four\n"
+ Then files with contents "[INFO] one\n[INFO] three\n" and "[WARNING] two\n[WARNING] four\n" are placed in the "/tmp/output" directory in less than 15 seconds
diff --git a/extensions/standard-processors/tests/features/splittext.feature b/extensions/standard-processors/tests/features/splittext.feature
index 0dffd636f..aefe407af 100644
--- a/extensions/standard-processors/tests/features/splittext.feature
+++ b/extensions/standard-processors/tests/features/splittext.feature
@@ -28,7 +28,7 @@ Feature: Split input text line-by-line using SplitText
And the "success" relationship of the GetFile processor is connected to the SplitText
And the "splits" relationship of the SplitText processor is connected to the PutFile
When the MiNiFi instance starts up
- Then the contents of "/tmp/output" in less than 15 seconds are: "[HEADER]header line 1\n[HEADER]header line 2\nDATA LINE 1,[HEADER]header line 1\n[HEADER]header line 2\nDATA LINE 2,[HEADER]header line 1\n[HEADER]header line 2"
+ Then files with contents "[HEADER]header line 1\n[HEADER]header line 2\nDATA LINE 1,[HEADER]header line 1\n[HEADER]header line 2\nDATA LINE 2,[HEADER]header line 1\n[HEADER]header line 2" are placed in the "/tmp/output" directory in less than 15 seconds
Scenario: Split textfile containing header lines specified by header line marker characters
Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
@@ -42,4 +42,4 @@ Feature: Split input text line-by-line using SplitText
And SplitText's original relationship is auto-terminated
And PutFile's success relationship is auto-terminated
When the MiNiFi instance starts up
- Then the contents of "/tmp/output" in less than 15 seconds are: "[HEADER]header line 1\n[HEADER]header line 2\nA BIT LONGER DATA LINE 1,[HEADER]header line 1\n[HEADER]header line 2\nDATA 2\nDATA 3"
+ Then files with contents "[HEADER]header line 1\n[HEADER]header line 2\nA BIT LONGER DATA LINE 1,[HEADER]header line 1\n[HEADER]header line 2\nDATA 2\nDATA 3" are placed in the "/tmp/output" directory in less than 15 seconds
diff --git a/extensions/standard-processors/tests/features/syslog_listener.feature b/extensions/standard-processors/tests/features/syslog_listener.feature
index 324404b6d..2ea0ec990 100644
--- a/extensions/standard-processors/tests/features/syslog_listener.feature
+++ b/extensions/standard-processors/tests/features/syslog_listener.feature
@@ -42,4 +42,4 @@ Feature: Minifi C++ can act as a syslog listener
And PutFile's success relationship is auto-terminated
When both instances start up
- Then there are at least 1 files is in the "/tmp/output" directory in less than 10 seconds
+ Then at least 1 file is placed in the "/tmp/output" directory in less than 10 seconds