Questions about Kafka Consumer of Transient messages via Akka-Streams on Multiple Nodes

[kraythe]

We are using Kafka to store messages that are produced by a node in our cluster and to be distributed to all nodes in the cluster and I have it mostly working with akka-streams but there is a couple of questions I have to tie this up. There are some constraints to this.

First of all the message has to be consumed by every node in the cluster but produced by only one node. I understand I can assign each node a group id that is probably its node ID which means each node will get the message. That sorted. But here are the questions.

The data is extremely transient and fairly large (just under a meg) and cannot be compressed further or broken up. If there is a new message on the topic the old one is pretty much trash. How can I limit the topic to basically just one message currently maximum?

Given that the data is necessary for the node to start, I need to consume the latest message on the topic no matter whether I have consumed it before and, hopefully without creating a unique group id every time I start the server. Is this possible and if so, how can it be done.

Finally, the data is usually on the topic but on occasion it is not there and I, ideally, need to be able to check if there is a message there and if not ask the producer to create the message. Is this possible?

Thanks a bunch.

[kraythe]

Here is the code I am using currently to start this: 

private Control startMatrixConsumer() {
    final ConsumerSettings<Long, byte[]> matrixConsumerSettings = ConsumerSettings
            .create(services.actorSystem(), new LongDeserializer(), new ByteArrayDeserializer())
            .withBootstrapServers(services.config().getString("kafka.bootstrapServers"))
            .withGroupId("group1") // todo put in the conf ??
            .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    final String topicName = Matrix.class.getSimpleName() + '-' + eventId;
    final AutoSubscription subscription = Subscriptions.topics(topicName);
    return Consumer.plainSource(MatrixConsumerSettings, subscription)
            .named(Matrix.class.getSimpleName() + "-Kafka-Consumer-" + eventId)
            .map(data -> {
                final Matrix matrix = services.kryoDeserialize(data.value(), Matrix.class);
                log.debug(format("Received %s for event %d from Kafka", Matrix.class.getSimpleName(), matrix.getEventId()));
                return matrix;
            })
            .filter(Objects::nonNull)
            .to(Sink.actorRef(getSelf(), NotUsed.getInstance()))
            .run(ActorMaterializer.create(getContext()));
}