启动版本 2.1.3 ，spring kafka在 MessageListenerContainer :

/**
 * Pause this container before the next poll().
 * @since 2.1.3
 */
default void pause() {
    throw new UnsupportedOperationException("This container doesn't support pause");
}

/**
 * Resume this container, if paused, after the next poll().
 * @since 2.1.3
 */
default void resume() {
    throw new UnsupportedOperationException("This container doesn't support resume");
}

/**
 * Return true if {@link #pause()} has been called; the container might not have actually
 * paused yet.
 * @return true if pause has been requested.
 * @since 2.1.5
 */
default boolean isPauseRequested() {
    throw new UnsupportedOperationException("This container doesn't support pause/resume");
}

/**
 * Return true if {@link #pause()} has been called; and all consumers in this container
 * have actually paused.
 * @return true if the container is paused.
 * @since 2.1.5
 */
default boolean isContainerPaused() {
    throw new UnsupportedOperationException("This container doesn't support pause/resume");
}

因此，您确实可以从应用程序的任何一点暂停和恢复侦听器容器 containertest 提供适当的服务。
这个 KafkaMessageDrivenChannelAdapter 也暴露了 pause() 以及 resume() 钩子。
另外，在 KafkaMessageDrivenChannelAdapter ，的 MessagingMessageConverter 将这些标头填充到消息中以进行下游处理：

rawHeaders.put(KafkaHeaders.RECEIVED_MESSAGE_KEY, record.key());
    rawHeaders.put(KafkaHeaders.RECEIVED_TOPIC, record.topic());
    rawHeaders.put(KafkaHeaders.RECEIVED_PARTITION_ID, record.partition());
    rawHeaders.put(KafkaHeaders.OFFSET, record.offset());
    rawHeaders.put(KafkaHeaders.TIMESTAMP_TYPE, record.timestampType().name());
    rawHeaders.put(KafkaHeaders.RECEIVED_TIMESTAMP, record.timestamp());

    if (acknowledgment != null) {
        rawHeaders.put(KafkaHeaders.ACKNOWLEDGMENT, acknowledgment);
    }
    if (consumer != null) {
        rawHeaders.put(KafkaHeaders.CONSUMER, consumer);
    }

因此，您可以获得 KafkaHeaders.CONSUMER 标题并使用本机 pause()/resume() 从 KafkaConsumer .