Je suis nouveau sur kafka et quarkus, je souhaite envoyer un message au sujet kafka lorsqu'une demande d'utilisateur a été traitée.

J'ai parcouru l'exemple de kafka fourni dans quarkus-quickstart. J'ai essayé avec KafkaMessage

// when GET called send message to topic
@GET
@Produces(MediaType.TEXT_PLAIN)
public String hello() {
    generateSingle();
    return "hello";
}

@Outgoing("single-stations")
public KafkaMessage<Integer, String> generateSingle() {
    return KafkaMessage.of(1, "value");
};

Mais j'ai obtenu un résultat en envoyant un message au sujet kafka en continu.

Je veux savoir s'il y a une autre méthode ou y a-t-il un problème avec mon code.

Aide appréciée

2
Jithin S 11 oct. 2019 à 14:45

1 réponse

Meilleure réponse

La documentation sur ce sujet est laconique et incomplète pour le moment (Quarkus 0.25.0). J'ai réussi à le faire, mais il a fallu beaucoup d'expérimentation et quelque chose que je crois est un hack qui, espérons-le, sera corrigé dans les versions ultérieures de Quarkus.

Le principe est que la méthode @Outgoing doit produire un stream contrôlé de l'extérieur. Ceci est accompli en créant le flux via Flowable.create() dans une méthode @PostConstruct et en exposant l'émetteur à un membre de la classe. La méthode @Outgoing renvoie simplement le flux déjà construit.

Le composant suivant expose une méthode publique, produce(String message) qui enverra ce message texte à Kafka :

package ...

import java.util.UUID;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.reactivestreams.Publisher;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.smallrye.reactive.messaging.kafka.KafkaMessage;

@ApplicationScoped
public class KafkaController {

    private FlowableEmitter<KafkaMessage<String, String>> emitter;

    private Flowable<KafkaMessage<String, String>> outgoingStream;

    @PostConstruct
    void init() {
        outgoingStream = Flowable.create(emitter -> this.emitter = emitter, BackpressureStrategy.BUFFER);
    }

    public void produce(String message) {
        emitter.onNext(KafkaMessage.of(UUID.randomUUID().toString(), message));
    }

    @PreDestroy
    void dispose() {
        emitter.onComplete();
    }

    @Outgoing("internal")
    Publisher<KafkaMessage<String, String>> produceKafkaMessage() {
        return outgoingStream;
    }

    @Incoming("internal")
    @Outgoing("kafka-test")
    KafkaMessage<String, String> transform(Message<KafkaMessage<String, String>> arg) {
        return arg.getPayload();
    }
}

J'ai créé cette classe dans une application Quarkus générée, comme documenté ici :

mvn io.quarkus:quarkus-maven-plugin:0.25.0:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=kafka-quickstart \
    -Dextensions="kafka"

Et configuré (application.properties) comme suit :

kafka.bootstrap.servers=localhost:9092

mp.messaging.outgoing.kafka-test.connector=smallrye-kafka
mp.messaging.outgoing.kafka-test.topic=test
mp.messaging.outgoing.kafka-test.key.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.outgoing.kafka-test.value.serializer=org.apache.kafka.common.serialization.StringSerializer

Une instance Kafka est démarrée exactement comme décrit dans le quickstart. Vous pouvez regarder le sujet test avec un écouteur de console comme suit :

./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic test --from-beginning --group test-console.consumer

Pour le tester, vous pouvez créer une ressource JAX-RS pour appeler produce() :

package ...

import javax.inject.Inject;
import javax.ws.rs.POST;
import javax.ws.rs.Path;

@Path("/control")
public class KafkaProduceControlResource {

    @Inject
    KafkaController kafkaController;

    @POST
    @Path("/produce")
    public void produceMessage(String message) {
        kafkaController.produce(message);
    }
}

Appelez-le à partir de la ligne de commande comme suit et observez le consommateur de la console :

curl -i -s -X POST -d "A text message" \
    http://localhost:8080/control/produce

LE HACK : Il semble que l'annotation de produceKafkaMessage() avec @Outgoing("kafka-test") échoue, car Quarkus ne comprend PAS qu'un KafkaMessage est un Message, et l'encapsule dans un seul, ce qui entraîne des erreurs de sérialisation. Je contourne cela avec le flux "internal".

3
Nikos Paraskevopoulos 16 oct. 2019 à 21:04