You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by tu...@apache.org on 2023/05/16 08:00:32 UTC
[nifi] branch support/nifi-1.x updated: NIFI-11544 Fixed REL_FAILURE usage in AbstractIcebergProcessor
This is an automated email from the ASF dual-hosted git repository.
turcsanyi 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 8aa0d5549e NIFI-11544 Fixed REL_FAILURE usage in AbstractIcebergProcessor
8aa0d5549e is described below
commit 8aa0d5549ed034bd931902136d4390ad81167e6c
Author: krisztina-zsihovszki <zs...@gmail.com>
AuthorDate: Fri May 12 17:09:59 2023 +0200
NIFI-11544 Fixed REL_FAILURE usage in AbstractIcebergProcessor
This closes #7241.
Signed-off-by: Peter Turcsanyi <tu...@apache.org>
---
.../nifi/processors/iceberg/AbstractIcebergProcessor.java | 11 ++++++++---
.../java/org/apache/nifi/processors/iceberg/PutIceberg.java | 5 -----
.../processors/iceberg/writer/IcebergPartitionedWriter.java | 2 +-
3 files changed, 9 insertions(+), 9 deletions(-)
diff --git a/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/AbstractIcebergProcessor.java b/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/AbstractIcebergProcessor.java
index a4ec2ccf07..9f527344ec 100644
--- a/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/AbstractIcebergProcessor.java
+++ b/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/AbstractIcebergProcessor.java
@@ -26,6 +26,7 @@ import org.apache.nifi.kerberos.KerberosUserService;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.security.krb.KerberosLoginException;
import org.apache.nifi.security.krb.KerberosUser;
@@ -35,14 +36,13 @@ import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import static org.apache.nifi.hadoop.SecurityUtil.getUgiForKerberosUser;
-import static org.apache.nifi.processors.iceberg.PutIceberg.REL_FAILURE;
/**
* Base Iceberg processor class.
*/
public abstract class AbstractIcebergProcessor extends AbstractProcessor {
- static final PropertyDescriptor CATALOG = new PropertyDescriptor.Builder()
+ public static final PropertyDescriptor CATALOG = new PropertyDescriptor.Builder()
.name("catalog-service")
.displayName("Catalog Service")
.description("Specifies the Controller Service to use for handling references to table’s metadata files.")
@@ -50,13 +50,18 @@ public abstract class AbstractIcebergProcessor extends AbstractProcessor {
.required(true)
.build();
- static final PropertyDescriptor KERBEROS_USER_SERVICE = new PropertyDescriptor.Builder()
+ public static final PropertyDescriptor KERBEROS_USER_SERVICE = new PropertyDescriptor.Builder()
.name("kerberos-user-service")
.displayName("Kerberos User Service")
.description("Specifies the Kerberos User Controller Service that should be used for authenticating with Kerberos.")
.identifiesControllerService(KerberosUserService.class)
.build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("A FlowFile is routed to this relationship if the operation failed and retrying the operation will also fail, such as an invalid data or schema.")
+ .build();
+
private volatile KerberosUser kerberosUser;
private volatile UserGroupInformation ugi;
diff --git a/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/PutIceberg.java b/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/PutIceberg.java
index d85a590a08..e5142c3345 100644
--- a/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/PutIceberg.java
+++ b/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/PutIceberg.java
@@ -163,11 +163,6 @@ public class PutIceberg extends AbstractIcebergProcessor {
.description("A FlowFile is routed to this relationship after the data ingestion was successful.")
.build();
- static final Relationship REL_FAILURE = new Relationship.Builder()
- .name("failure")
- .description("A FlowFile is routed to this relationship if the data ingestion failed and retrying the operation will also fail, such as an invalid data or schema.")
- .build();
-
private static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(Arrays.asList(
RECORD_READER,
CATALOG,
diff --git a/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/writer/IcebergPartitionedWriter.java b/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/writer/IcebergPartitionedWriter.java
index fca8b92fc8..0b3f98e7c5 100644
--- a/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/writer/IcebergPartitionedWriter.java
+++ b/nifi-nar-bundles/nifi-iceberg-bundle/nifi-iceberg-processors/src/main/java/org/apache/nifi/processors/iceberg/writer/IcebergPartitionedWriter.java
@@ -35,7 +35,7 @@ public class IcebergPartitionedWriter extends PartitionedFanoutWriter<Record> {
private final PartitionKey partitionKey;
private final InternalRecordWrapper wrapper;
- IcebergPartitionedWriter(PartitionSpec spec, FileFormat format, FileAppenderFactory<Record> appenderFactory, OutputFileFactory fileFactory,
+ public IcebergPartitionedWriter(PartitionSpec spec, FileFormat format, FileAppenderFactory<Record> appenderFactory, OutputFileFactory fileFactory,
FileIO io, long targetFileSize, Schema schema) {
super(spec, format, appenderFactory, fileFactory, io, targetFileSize);
this.partitionKey = new PartitionKey(spec, schema);