@Test public void onTrigger() throws Exception { configure(processor, 1); runner.setProperty(PutHive3Streaming.METASTORE_URI, "thrift: runner.setProperty(PutHive3Streaming.DB_NAME, "default"); runner.setProperty(PutHive3Streaming.TABLE_NAME, "users"); runner.enqueue(new byte[0]); runner.run(); runner.assertTransferCount(PutHive3Streaming.REL_SUCCESS, 1); final MockFlowFile flowFile = runner.getFlowFilesForRelationship(PutHive3Streaming.REL_SUCCESS).get(0); assertEquals("1", flowFile.getAttribute(HIVE_STREAMING_RECORD_COUNT_ATTR)); assertEquals("default.users", flowFile.getAttribute(ATTR_OUTPUT_TABLES)); }
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(); final String metastoreUri = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue(); final String partitionValuesString = context.getProperty(PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue(); final boolean autoCreatePartitions = context.getProperty(AUTOCREATE_PARTITIONS).asBoolean(); final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean(); HiveOptions o = new HiveOptions(metastoreUri, dbName, tableName) .withHiveConf(hiveConfig) .withAutoCreatePartitions(autoCreatePartitions) .withCallTimeout(callTimeout) .withStreamingOptimizations(!disableStreamingOptimizations); if (!StringUtils.isEmpty(partitionValuesString)) { List<String> staticPartitionValues = Arrays.stream(partitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList()); o = o.withStaticPartitionValues(staticPartitionValues); } if (SecurityUtil.isSecurityEnabled(hiveConfig)) { final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class); o = o.withKerberosPrincipal(credentialsService.getPrincipal()).withKerberosKeytab(credentialsService.getKeytab()); } final HiveOptions options = o; ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader()); StreamingConnection hiveStreamingConnection = null; try (final InputStream rawIn = session.read(flowFile)) { final RecordReader reader; try (final BufferedInputStream in = new BufferedInputStream(rawIn)) { try { reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger()); } catch (Exception e) { throw new RecordReaderFactoryException("Unable to create RecordReader", e); } hiveStreamingConnection = makeStreamingConnection(options, reader); ShutdownHookManager.addShutdownHook(hiveStreamingConnection::close, FileSystem.SHUTDOWN_HOOK_PRIORITY + 1); hiveStreamingConnection.beginTransaction(); hiveStreamingConnection.write(in); hiveStreamingConnection.commitTransaction(); rawIn.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()); session.transfer(flowFile, REL_SUCCESS); } catch (TransactionError te) { if (rollbackOnFailure) { throw new ProcessException(te.getLocalizedMessage(), te); } else { throw new ShouldRetryException(te.getLocalizedMessage(), te); } } catch (RecordReaderFactoryException rrfe) { throw new ProcessException(rrfe); } } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) { if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(e.getLocalizedMessage(), e); } else { 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.transfer(flowFile, REL_FAILURE); } } catch (DiscontinuedException e) { getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e); session.transfer(flowFile, Relationship.SELF); } catch (ConnectionError ce) { context.yield(); throw new ProcessException("A connection to metastore cannot be established", ce); } catch (ShouldRetryException e) { getLogger().error(e.getLocalizedMessage(), e); if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } flowFile = session.penalize(flowFile); session.transfer(flowFile, REL_RETRY); } catch (StreamingException se) { Throwable cause = se.getCause(); if (cause == null) cause = se; if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(cause.getLocalizedMessage(), cause); } else { flowFile = session.penalize(flowFile); 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.transfer(flowFile, REL_FAILURE); } } catch (Throwable t) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t); } finally { closeConnection(hiveStreamingConnection); Thread.currentThread().setContextClassLoader(originalClassloader); } }
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(); final String metastoreUri = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue(); final String partitionValuesString = context.getProperty(PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue(); final boolean autoCreatePartitions = context.getProperty(AUTOCREATE_PARTITIONS).asBoolean(); final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean(); HiveOptions o = new HiveOptions(metastoreUri, dbName, tableName) .withHiveConf(hiveConfig) .withAutoCreatePartitions(autoCreatePartitions) .withCallTimeout(callTimeout) .withStreamingOptimizations(!disableStreamingOptimizations); if (!StringUtils.isEmpty(partitionValuesString)) { List<String> staticPartitionValues = Arrays.stream(partitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList()); o = o.withStaticPartitionValues(staticPartitionValues); } if (SecurityUtil.isSecurityEnabled(hiveConfig)) { final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class); o = o.withKerberosPrincipal(credentialsService.getPrincipal()).withKerberosKeytab(credentialsService.getKeytab()); } final HiveOptions options = o; ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader()); StreamingConnection hiveStreamingConnection = null; try (final InputStream rawIn = session.read(flowFile)) { final RecordReader reader; try (final BufferedInputStream in = new BufferedInputStream(rawIn)) { try { reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger()); } catch (Exception e) { throw new RecordReaderFactoryException("Unable to create RecordReader", e); } hiveStreamingConnection = makeStreamingConnection(options, reader); ShutdownHookManager.addShutdownHook(hiveStreamingConnection::close, FileSystem.SHUTDOWN_HOOK_PRIORITY + 1); hiveStreamingConnection.beginTransaction(); hiveStreamingConnection.write(in); hiveStreamingConnection.commitTransaction(); rawIn.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()); session.transfer(flowFile, REL_SUCCESS); } catch (TransactionError te) { if (rollbackOnFailure) { throw new ProcessException(te.getLocalizedMessage(), te); } else { throw new ShouldRetryException(te.getLocalizedMessage(), te); } } catch (RecordReaderFactoryException rrfe) { throw new ProcessException(rrfe); } } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) { if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(e.getLocalizedMessage(), e); } else { 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.transfer(flowFile, REL_FAILURE); } } catch (DiscontinuedException e) { getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e); session.transfer(flowFile, Relationship.SELF); } catch (ConnectionError ce) { context.yield(); throw new ProcessException("A connection to metastore cannot be established", ce); } catch (ShouldRetryException e) { getLogger().error(e.getLocalizedMessage(), e); if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } flowFile = session.penalize(flowFile); session.transfer(flowFile, REL_RETRY); } catch (StreamingException se) { Throwable cause = se.getCause(); if (cause == null) cause = se; if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(cause.getLocalizedMessage(), cause); } else { flowFile = session.penalize(flowFile); 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.transfer(flowFile, REL_FAILURE); } } catch (Throwable t) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t); } finally { closeConnection(hiveStreamingConnection); Thread.currentThread().setContextClassLoader(originalClassloader); } } }
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(); final String metastoreUri = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue(); final String partitionValuesString = context.getProperty(PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue(); final boolean autoCreatePartitions = context.getProperty(AUTOCREATE_PARTITIONS).asBoolean(); final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean(); HiveOptions o = new HiveOptions(metastoreUri, dbName, tableName) .withHiveConf(hiveConfig) .withAutoCreatePartitions(autoCreatePartitions) .withCallTimeout(callTimeout) .withStreamingOptimizations(!disableStreamingOptimizations); if (!StringUtils.isEmpty(partitionValuesString)) { List<String> staticPartitionValues = Arrays.stream(partitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList()); o = o.withStaticPartitionValues(staticPartitionValues); } if (SecurityUtil.isSecurityEnabled(hiveConfig)) { final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class); o = o.withKerberosPrincipal(credentialsService.getPrincipal()).withKerberosKeytab(credentialsService.getKeytab()); } final HiveOptions options = o; ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader()); StreamingConnection hiveStreamingConnection = null; try (final InputStream rawIn = session.read(flowFile)) { final RecordReader reader; try (final BufferedInputStream in = new BufferedInputStream(rawIn)) { try { reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger()); } catch (Exception e) { throw new RecordReaderFactoryException("Unable to create RecordReader", e); } hiveStreamingConnection = makeStreamingConnection(options, reader); ShutdownHookManager.addShutdownHook(hiveStreamingConnection::close, FileSystem.SHUTDOWN_HOOK_PRIORITY + 1); hiveStreamingConnection.beginTransaction(); hiveStreamingConnection.write(in); hiveStreamingConnection.commitTransaction(); rawIn.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()); session.transfer(flowFile, REL_SUCCESS); } catch (TransactionError te) { if (rollbackOnFailure) { throw new ProcessException(te.getLocalizedMessage(), te); } else { throw new ShouldRetryException(te.getLocalizedMessage(), te); } } catch (RecordReaderFactoryException rrfe) { throw new ProcessException(rrfe); } } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) { if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(e.getLocalizedMessage(), e); } else { 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.transfer(flowFile, REL_FAILURE); } } catch (DiscontinuedException e) { getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e); session.transfer(flowFile, Relationship.SELF); } catch (ConnectionError ce) { context.yield(); throw new ProcessException("A connection to metastore cannot be established", ce); } catch (ShouldRetryException e) { getLogger().error(e.getLocalizedMessage(), e); if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } flowFile = session.penalize(flowFile); session.transfer(flowFile, REL_RETRY); } catch (StreamingException se) { Throwable cause = se.getCause(); if (cause == null) cause = se; if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(cause.getLocalizedMessage(), cause); } else { flowFile = session.penalize(flowFile); 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.transfer(flowFile, REL_FAILURE); } } catch (Throwable t) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t); } finally { closeConnection(hiveStreamingConnection); Thread.currentThread().setContextClassLoader(originalClassloader); } }  }
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(); final String metastoreUri = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue(); final String partitionValuesString = context.getProperty(PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue(); final boolean autoCreatePartitions = context.getProperty(AUTOCREATE_PARTITIONS).asBoolean(); final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean(); HiveOptions o = new HiveOptions(metastoreUri, dbName, tableName) .withHiveConf(hiveConfig) .withAutoCreatePartitions(autoCreatePartitions) .withCallTimeout(callTimeout) .withStreamingOptimizations(!disableStreamingOptimizations); if (!StringUtils.isEmpty(partitionValuesString)) { List<String> staticPartitionValues = Arrays.stream(partitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList()); o = o.withStaticPartitionValues(staticPartitionValues); } if (SecurityUtil.isSecurityEnabled(hiveConfig)) { final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class); o = o.withKerberosPrincipal(credentialsService.getPrincipal()).withKerberosKeytab(credentialsService.getKeytab()); } final HiveOptions options = o; ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader()); StreamingConnection hiveStreamingConnection = null; try (final InputStream rawIn = session.read(flowFile)) { final RecordReader reader; try (final BufferedInputStream in = new BufferedInputStream(rawIn)) { try { reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger()); } catch (Exception e) { throw new RecordReaderFactoryException("Unable to create RecordReader", e); } hiveStreamingConnection = makeStreamingConnection(options, reader); ShutdownHookManager.addShutdownHook(hiveStreamingConnection::close, FileSystem.SHUTDOWN_HOOK_PRIORITY + 1); hiveStreamingConnection.beginTransaction(); hiveStreamingConnection.write(in); hiveStreamingConnection.commitTransaction(); rawIn.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()); session.transfer(flowFile, REL_SUCCESS); } catch (TransactionError te) { if (rollbackOnFailure) { throw new ProcessException(te.getLocalizedMessage(), te); } else { throw new ShouldRetryException(te.getLocalizedMessage(), te); } } catch (RecordReaderFactoryException rrfe) { throw new ProcessException(rrfe); } } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) { if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(e.getLocalizedMessage(), e); } else { 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.transfer(flowFile, REL_FAILURE); } } catch (DiscontinuedException e) { getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e); session.transfer(flowFile, Relationship.SELF); } catch (ConnectionError ce) { context.yield(); throw new ProcessException("A connection to metastore cannot be established", ce); } catch (ShouldRetryException e) { getLogger().error(e.getLocalizedMessage(), e); if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } flowFile = session.penalize(flowFile); session.transfer(flowFile, REL_RETRY); } catch (StreamingException se) { Throwable cause = se.getCause(); if (cause == null) cause = se; if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(cause.getLocalizedMessage(), cause); } else { flowFile = session.penalize(flowFile); 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.transfer(flowFile, REL_FAILURE); } } catch (Throwable t) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t); } finally { closeConnection(hiveStreamingConnection); Thread.currentThread().setContextClassLoader(originalClassloader); } }  @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); void onTrigger(ProcessContext context, ProcessSession session); @OnStopped void cleanup(); }
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(); final String metastoreUri = context.getProperty(METASTORE_URI).evaluateAttributeExpressions(flowFile).getValue(); final String partitionValuesString = context.getProperty(PARTITION_VALUES).evaluateAttributeExpressions(flowFile).getValue(); final boolean autoCreatePartitions = context.getProperty(AUTOCREATE_PARTITIONS).asBoolean(); final boolean disableStreamingOptimizations = context.getProperty(DISABLE_STREAMING_OPTIMIZATIONS).asBoolean(); HiveOptions o = new HiveOptions(metastoreUri, dbName, tableName) .withHiveConf(hiveConfig) .withAutoCreatePartitions(autoCreatePartitions) .withCallTimeout(callTimeout) .withStreamingOptimizations(!disableStreamingOptimizations); if (!StringUtils.isEmpty(partitionValuesString)) { List<String> staticPartitionValues = Arrays.stream(partitionValuesString.split(",")).filter(Objects::nonNull).map(String::trim).collect(Collectors.toList()); o = o.withStaticPartitionValues(staticPartitionValues); } if (SecurityUtil.isSecurityEnabled(hiveConfig)) { final KerberosCredentialsService credentialsService = context.getProperty(KERBEROS_CREDENTIALS_SERVICE).asControllerService(KerberosCredentialsService.class); o = o.withKerberosPrincipal(credentialsService.getPrincipal()).withKerberosKeytab(credentialsService.getKeytab()); } final HiveOptions options = o; ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader(); Thread.currentThread().setContextClassLoader(this.getClass().getClassLoader()); StreamingConnection hiveStreamingConnection = null; try (final InputStream rawIn = session.read(flowFile)) { final RecordReader reader; try (final BufferedInputStream in = new BufferedInputStream(rawIn)) { try { reader = recordReaderFactory.createRecordReader(flowFile, in, getLogger()); } catch (Exception e) { throw new RecordReaderFactoryException("Unable to create RecordReader", e); } hiveStreamingConnection = makeStreamingConnection(options, reader); ShutdownHookManager.addShutdownHook(hiveStreamingConnection::close, FileSystem.SHUTDOWN_HOOK_PRIORITY + 1); hiveStreamingConnection.beginTransaction(); hiveStreamingConnection.write(in); hiveStreamingConnection.commitTransaction(); rawIn.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()); session.transfer(flowFile, REL_SUCCESS); } catch (TransactionError te) { if (rollbackOnFailure) { throw new ProcessException(te.getLocalizedMessage(), te); } else { throw new ShouldRetryException(te.getLocalizedMessage(), te); } } catch (RecordReaderFactoryException rrfe) { throw new ProcessException(rrfe); } } catch (InvalidTable | SerializationError | StreamingIOFailure | IOException e) { if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(e.getLocalizedMessage(), e); } else { 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.transfer(flowFile, REL_FAILURE); } } catch (DiscontinuedException e) { getLogger().warn("Discontinued processing for {} due to {}", new Object[]{flowFile, e}, e); session.transfer(flowFile, Relationship.SELF); } catch (ConnectionError ce) { context.yield(); throw new ProcessException("A connection to metastore cannot be established", ce); } catch (ShouldRetryException e) { getLogger().error(e.getLocalizedMessage(), e); if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } flowFile = session.penalize(flowFile); session.transfer(flowFile, REL_RETRY); } catch (StreamingException se) { Throwable cause = se.getCause(); if (cause == null) cause = se; if (rollbackOnFailure) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw new ProcessException(cause.getLocalizedMessage(), cause); } else { flowFile = session.penalize(flowFile); 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.transfer(flowFile, REL_FAILURE); } } catch (Throwable t) { if (hiveStreamingConnection != null) { abortConnection(hiveStreamingConnection); } throw (t instanceof ProcessException) ? (ProcessException) t : new ProcessException(t); } finally { closeConnection(hiveStreamingConnection); Thread.currentThread().setContextClassLoader(originalClassloader); } }  @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); void onTrigger(ProcessContext context, ProcessSession session); @OnStopped void cleanup(); static final String HIVE_STREAMING_RECORD_COUNT_ATTR; static final Relationship REL_SUCCESS; static final Relationship REL_FAILURE; static final Relationship REL_RETRY; }