You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2023/03/09 21:25:52 UTC
[nifi] branch support/nifi-1.x updated: NIFI-11205 Added logging for deprecated Kafka 2_6 Kerberos properties
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new dab3d6bfee NIFI-11205 Added logging for deprecated Kafka 2_6 Kerberos properties
dab3d6bfee is described below
commit dab3d6bfee78b78d1207b6877901db15ec320fad
Author: Nandor Soma Abonyi <ab...@gmail.com>
AuthorDate: Sun Feb 26 16:09:16 2023 +0100
NIFI-11205 Added logging for deprecated Kafka 2_6 Kerberos properties
- Added Kerberos User Service property to KafkaRecordSink_2_6
This closes #6990
Signed-off-by: David Handermann <ex...@apache.org>
---
.../kafka/pubsub/ConsumeKafkaRecord_2_6.java | 4 ++
.../processors/kafka/pubsub/ConsumeKafka_2_6.java | 3 ++
.../kafka/pubsub/PublishKafkaRecord_2_6.java | 3 ++
.../processors/kafka/pubsub/PublishKafka_2_6.java | 3 ++
.../record/sink/kafka/KafkaRecordSink_2_6.java | 3 ++
.../nifi-kafka-bundle/nifi-kafka-shared/pom.xml | 4 ++
.../validation/KafkaDeprecationValidator.java | 57 ++++++++++++++++++++++
7 files changed, 77 insertions(+)
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_2_6.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_2_6.java
index 025a830552..1236775e1a 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_2_6.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaRecord_2_6.java
@@ -45,6 +45,7 @@ import org.apache.nifi.kafka.shared.property.KeyFormat;
import org.apache.nifi.kafka.shared.property.provider.KafkaPropertyProvider;
import org.apache.nifi.kafka.shared.property.provider.StandardKafkaPropertyProvider;
import org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
+import org.apache.nifi.kafka.shared.validation.KafkaDeprecationValidator;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
@@ -314,6 +315,7 @@ public class ConsumeKafkaRecord_2_6 extends AbstractProcessor implements KafkaCl
descriptors.add(SECURITY_PROTOCOL);
descriptors.add(SASL_MECHANISM);
descriptors.add(KERBEROS_CREDENTIALS_SERVICE);
+ descriptors.add(SELF_CONTAINED_KERBEROS_USER_SERVICE);
descriptors.add(KERBEROS_SERVICE_NAME);
descriptors.add(KERBEROS_PRINCIPAL);
descriptors.add(KERBEROS_KEYTAB);
@@ -368,6 +370,8 @@ public class ConsumeKafkaRecord_2_6 extends AbstractProcessor implements KafkaCl
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
+ KafkaDeprecationValidator.validate(getClass(), getIdentifier(), validationContext);
+
final KafkaClientCustomValidationFunction validationFunction = new KafkaClientCustomValidationFunction();
final Collection<ValidationResult> validationResults = validationFunction.apply(validationContext);
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_2_6.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_2_6.java
index 076cd474bb..ccb7d9628e 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_2_6.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafka_2_6.java
@@ -41,6 +41,7 @@ import org.apache.nifi.kafka.shared.component.KafkaClientComponent;
import org.apache.nifi.kafka.shared.property.provider.KafkaPropertyProvider;
import org.apache.nifi.kafka.shared.property.provider.StandardKafkaPropertyProvider;
import org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
+import org.apache.nifi.kafka.shared.validation.KafkaDeprecationValidator;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.components.ConfigVerificationResult;
@@ -319,6 +320,8 @@ public class ConsumeKafka_2_6 extends AbstractProcessor implements KafkaClientCo
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
+ KafkaDeprecationValidator.validate(getClass(), getIdentifier(), validationContext);
+
final KafkaClientCustomValidationFunction validationFunction = new KafkaClientCustomValidationFunction();
final Collection<ValidationResult> validationResults = validationFunction.apply(validationContext);
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafkaRecord_2_6.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafkaRecord_2_6.java
index 8b886584c5..f15a97e98a 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafkaRecord_2_6.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafkaRecord_2_6.java
@@ -45,6 +45,7 @@ import org.apache.nifi.kafka.shared.property.provider.KafkaPropertyProvider;
import org.apache.nifi.kafka.shared.component.KafkaPublishComponent;
import org.apache.nifi.kafka.shared.property.provider.StandardKafkaPropertyProvider;
import org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
+import org.apache.nifi.kafka.shared.validation.KafkaDeprecationValidator;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.DataUnit;
@@ -391,6 +392,8 @@ public class PublishKafkaRecord_2_6 extends AbstractProcessor implements KafkaPu
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
+ KafkaDeprecationValidator.validate(getClass(), getIdentifier(), validationContext);
+
final KafkaClientCustomValidationFunction validationFunction = new KafkaClientCustomValidationFunction();
final Collection<ValidationResult> results = validationFunction.apply(validationContext);
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_2_6.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_2_6.java
index 2a08b1efb5..ca1e381acf 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_2_6.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/PublishKafka_2_6.java
@@ -44,6 +44,7 @@ import org.apache.nifi.kafka.shared.property.provider.KafkaPropertyProvider;
import org.apache.nifi.kafka.shared.component.KafkaPublishComponent;
import org.apache.nifi.kafka.shared.property.provider.StandardKafkaPropertyProvider;
import org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
+import org.apache.nifi.kafka.shared.validation.KafkaDeprecationValidator;
import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.DataUnit;
@@ -348,6 +349,8 @@ public class PublishKafka_2_6 extends AbstractProcessor implements KafkaPublishC
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
+ KafkaDeprecationValidator.validate(getClass(), getIdentifier(), validationContext);
+
final KafkaClientCustomValidationFunction validationFunction = new KafkaClientCustomValidationFunction();
final Collection<ValidationResult> results = validationFunction.apply(validationContext);
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/record/sink/kafka/KafkaRecordSink_2_6.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/record/sink/kafka/KafkaRecordSink_2_6.java
index a71ef6218e..1156a12914 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/record/sink/kafka/KafkaRecordSink_2_6.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-2-6-processors/src/main/java/org/apache/nifi/record/sink/kafka/KafkaRecordSink_2_6.java
@@ -40,6 +40,7 @@ import org.apache.nifi.kafka.shared.property.provider.KafkaPropertyProvider;
import org.apache.nifi.kafka.shared.property.provider.StandardKafkaPropertyProvider;
import org.apache.nifi.kafka.shared.validation.KafkaClientCustomValidationFunction;
import org.apache.nifi.kafka.shared.validation.DynamicPropertyValidator;
+import org.apache.nifi.kafka.shared.validation.KafkaDeprecationValidator;
import org.apache.nifi.processor.DataUnit;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.record.sink.RecordSinkService;
@@ -171,6 +172,7 @@ public class KafkaRecordSink_2_6 extends AbstractControllerService implements Ka
properties.add(MESSAGE_HEADER_ENCODING);
properties.add(SECURITY_PROTOCOL);
properties.add(KERBEROS_CREDENTIALS_SERVICE);
+ properties.add(SELF_CONTAINED_KERBEROS_USER_SERVICE);
properties.add(KERBEROS_SERVICE_NAME);
properties.add(SSL_CONTEXT_SERVICE);
properties.add(MAX_REQUEST_SIZE);
@@ -198,6 +200,7 @@ public class KafkaRecordSink_2_6 extends AbstractControllerService implements Ka
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
+ KafkaDeprecationValidator.validate(getClass(), getIdentifier(), validationContext);
return new KafkaClientCustomValidationFunction().apply(validationContext);
}
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/pom.xml b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/pom.xml
index d4de0d1437..993e14da71 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/pom.xml
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/pom.xml
@@ -47,6 +47,10 @@
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-security-kerberos-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-deprecation-log</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-mock</artifactId>
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/validation/KafkaDeprecationValidator.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/validation/KafkaDeprecationValidator.java
new file mode 100644
index 0000000000..2555baf369
--- /dev/null
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-shared/src/main/java/org/apache/nifi/kafka/shared/validation/KafkaDeprecationValidator.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.kafka.shared.validation;
+
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.deprecation.log.DeprecationLogger;
+import org.apache.nifi.deprecation.log.DeprecationLoggerFactory;
+
+import static org.apache.nifi.kafka.shared.component.KafkaClientComponent.KERBEROS_CREDENTIALS_SERVICE;
+import static org.apache.nifi.kafka.shared.component.KafkaClientComponent.KERBEROS_KEYTAB;
+import static org.apache.nifi.kafka.shared.component.KafkaClientComponent.KERBEROS_PRINCIPAL;
+import static org.apache.nifi.kafka.shared.component.KafkaClientComponent.SELF_CONTAINED_KERBEROS_USER_SERVICE;
+
+public class KafkaDeprecationValidator {
+
+ public static void validate(Class<?> componentClass, final String identifier, final ValidationContext validationContext) {
+ final DeprecationLogger deprecationLogger = DeprecationLoggerFactory.getLogger(componentClass);
+
+ final PropertyValue credentialsServiceProperty = validationContext.getProperty(KERBEROS_CREDENTIALS_SERVICE);
+ final PropertyValue principalProperty = validationContext.getProperty(KERBEROS_PRINCIPAL).evaluateAttributeExpressions();
+ final PropertyValue keyTabProperty = validationContext.getProperty(KERBEROS_KEYTAB).evaluateAttributeExpressions();
+
+ if (credentialsServiceProperty.isSet()) {
+ deprecationLogger.warn("{}[id={}] [{}] Property should be replaced with [{}] Property",
+ componentClass.getSimpleName(),
+ identifier,
+ KERBEROS_CREDENTIALS_SERVICE.getDisplayName(),
+ SELF_CONTAINED_KERBEROS_USER_SERVICE.getDisplayName()
+ );
+ }
+
+ if (principalProperty.isSet() || keyTabProperty.isSet()) {
+ deprecationLogger.warn("{}[id={}] [{}] and [{}] Properties should be replaced with [{}] Property",
+ componentClass.getSimpleName(),
+ identifier,
+ KERBEROS_PRINCIPAL.getDisplayName(),
+ KERBEROS_KEYTAB.getDisplayName(),
+ SELF_CONTAINED_KERBEROS_USER_SERVICE.getDisplayName()
+ );
+ }
+ }
+}