You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2020/03/04 16:21:47 UTC
[nifi] branch master updated: NIFI-7025: Wrap Hive 3 calls with
UGI.doAs Updated PutHive3Streaming to wrap calls to Hive in UGI.doAs
methods Fixed misleading logging message after the principal has been
authenticated with the KDC When connecting to unsecured Hive 3,
a UGI with "simple" auth will be used
This is an automated email from the ASF dual-hosted git repository.
mattyb149 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/master by this push:
new 4b6de8d NIFI-7025: Wrap Hive 3 calls with UGI.doAs Updated PutHive3Streaming to wrap calls to Hive in UGI.doAs methods Fixed misleading logging message after the principal has been authenticated with the KDC When connecting to unsecured Hive 3, a UGI with "simple" auth will be used
4b6de8d is described below
commit 4b6de8d164a2fe52d03fe06e751e2ece4ce7c680
Author: jstorck <jt...@gmail.com>
AuthorDate: Tue Mar 3 20:03:59 2020 -0500
NIFI-7025: Wrap Hive 3 calls with UGI.doAs
Updated PutHive3Streaming to wrap calls to Hive in UGI.doAs methods
Fixed misleading logging message after the principal has been authenticated with the KDC
When connecting to unsecured Hive 3, a UGI with "simple" auth will be used
Signed-off-by: Matthew Burgess <ma...@apache.org>
This closes #4108
---
.../nifi/processors/hive/PutHive3Streaming.java | 317 +++++++++++----------
.../processors/hive/TestPutHive3Streaming.java | 12 +-
2 files changed, 184 insertions(+), 145 deletions(-)
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java
index a1123d2..23b873f 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/main/java/org/apache/nifi/processors/hive/PutHive3Streaming.java
@@ -67,8 +67,10 @@ import org.apache.nifi.util.hive.HiveOptions;
import org.apache.nifi.util.hive.HiveUtils;
import org.apache.nifi.util.hive.ValidationResources;
+import javax.security.auth.login.LoginException;
import java.io.IOException;
import java.io.InputStream;
+import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -321,7 +323,7 @@ public class PutHive3Streaming extends AbstractProcessor {
}
@OnScheduled
- public void setup(final ProcessContext context) {
+ public void setup(final ProcessContext context) throws IOException {
ComponentLog log = getLogger();
rollbackOnFailure = context.getProperty(ROLLBACK_ON_FAILURE).asBoolean();
@@ -368,9 +370,9 @@ public class PutHive3Streaming extends AbstractProcessor {
throw new ProcessException(ae);
}
- log.info("Successfully logged in as principal {} with keytab {}", new Object[]{resolvedPrincipal, resolvedKeytab});
+ log.info("Successfully logged in as principal " + resolvedPrincipal);
} else {
- ugi = null;
+ ugi = SecurityUtil.loginSimple(hiveConfig);
kerberosUserReference.set(null);
}
@@ -381,172 +383,181 @@ public class PutHive3Streaming extends AbstractProcessor {
}
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
-
- final RecordReaderFactory recordReaderFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
- final String dbName = context.getProperty(DB_NAME).evaluateAttributeExpressions(flowFile).getValue();
- final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
-
- final ComponentLog log = getLogger();
- String metastoreURIs = null;
- if (context.getProperty(METASTORE_URI).isSet()) {
- metastoreURIs = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue();
- if (StringUtils.isEmpty(metastoreURIs)) {
- // Shouldn't be empty at this point, log an error, penalize the flow file, and return
- log.error("The '" + METASTORE_URI.getDisplayName() + "' property evaluated to null or empty, penalizing flow file, routing to failure");
- session.transfer(session.penalize(flowFile), REL_FAILURE);
+ getUgi().doAs((PrivilegedAction<Void>) () -> {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return null;
}
- }
-
- final String staticPartitionValuesString = context.getProperty(STATIC_PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue();
- final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean();
-
- // Override the Hive Metastore URIs in the config if set by the user
- if (metastoreURIs != null) {
- hiveConfig.set(MetastoreConf.ConfVars.THRIFT_URIS.getHiveName(), metastoreURIs);
- }
- HiveOptions o = new HiveOptions(metastoreURIs, dbName, tableName)
- .withHiveConf(hiveConfig)
- .withCallTimeout(callTimeout)
- .withStreamingOptimizations(!disableStreamingOptimizations);
+ final RecordReaderFactory recordReaderFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final String dbName = context.getProperty(DB_NAME).evaluateAttributeExpressions(flowFile).getValue();
+ final String tableName = context.getProperty(TABLE_NAME).evaluateAttributeExpressions(flowFile).getValue();
+
+ final ComponentLog log = getLogger();
+ String metastoreURIs = null;
+ if (context.getProperty(METASTORE_URI).isSet()) {
+ metastoreURIs = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isEmpty(metastoreURIs)) {
+ // Shouldn't be empty at this point, log an error, penalize the flow file, and return
+ log.error("The '" + METASTORE_URI.getDisplayName() + "' property evaluated to null or empty, penalizing flow file, routing to failure");
+ session.transfer(session.penalize(flowFile), REL_FAILURE);
+ }
+ }
- if (!StringUtils.isEmpty(staticPartitionValuesString)) {
- List<String> staticPartitionValues = Arrays.stream(staticPartitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList());
- o = o.withStaticPartitionValues(staticPartitionValues);
- }
+ final String staticPartitionValuesString = context.getProperty(STATIC_PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue();
+ final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean();
- if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
- final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
- o = o.withKerberosPrincipal(credentialsService.getPrincipal()).withKerberosKeytab(credentialsService.getKeytab());
- }
+ // Override the Hive Metastore URIs in the config if set by the user
+ if (metastoreURIs != null) {
+ hiveConfig.set(MetastoreConf.ConfVars.THRIFT_URIS.getHiveName(), metastoreURIs);
+ }
- final HiveOptions options = o;
+ HiveOptions o = new HiveOptions(metastoreURIs, dbName, tableName)
+ .withHiveConf(hiveConfig)
+ .withCallTimeout(callTimeout)
+ .withStreamingOptimizations(!disableStreamingOptimizations);
- // Store the original class loader, then explicitly set it to this class's classloader (for use by the Hive Metastore)
- ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader();
- Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
+ if (!StringUtils.isEmpty(staticPartitionValuesString)) {
+ List<String> staticPartitionValues = Arrays.stream(staticPartitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList());
+ o = o.withStaticPartitionValues(staticPartitionValues);
+ }
- StreamingConnection hiveStreamingConnection = null;
+ if (SecurityUtil.isSecurityEnabled(hiveConfig)) {
+ final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class);
+ final String explicitPrincipal = context.getProperty(KERBEROS_PRINCIPAL).evaluateAttributeExpressions().getValue();
+ final String resolvedPrincipal;
+ if (credentialsService != null) {
+ resolvedPrincipal = credentialsService.getPrincipal();
+ o = o.withKerberosKeytab(credentialsService.getKeytab());
+ } else resolvedPrincipal = explicitPrincipal;
+ o = o.withKerberosPrincipal(resolvedPrincipal);
+ }
- try {
- final RecordReader reader;
-
- try(final InputStream in = session.read(flowFile)) {
- // if we fail to create the RecordReader then we want to route to failure, so we need to
- // handle this separately from the other IOExceptions which normally route to retry
- try {
- reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger());
- } catch (Exception e) {
- throw new RecordReaderFactoryException("Unable to create RecordReader", e);
- }
+ final HiveOptions options = o;
- hiveStreamingConnection = makeStreamingConnection(options, reader);
+ // Store the original class loader, then explicitly set it to this class's classloader (for use by the Hive Metastore)
+ ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader());
- // Write records to Hive streaming, then commit and close
- hiveStreamingConnection.beginTransaction();
- hiveStreamingConnection.write(in);
- hiveStreamingConnection.commitTransaction();
- in.close();
+ StreamingConnection hiveStreamingConnection = null;
- Map<String, String> updateAttributes = new HashMap<>();
- updateAttributes.put(HIVE_STREAMING_RECORD_COUNT_ATTR, Long.toString(hiveStreamingConnection.getConnectionStats().getRecordsWritten()));
- updateAttributes.put(ATTR_OUTPUT_TABLES, options.getQualifiedTableName());
- flowFile = session.putAllAttributes(flowFile, updateAttributes);
- session.getProvenanceReporter().send(flowFile, hiveStreamingConnection.getMetastoreUri());
- } catch (TransactionError te) {
- if (rollbackOnFailure) {
- throw new ProcessException(te.getLocalizedMessage(), te);
- } else {
- throw new ShouldRetryException(te.getLocalizedMessage(), te);
+ try {
+ final RecordReader reader;
+
+ try(final InputStream in = session.read(flowFile)) {
+ // if we fail to create the RecordReader then we want to route to failure, so we need to
+ // handle this separately from the other IOExceptions which normally route to retry
+ try {
+ reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger());
+ } catch (Exception e) {
+ throw new RecordReaderFactoryException("Unable to create RecordReader", e);
+ }
+
+ hiveStreamingConnection = makeStreamingConnection(options, reader);
+
+ // Write records to Hive streaming, then commit and close
+ hiveStreamingConnection.beginTransaction();
+ hiveStreamingConnection.write(in);
+ hiveStreamingConnection.commitTransaction();
+ in.close();
+
+ Map<String, String> updateAttributes = new HashMap<>();
+ updateAttributes.put(HIVE_STREAMING_RECORD_COUNT_ATTR, Long.toString(hiveStreamingConnection.getConnectionStats().getRecordsWritten()));
+ updateAttributes.put(ATTR_OUTPUT_TABLES, options.getQualifiedTableName());
+ flowFile = session.putAllAttributes(flowFile, updateAttributes);
+ session.getProvenanceReporter().send(flowFile, hiveStreamingConnection.getMetastoreUri());
+ } catch (TransactionError te) {
+ if (rollbackOnFailure) {
+ throw new ProcessException(te.getLocalizedMessage(), te);
+ } else {
+ throw new ShouldRetryException(te.getLocalizedMessage(), te);
+ }
+ } catch (RecordReaderFactoryException rrfe) {
+ if (rollbackOnFailure) {
+ throw new ProcessException(rrfe);
+ } else {
+ log.error(
+ "Failed to create {} for {} - routing to failure",
+ new Object[]{RecordReader.class.getSimpleName(), flowFile},
+ rrfe
+ );
+ session.transfer(flowFile, REL_FAILURE);
+ return null;
+ }
}
- } catch (RecordReaderFactoryException rrfe) {
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) {
if (rollbackOnFailure) {
- throw new ProcessException(rrfe);
+ if (hiveStreamingConnection != null) {
+ abortConnection(hiveStreamingConnection);
+ }
+ throw new ProcessException(e.getLocalizedMessage(), e);
} else {
+ Map<String, String> updateAttributes = new HashMap<>();
+ final String recordCountAttribute = (hiveStreamingConnection != null) ? Long.toString(hiveStreamingConnection.getConnectionStats().getRecordsWritten()) : "0";
+ updateAttributes.put(HIVE_STREAMING_RECORD_COUNT_ATTR, recordCountAttribute);
+ updateAttributes.put(ATTR_OUTPUT_TABLES, options.getQualifiedTableName());
+ flowFile = session.putAllAttributes(flowFile, updateAttributes);
log.error(
- "Failed to create {} for {} - routing to failure",
- new Object[]{RecordReader.class.getSimpleName(), flowFile},
- rrfe
+ "Exception while processing {} - routing to failure",
+ new Object[]{flowFile},
+ e
);
session.transfer(flowFile, REL_FAILURE);
- return;
}
- }
- session.transfer(flowFile, REL_SUCCESS);
- } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) {
- if (rollbackOnFailure) {
+ } catch (DiscontinuedException e) {
+ // The input FlowFile processing is discontinued. Keep it in the input queue.
+ getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e);
+ session.transfer(flowFile, Relationship.SELF);
+ } catch (ConnectionError ce) {
+ // If we can't connect to the metastore, yield the processor
+ context.yield();
+ throw new ProcessException("A connection to metastore cannot be established", ce);
+ } catch (ShouldRetryException e) {
+ // This exception is already a result of adjusting an error, so simply transfer the FlowFile to retry. Still need to abort the txn
+ getLogger().error(e.getLocalizedMessage(), e);
if (hiveStreamingConnection != null) {
abortConnection(hiveStreamingConnection);
}
- throw new ProcessException(e.getLocalizedMessage(), e);
- } else {
- Map<String, String> updateAttributes = new HashMap<>();
- final String recordCountAttribute = (hiveStreamingConnection != null) ? Long.toString(hiveStreamingConnection.getConnectionStats().getRecordsWritten()) : "0";
- updateAttributes.put(HIVE_STREAMING_RECORD_COUNT_ATTR, recordCountAttribute);
- updateAttributes.put(ATTR_OUTPUT_TABLES, options.getQualifiedTableName());
- flowFile = session.putAllAttributes(flowFile, updateAttributes);
- log.error(
- "Exception while processing {} - routing to failure",
- new Object[]{flowFile},
- e
- );
- session.transfer(flowFile, REL_FAILURE);
- }
- } catch (DiscontinuedException e) {
- // The input FlowFile processing is discontinued. Keep it in the input queue.
- getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e);
- session.transfer(flowFile, Relationship.SELF);
- } catch (ConnectionError ce) {
- // If we can't connect to the metastore, yield the processor
- context.yield();
- throw new ProcessException("A connection to metastore cannot be established", ce);
- } catch (ShouldRetryException e) {
- // This exception is already a result of adjusting an error, so simply transfer the FlowFile to retry. Still need to abort the txn
- getLogger().error(e.getLocalizedMessage(), e);
- if (hiveStreamingConnection != null) {
- abortConnection(hiveStreamingConnection);
- }
- flowFile = session.penalize(flowFile);
- session.transfer(flowFile, REL_RETRY);
- } catch (StreamingException se) {
- // Handle all other exceptions. These are often record-based exceptions (since Hive will throw a subclass of the exception caught above)
- Throwable cause = se.getCause();
- if (cause == null) cause = se;
- // This is a failure on the incoming data, rollback on failure if specified; otherwise route to failure after penalizing (and abort txn in any case)
- if (rollbackOnFailure) {
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_RETRY);
+ } catch (StreamingException se) {
+ // Handle all other exceptions. These are often record-based exceptions (since Hive will throw a subclass of the exception caught above)
+ Throwable cause = se.getCause();
+ if (cause == null) cause = se;
+ // This is a failure on the incoming data, rollback on failure if specified; otherwise route to failure after penalizing (and abort txn in any case)
+ if (rollbackOnFailure) {
+ if (hiveStreamingConnection != null) {
+ abortConnection(hiveStreamingConnection);
+ }
+ throw new ProcessException(cause.getLocalizedMessage(), cause);
+ } else {
+ flowFile = session.penalize(flowFile);
+ Map<String, String> updateAttributes = new HashMap<>();
+ final String recordCountAttribute = (hiveStreamingConnection != null) ? Long.toString(hiveStreamingConnection.getConnectionStats().getRecordsWritten()) : "0";
+ updateAttributes.put(HIVE_STREAMING_RECORD_COUNT_ATTR, recordCountAttribute);
+ updateAttributes.put(ATTR_OUTPUT_TABLES, options.getQualifiedTableName());
+ flowFile = session.putAllAttributes(flowFile, updateAttributes);
+ log.error(
+ "Exception while trying to stream {} to hive - routing to failure",
+ new Object[]{flowFile},
+ se
+ );
+ session.transfer(flowFile, REL_FAILURE);
+ }
+
+ } catch (Throwable t) {
if (hiveStreamingConnection != null) {
abortConnection(hiveStreamingConnection);
}
- throw new ProcessException(cause.getLocalizedMessage(), cause);
- } else {
- flowFile = session.penalize(flowFile);
- Map<String, String> updateAttributes = new HashMap<>();
- final String recordCountAttribute = (hiveStreamingConnection != null) ? Long.toString(hiveStreamingConnection.getConnectionStats().getRecordsWritten()) : "0";
- updateAttributes.put(HIVE_STREAMING_RECORD_COUNT_ATTR, recordCountAttribute);
- updateAttributes.put(ATTR_OUTPUT_TABLES, options.getQualifiedTableName());
- flowFile = session.putAllAttributes(flowFile, updateAttributes);
- log.error(
- "Exception while trying to stream {} to hive - routing to failure",
- new Object[]{flowFile},
- se
- );
- session.transfer(flowFile, REL_FAILURE);
- }
-
- } catch (Throwable t) {
- if (hiveStreamingConnection != null) {
- abortConnection(hiveStreamingConnection);
+ throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t);
+ } finally {
+ closeConnection(hiveStreamingConnection);
+ // Restore original class loader, might not be necessary but is good practice since the processor task changed it
+ Thread.currentThread().setContextClassLoader(originalClassloader);
}
- throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t);
- } finally {
- closeConnection(hiveStreamingConnection);
- // Restore original class loader, might not be necessary but is good practice since the processor task changed it
- Thread.currentThread().setContextClassLoader(originalClassloader);
- }
+ return null;
+ });
}
StreamingConnection makeStreamingConnection(HiveOptions options, RecordReader reader) throws StreamingException {
@@ -623,5 +634,23 @@ public class PutHive3Streaming extends AbstractProcessor {
super(message, cause);
}
}
+
+ UserGroupInformation getUgi() {
+ getLogger().trace("getting UGI instance");
+ if (kerberosUserReference.get() != null) {
+ // if there's a KerberosUser associated with this UGI, check the TGT and relogin if it is close to expiring
+ KerberosUser kerberosUser = kerberosUserReference.get();
+ getLogger().debug("kerberosUser is " + kerberosUser);
+ try {
+ getLogger().debug("checking TGT on kerberosUser [{}]", new Object[] {kerberosUser});
+ kerberosUser.checkTGTAndRelogin();
+ } catch (LoginException e) {
+ throw new ProcessException("Unable to relogin with kerberos credentials for " + kerberosUser.getPrincipal(), e);
+ }
+ } else {
+ getLogger().debug("kerberosUser was null, will not refresh TGT with KerberosUser");
+ }
+ return ugi;
+ }
}
diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java
index d9113fe..edaefe8 100644
--- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java
+++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive3-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHive3Streaming.java
@@ -155,7 +155,7 @@ public class TestPutHive3Streaming {
System.setProperty("java.security.krb5.kdc", "nifi.kdc");
ugi = null;
- processor = new MockPutHive3Streaming();
+ processor = new MockPutHive3Streaming(ugi);
hiveConfigurator = mock(HiveConfigurator.class);
hiveConf = new HiveConf();
when(hiveConfigurator.getConfigurationFromFiles(anyString())).thenReturn(hiveConf);
@@ -272,6 +272,7 @@ public class TestPutHive3Streaming {
runner.setProperty(PutHive3Streaming.DB_NAME, "default");
runner.setProperty(PutHive3Streaming.TABLE_NAME, "users");
processor.ugi = mock(UserGroupInformation.class);
+ processor.kerberosUserReference.set(mock(KerberosUser.class));
runner.run();
assertNull(processor.ugi);
assertNull(processor.kerberosUserReference.get());
@@ -1128,6 +1129,10 @@ public class TestPutHive3Streaming {
new FieldSchema("scale", serdeConstants.DOUBLE_TYPE_NAME, "")
);
+ private MockPutHive3Streaming(UserGroupInformation ugi) {
+ this.ugi = ugi;
+ }
+
@Override
StreamingConnection makeStreamingConnection(HiveOptions options, RecordReader reader) throws StreamingException {
@@ -1175,6 +1180,11 @@ public class TestPutHive3Streaming {
public void setGeneratePermissionsFailure(boolean generatePermissionsFailure) {
this.generatePermissionsFailure = generatePermissionsFailure;
}
+
+ @Override
+ UserGroupInformation getUgi() {
+ return ugi;
+ }
}
private class MockHiveStreamingConnection implements StreamingConnection {