granthenke commented on a change in pull request #3387: NIFI-6009 ScanKudu
Processor & KuduPut Processor Delete Operation
URL: https://github.com/apache/nifi/pull/3387#discussion_r280568966
##########
File path:
nifi-nar-bundles/nifi-kudu-bundle/nifi-kudu-processors/src/main/java/org/apache/nifi/processors/kudu/AbstractKuduProcessor.java
##########
@@ -0,0 +1,434 @@
+package org.apache.nifi.processors.kudu;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.Schema;
+import org.apache.kudu.Type;
+import org.apache.kudu.client.AsyncKuduClient;
+import org.apache.kudu.client.AsyncKuduScanner;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduTable;
+import org.apache.kudu.client.KuduSession;
+import org.apache.kudu.client.KuduPredicate;
+import org.apache.kudu.client.KuduException;
+import org.apache.kudu.client.OperationResponse;
+import org.apache.kudu.client.PartialRow;
+import org.apache.kudu.client.RowError;
+import org.apache.kudu.client.RowResult;
+import org.apache.kudu.client.SessionConfiguration;
+import org.apache.kudu.client.Delete;
+import org.apache.kudu.client.Insert;
+import org.apache.kudu.client.Upsert;
+import org.apache.kudu.client.Update;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.PropertyDescriptor.Builder;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.kerberos.KerberosCredentialsService;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processors.kudu.io.ResultHandler;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.security.krb.KerberosAction;
+import org.apache.nifi.security.krb.KerberosKeytabUser;
+import org.apache.nifi.security.krb.KerberosUser;
+import org.apache.nifi.serialization.record.Record;
+
+import javax.security.auth.login.LoginException;
+import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+public abstract class AbstractKuduProcessor extends AbstractProcessor {
+
+ static final PropertyDescriptor KUDU_MASTERS = new Builder()
+ .name("Kudu Masters")
+ .description("Comma separated addresses of the Kudu masters to
connect to.")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+
+ static final PropertyDescriptor KERBEROS_CREDENTIALS_SERVICE = new
Builder()
+ .name("kerberos-credentials-service")
+ .displayName("Kerberos Credentials Service")
+ .description("Specifies the Kerberos Credentials to use for
authentication")
+ .required(false)
+ .identifiesControllerService(KerberosCredentialsService.class)
+ .build();
+
+ static final PropertyDescriptor KUDU_OPERATION_TIMEOUT_MS = new Builder()
+ .name("kudu-operations-timeout-ms")
+ .displayName("Kudu Operation Timeout in MS")
+ .description("Default timeout used for user operations (using
sessions and scanners)")
+ .required(false)
+ .defaultValue(String.valueOf(AsyncKuduClient.
DEFAULT_OPERATION_TIMEOUT_MS))
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+
+ static final PropertyDescriptor KUDU_SOCKET_OPERATION_TIMEOUT_MS = new
Builder()
+ .name("kudu-socket-operation-timeout-ms")
+ .displayName("Kudu Socket Operation Timeout in MS")
+ .description("Default timeout used for user operations")
+ .required(false)
+
.defaultValue(String.valueOf(AsyncKuduClient.DEFAULT_SOCKET_READ_TIMEOUT_MS))
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+
.expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+ .build();
+
+ protected KuduClient kuduClient;
+
+ private AsyncKuduClient asyncKuduClient;
+
+ private volatile KerberosUser kerberosUser;
+
+ public KerberosUser getKerberosUser() {
+ return this.kerberosUser;
+ }
+
+ public KuduClient getKuduClient() {
+ return this.kuduClient;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session)
throws ProcessException {
+
+ try {
+ this.asyncKuduClient = createAsyncKuduClient(context);
+ } catch (LoginException e){
+ e.printStackTrace();
+ }
+
+ this.kuduClient = this.asyncKuduClient.syncClient();
+
+ if(this.kuduClient != null) {
+ try {
+ this.kuduClient.getTablesList();
+ } catch (KuduException e) {
+ e.printStackTrace();
+ }
+ }
+ doOnTrigger(context, session);
+ }
+
+ private AsyncKuduClient createAsyncKuduClient(ProcessContext context)
throws LoginException {
+ final String kuduMasters =
context.getProperty(KUDU_MASTERS).evaluateAttributeExpressions().getValue();
+ final KerberosCredentialsService credentialsService =
context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
+
+ if (credentialsService == null) {
+ return buildClient(kuduMasters, context);
+ }
+
+ final String keytab = credentialsService.getKeytab();
+ final String principal = credentialsService.getPrincipal();
+ kerberosUser = loginKerberosUser(principal, keytab);
+
+ final KerberosAction<AsyncKuduClient> kerberosAction = new
KerberosAction<>(kerberosUser, () -> buildClient(kuduMasters, context),
getLogger());
+ return kerberosAction.execute();
+ }
+
+
+ protected AsyncKuduClient buildClient(final String masters, final
ProcessContext context) {
+ final String operationTimeout =
context.getProperty(KUDU_OPERATION_TIMEOUT_MS).evaluateAttributeExpressions().getValue();
+ final String adminOperationTimeout =
context.getProperty(KUDU_SOCKET_OPERATION_TIMEOUT_MS).evaluateAttributeExpressions().getValue();
+
+ return new AsyncKuduClient.AsyncKuduClientBuilder(masters)
Review comment:
I suggest building a standard KuduClient via the KuduClientBuilder instead.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
