@Test public void testMessageNotConsumedOnCommitFail() throws NoSuchFieldException, IllegalAccessException, NoSuchMethodException, InvocationTargetException { testRunner.run(1, false); ConsumeMQTT processor = (ConsumeMQTT) testRunner.getProcessor(); MQTTQueueMessage mock = mock(MQTTQueueMessage.class); when(mock.getPayload()).thenReturn(new byte[0]); when(mock.getTopic()).thenReturn("testTopic"); BlockingQueue<MQTTQueueMessage> mqttQueue = getMqttQueue(processor); mqttQueue.add(mock); try { ProcessSession session = testRunner.getProcessSessionFactory().createSession(); transferQueue(processor, (ProcessSession) Proxy.newProxyInstance(getClass().getClassLoader(), new Class[] { ProcessSession.class }, (proxy, method, args) -> { if (method.getName().equals("commit")) { throw new RuntimeException(); } else { return method.invoke(session, args); } })); fail("Expected runtime exception"); } catch (InvocationTargetException e) { assertTrue("Expected generic runtime exception, not " + e, e.getCause() instanceof RuntimeException); } assertTrue("Expected mqttQueue to contain uncommitted message.", mqttQueue.contains(mock)); }
private void transferQueue(ProcessSession session){ while (!mqttQueue.isEmpty()) { FlowFile messageFlowfile = session.create(); final MQTTQueueMessage mqttMessage = mqttQueue.peek(); Map<String, String> attrs = new HashMap<>(); attrs.put(BROKER_ATTRIBUTE_KEY, broker); attrs.put(TOPIC_ATTRIBUTE_KEY, mqttMessage.getTopic()); attrs.put(QOS_ATTRIBUTE_KEY, String.valueOf(mqttMessage.getQos())); attrs.put(IS_DUPLICATE_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isDuplicate())); attrs.put(IS_RETAINED_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isRetained())); messageFlowfile = session.putAllAttributes(messageFlowfile, attrs); messageFlowfile = session.write(messageFlowfile, new OutputStreamCallback() { @Override public void process(final OutputStream out) throws IOException { out.write(mqttMessage.getPayload()); } }); String transitUri = new StringBuilder(broker).append(mqttMessage.getTopic()).toString(); session.getProvenanceReporter().receive(messageFlowfile, transitUri); session.transfer(messageFlowfile, REL_MESSAGE); session.commit(); if (!mqttQueue.remove(mqttMessage) && logger.isWarnEnabled()) { logger.warn(new StringBuilder("FlowFile ") .append(messageFlowfile.getAttribute(CoreAttributes.UUID.key())) .append(" for Mqtt message ") .append(mqttMessage) .append(" had already been removed from queue, possible duplication of flow files") .toString()); } } }
ConsumeMQTT extends AbstractMQTTProcessor { private void transferQueue(ProcessSession session){ while (!mqttQueue.isEmpty()) { FlowFile messageFlowfile = session.create(); final MQTTQueueMessage mqttMessage = mqttQueue.peek(); Map<String, String> attrs = new HashMap<>(); attrs.put(BROKER_ATTRIBUTE_KEY, broker); attrs.put(TOPIC_ATTRIBUTE_KEY, mqttMessage.getTopic()); attrs.put(QOS_ATTRIBUTE_KEY, String.valueOf(mqttMessage.getQos())); attrs.put(IS_DUPLICATE_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isDuplicate())); attrs.put(IS_RETAINED_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isRetained())); messageFlowfile = session.putAllAttributes(messageFlowfile, attrs); messageFlowfile = session.write(messageFlowfile, new OutputStreamCallback() { @Override public void process(final OutputStream out) throws IOException { out.write(mqttMessage.getPayload()); } }); String transitUri = new StringBuilder(broker).append(mqttMessage.getTopic()).toString(); session.getProvenanceReporter().receive(messageFlowfile, transitUri); session.transfer(messageFlowfile, REL_MESSAGE); session.commit(); if (!mqttQueue.remove(mqttMessage) && logger.isWarnEnabled()) { logger.warn(new StringBuilder("FlowFile ") .append(messageFlowfile.getAttribute(CoreAttributes.UUID.key())) .append(" for Mqtt message ") .append(mqttMessage) .append(" had already been removed from queue, possible duplication of flow files") .toString()); } } } }
ConsumeMQTT extends AbstractMQTTProcessor { private void transferQueue(ProcessSession session){ while (!mqttQueue.isEmpty()) { FlowFile messageFlowfile = session.create(); final MQTTQueueMessage mqttMessage = mqttQueue.peek(); Map<String, String> attrs = new HashMap<>(); attrs.put(BROKER_ATTRIBUTE_KEY, broker); attrs.put(TOPIC_ATTRIBUTE_KEY, mqttMessage.getTopic()); attrs.put(QOS_ATTRIBUTE_KEY, String.valueOf(mqttMessage.getQos())); attrs.put(IS_DUPLICATE_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isDuplicate())); attrs.put(IS_RETAINED_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isRetained())); messageFlowfile = session.putAllAttributes(messageFlowfile, attrs); messageFlowfile = session.write(messageFlowfile, new OutputStreamCallback() { @Override public void process(final OutputStream out) throws IOException { out.write(mqttMessage.getPayload()); } }); String transitUri = new StringBuilder(broker).append(mqttMessage.getTopic()).toString(); session.getProvenanceReporter().receive(messageFlowfile, transitUri); session.transfer(messageFlowfile, REL_MESSAGE); session.commit(); if (!mqttQueue.remove(mqttMessage) && logger.isWarnEnabled()) { logger.warn(new StringBuilder("FlowFile ") .append(messageFlowfile.getAttribute(CoreAttributes.UUID.key())) .append(" for Mqtt message ") .append(mqttMessage) .append(" had already been removed from queue, possible duplication of flow files") .toString()); } } }  }
ConsumeMQTT extends AbstractMQTTProcessor { private void transferQueue(ProcessSession session){ while (!mqttQueue.isEmpty()) { FlowFile messageFlowfile = session.create(); final MQTTQueueMessage mqttMessage = mqttQueue.peek(); Map<String, String> attrs = new HashMap<>(); attrs.put(BROKER_ATTRIBUTE_KEY, broker); attrs.put(TOPIC_ATTRIBUTE_KEY, mqttMessage.getTopic()); attrs.put(QOS_ATTRIBUTE_KEY, String.valueOf(mqttMessage.getQos())); attrs.put(IS_DUPLICATE_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isDuplicate())); attrs.put(IS_RETAINED_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isRetained())); messageFlowfile = session.putAllAttributes(messageFlowfile, attrs); messageFlowfile = session.write(messageFlowfile, new OutputStreamCallback() { @Override public void process(final OutputStream out) throws IOException { out.write(mqttMessage.getPayload()); } }); String transitUri = new StringBuilder(broker).append(mqttMessage.getTopic()).toString(); session.getProvenanceReporter().receive(messageFlowfile, transitUri); session.transfer(messageFlowfile, REL_MESSAGE); session.commit(); if (!mqttQueue.remove(mqttMessage) && logger.isWarnEnabled()) { logger.warn(new StringBuilder("FlowFile ") .append(messageFlowfile.getAttribute(CoreAttributes.UUID.key())) .append(" for Mqtt message ") .append(mqttMessage) .append(" had already been removed from queue, possible duplication of flow files") .toString()); } } }  @Override void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue); @Override Collection<ValidationResult> customValidate(ValidationContext context); @Override Set<Relationship> getRelationships(); @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void onScheduled(final ProcessContext context); @OnUnscheduled void onUnscheduled(final ProcessContext context); @OnStopped void onStopped(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
ConsumeMQTT extends AbstractMQTTProcessor { private void transferQueue(ProcessSession session){ while (!mqttQueue.isEmpty()) { FlowFile messageFlowfile = session.create(); final MQTTQueueMessage mqttMessage = mqttQueue.peek(); Map<String, String> attrs = new HashMap<>(); attrs.put(BROKER_ATTRIBUTE_KEY, broker); attrs.put(TOPIC_ATTRIBUTE_KEY, mqttMessage.getTopic()); attrs.put(QOS_ATTRIBUTE_KEY, String.valueOf(mqttMessage.getQos())); attrs.put(IS_DUPLICATE_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isDuplicate())); attrs.put(IS_RETAINED_ATTRIBUTE_KEY, String.valueOf(mqttMessage.isRetained())); messageFlowfile = session.putAllAttributes(messageFlowfile, attrs); messageFlowfile = session.write(messageFlowfile, new OutputStreamCallback() { @Override public void process(final OutputStream out) throws IOException { out.write(mqttMessage.getPayload()); } }); String transitUri = new StringBuilder(broker).append(mqttMessage.getTopic()).toString(); session.getProvenanceReporter().receive(messageFlowfile, transitUri); session.transfer(messageFlowfile, REL_MESSAGE); session.commit(); if (!mqttQueue.remove(mqttMessage) && logger.isWarnEnabled()) { logger.warn(new StringBuilder("FlowFile ") .append(messageFlowfile.getAttribute(CoreAttributes.UUID.key())) .append(" for Mqtt message ") .append(mqttMessage) .append(" had already been removed from queue, possible duplication of flow files") .toString()); } } }  @Override void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue); @Override Collection<ValidationResult> customValidate(ValidationContext context); @Override Set<Relationship> getRelationships(); @Override final List<PropertyDescriptor> getSupportedPropertyDescriptors(); @OnScheduled void onScheduled(final ProcessContext context); @OnUnscheduled void onUnscheduled(final ProcessContext context); @OnStopped void onStopped(final ProcessContext context); @Override void onTrigger(final ProcessContext context, final ProcessSession session); final static String BROKER_ATTRIBUTE_KEY; final static String TOPIC_ATTRIBUTE_KEY; final static String QOS_ATTRIBUTE_KEY; final static String IS_DUPLICATE_ATTRIBUTE_KEY; final static String IS_RETAINED_ATTRIBUTE_KEY; static final PropertyDescriptor PROP_TOPIC_FILTER; static final PropertyDescriptor PROP_QOS; static final PropertyDescriptor PROP_MAX_QUEUE_SIZE; static final Relationship REL_MESSAGE; }