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();
        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;

        // 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());

        StreamingConnection hiveStreamingConnection = null;

        try (final InputStream rawIn = session.read(flowFile)) {
            final RecordReader reader;

            try (final BufferedInputStream in = new BufferedInputStream(rawIn)) {

                // 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);
                // Add shutdown handler with higher priority than FileSystem shutdown hook so that streaming connection gets closed first before
                // filesystem close (to avoid ClosedChannelException)
                ShutdownHookManager.addShutdownHook(hiveStreamingConnection::close, FileSystem.SHUTDOWN_HOOK_PRIORITY + 1);

                // Write records to Hive streaming, then commit and close
                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) {
            // 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) {
                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);
            // Restore original class loader, might not be necessary but is good practice since the processor task changed it
            Thread.currentThread().setContextClassLoader(originalClassloader);
        }
    }

}