GetKafka continually throws NullPointerException if it ever fails to write out message due to Kafka timeout

        catch (final Exception e) {
            this.shutdownConsumer();
            getLogger().error("Failed to receive FlowFile from Kafka due to {}", new Object[]{e});
            if (flowFile != null) {
                session.remove(flowFile);
            }

if (this.executor != null) {
            this.executor.shutdown();
            try {
                if (!this.executor.awaitTermination(30000, TimeUnit.MILLISECONDS)) {
                    this.executor.shutdownNow();
                    getLogger().warn("Executor did not stop in 30 sec. Terminated.");
                }
                this.executor = null;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        synchronized (this.consumerStreamsReady) {
            if (!this.consumerStreamsReady.get()) {
                Future<Void> f = this.executor.submit(new Callable<Void>() {
...

        if (this.consumerStreamsReady.get()) {
            Future<Void> consumptionFuture = this.executor.submit(new Callable<Void>() {
                @Override
                public Void call() throws Exception {
...