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

}