@Override
    public boolean complete() {
        if (snapshotInProgress) {
            return false;
        }
        while (emitFromTraverser(pendingTraverser)) {
            try {
                Message t = consumer.receiveNoWait();
                if (t == null) {
                    pendingTraverser = eventTimeMapper.flatMapIdle();
                    break;
                }
                if (guarantee == EXACTLY_ONCE) {
                    // We don't know whether the messages with the restored IDs were acknowledged in the previous
                    // execution or not. They are acknowledged in phase-2 of the snapshot which might not be executed.
                    // If we receive a message with a restored ID, we ignore it. But if we don't receive some ID,
                    // we can never safely throw it out.
                    // In order to avoid storing the restored IDs forever, we set a timeout after which we clear the
                    // collection. We start the timeout after receiving the first message, at which time we know the
                    // broker is working. We assume it will redeliver the messages promptly; if it doesn't, we assume
                    // they were acknowledged in the previous execution or delivered to another processor in this
                    // execution.
                    if (restoredIdsExpiration == Long.MAX_VALUE) {
                        restoredIdsExpiration = System.nanoTime() + RESTORED_IDS_TTL;
                    } else if (!restoredIds.isEmpty() && restoredIdsExpiration <= System.nanoTime()) {
                        restoredIds = emptySet();
                    }
                    Object msgId = messageIdFn.apply(t);
                    if (msgId == null) {
                        throw new JetException("Received a message without an ID. All messages must have an ID, " +
                                "you can specify an extracting function using "
                                + JmsSourceBuilder.class.getSimpleName() + ".messageIdFn()");
                    }
                    seenIds.add(msgId);
                    if (restoredIds.remove(msgId)) {
                        getLogger().fine("Redelivered message dropped: %s", t);
                        continue;
                    }
                }
                T projectedItem = projectionFn.apply(t);
                pendingTraverser = projectedItem != null
                        ? eventTimeMapper.flatMapEvent(projectedItem, 0, handleJmsTimestamp(t))
                        : eventTimeMapper.flatMapIdle();
            } catch (JMSException e) {
                throw sneakyThrow(e);
            }
        }
        return false;
    }
@Test
    public void when_projectionToNull_then_filteredOut() throws Exception {
        String queueName = randomString();
        logger.info("using queue: " + queueName);
        String message1 = sendMessage(queueName, true);
        String message2 = sendMessage(queueName, true);

        initializeProcessor(queueName, true, m -> {
            String msg = ((TextMessage) m).getText();
            return msg.equals(message1) ? null : msg;
        });
        Queue<Object> queue = outbox.queue(0);

        List<Object> actualOutput = new ArrayList<>();
        assertTrueEventually(() -> {
            outbox.reset();
            processor.complete();
            Object item = queue.poll();
            if (item != null) {
                actualOutput.add(item);
            }
            assertEquals(singletonList(message2), actualOutput);
        });
    }