Tenemos la contestación a este apuro, o por lo menos eso creemos. Si sigues con interrogantes dínoslo y con placer te responderemos
Solución:
Probablemente esté enviando el mensaje antes de que se haya asignado el tema o la partición al consumidor. Establecer propiedad…
spring:
kafka:
consumer:
auto-offset-reset: earliest
… por defecto es latest.
Esto es como usar --from-beginning con el consumidor de la consola.
EDITAR
Oh; no estás usando las propiedades de arranque.
Agregar
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
EDIT2
Por cierto, probablemente también deberías hacer un get(10L, TimeUnit.SECONDS) sobre el resultado de la template.send() (a Future<>) para afirmar que el envío fue exitoso.
EDIT3
Para anular el restablecimiento de compensación solo para la prueba, puede hacer lo mismo que hizo con las direcciones de los intermediarios:
@Value("$spring.kafka.consumer.auto-offset-reset:latest")
private String reset;
...
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, this.reset);
y
@TestPropertySource(properties = "spring.kafka.bootstrap-servers=$spring.embedded.kafka.brokers",
"spring.kafka.consumer.auto-offset-reset=earliest")
Sin embargo, tenga en cuenta que esta propiedad solo se aplica la primera vez que un grupo consume. Para comenzar siempre por el final cada vez que se inicia la aplicación, debe buscar hasta el final durante el inicio.
Además, recomendaría configurar enable.auto.commit para false para que el contenedor se encargue de comprometer las compensaciones en lugar de simplemente confiar en que el cliente consumidor lo haga en un horario.
Valoraciones y comentarios
Eres capaz de ayudar nuestra misión escribiendo un comentario y puntuándolo te estamos eternamente agradecidos.