Error on Kafka integration test with Apicurio Schema Registry

[SamyLegal]

Hi,

I develop a Quarkus application is a Kafka Consumer and we use Apicurio like Schema Registry.

For check, if my application works in native mode, i have develop an integration test.

When i execute this test, i have that error asking me that an Apicurio Http Client is not available.

org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:459)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:287)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:314)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:299)
        at fr.mgdis.data_sources.KafkaTestUtils.getKafkaProducer(KafkaTestUtils.java:41)
        at fr.mgdis.data_sources.entities.piece.consumer.PieceChangeStreamConsumerKafkaClientIT.testConsumeKafkaMessageInNativeMode(PieceChangeStreamConsumerKafkaClientIT.java:22)
        at java.base/java.lang.reflect.Method.invoke(Method.java:580)
        at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
        at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
Caused by: java.lang.IllegalStateException: java.lang.IllegalStateException: No available ApicurioHttClientProvider provider found, try manually setting the provider using ApicurioHttpClientFactory.setProvider
        at io.apicurio.registry.resolver.AbstractSchemaResolver.configure(AbstractSchemaResolver.java:92)
        at io.apicurio.registry.resolver.DefaultSchemaResolver.configure(DefaultSchemaResolver.java:61)
        at io.apicurio.registry.serde.SchemaResolverConfigurer.configure(SchemaResolverConfigurer.java:87)
        at io.apicurio.registry.serde.AbstractKafkaSerDe.configure(AbstractKafkaSerDe.java:70)
        at io.apicurio.registry.serde.avro.AvroKafkaSerializer.configure(AvroKafkaSerializer.java:91)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:387)
        ... 8 more
Caused by: java.lang.IllegalStateException: No available ApicurioHttClientProvider provider found, try manually setting the provider using ApicurioHttpClientFactory.setProvider
        at io.apicurio.rest.client.spi.ApicurioHttpClientFactory.resolveProviderInstance(ApicurioHttpClientFactory.java:42)
        at io.apicurio.rest.client.spi.ApicurioHttpClientFactory.create(ApicurioHttpClientFactory.java:23)
        at io.apicurio.registry.rest.client.RegistryClientFactory.create(RegistryClientFactory.java:79)
        at io.apicurio.registry.rest.client.RegistryClientFactory.create(RegistryClientFactory.java:67)
        at io.apicurio.registry.resolver.AbstractSchemaResolver.configure(AbstractSchemaResolver.java:88)
        ... 13 more

My unit test is as follows.
I create a Kafka Producer, that create a message in Kafka (that is a Quarkus Dev Service) and i check if my consumer, consume well the messages by calling the rest api.

For information the Kafka consumer take messages from Kafka and insert them in the database.

@QuarkusIntegrationTest
class PieceChangeStreamConsumerKafkaClientIT {
  final String topic = "mgdis-core-data-sources.pieces";

  @Test
  void testConsumeKafkaMessageInNativeMode() {
    try (final Producer<String, PieceChangeStream> producer = KafkaTestUtils.getKafkaProducer()) {
      var reference = "piece-mock-native";
      PieceChangeStream pieceChangeStream = Instancio.of(PieceChangeStream.class)
        .set(field(Piece.class, "reference"), reference)
        .set(field(Piece.class, "date"), null)
        .set(field(Piece.class, "description"), null)
        .generate(field(Piece.class, "statut"), gen -> gen.oneOf("PUBLIE", "BROUILLON", "DESACTIVE"))
        .set(field(Piece.class, "expiration"), null)
        .create();

      producer.send(new ProducerRecord<>(topic, reference, pieceChangeStream));
    }

    await().until(() -> {
      int count = given()
        .contentType("application/json")
        .when()
        .get("/api/entities/piece")
        .then()
        .extract()
        .response()
        .path("count");

      return count == 1;
    });

    // Remove all pieces from database with api rest
    given()
      .contentType("application/json")
      .when()
      .post("/api/entities/piece/truncate")
      .then()
      .statusCode(200);
  }

Here is the code of the KafkaTestUtils class to create the Kafka producer for my integration test.

package fr.mgdis.data_sources;

import io.apicurio.registry.serde.SerdeConfig;
import io.apicurio.registry.serde.avro.AvroKafkaSerializer;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.eclipse.microprofile.config.ConfigProvider;

/**
 * Class that provides utility methods for interacting with the Kafka broker
 * and facilitating the creation of data producers.
 */
public class KafkaTestUtils {

  /**
   * Method that constructs a Kafka producer for which the message value type is set by the caller.
   * For the Kafka message key, the default type is string.
   *
   * @param <T> Type of the Kafka message value
   * @return Kafka producer
   */
  public static <T> KafkaProducer<String, T> getKafkaProducer() {
    // Kafka url
    String kafkaUrl = ConfigProvider.getConfig().getValue("kafka.bootstrap.servers", String.class);

    String apicurioRegistryUrl =
      ConfigProvider.getConfig().getValue("mp.messaging.connector.smallrye-kafka.apicurio.registry.url", String.class);

    // Producer
    final Properties properties = new Properties();
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
    properties.put(SerdeConfig.REGISTRY_URL, apicurioRegistryUrl);
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AvroKafkaSerializer.class.getName());
    properties.put(SerdeConfig.ARTIFACT_RESOLVER_STRATEGY, "io.apicurio.registry.serde.avro.strategy.RecordIdStrategy");
    properties.put(SerdeConfig.FIND_LATEST_ARTIFACT, Boolean.TRUE);
    properties.put(SerdeConfig.AUTO_REGISTER_ARTIFACT, Boolean.TRUE);

    return new KafkaProducer<>(properties);
  }
}

If i change the annotation "QuarkusIntegrationTest" by "QuarkusTest" my integration test works.

For information, the commands that execute for execute my integration tests :

./mvnw -U $MAVEN_CLI_OPTS clean package -Dnative -DskipTests=true
./mvnw $MAVEN_CLI_OPTS test-compile failsafe:integration-test -Pnative

If you have some ideas it will be great.

Thank you.

[quarkus-bot[bot]]

/cc @alesj (kafka), @cescoffier (kafka), @ozangunalp (kafka)

[cescoffier]

Which extensions are you using?

[SamyLegal]

We use standard Quarkus extensions for work with Kafka and Apicurio like :

• smallrye-messaging
• quarkus-apicurio-registry-avro
• quarkus-avro

To answer the question precisely, i give an extract of my pom.xml which list extensions i use.

I have omit test dependencies for convenience.

<!-- Quarkus Core Dependencies -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-resteasy-reactive</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-arc</artifactId>
    </dependency>

    <!-- PostgreSQL driver -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-jdbc-postgresql</artifactId>
    </dependency>

    <!-- Persistence -->
    <dependency>
      <groupId>io.quarkiverse.jooq</groupId>
      <artifactId>quarkus-jooq</artifactId>
      <version>2.0.0</version>
    </dependency>
    <!-- Dependency required for build native image but we does not use Hibernate in application !-->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-hibernate-orm-panache</artifactId>
    </dependency>

    <!-- Liquibase - Db migrations -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-liquibase</artifactId>
    </dependency>

    <!-- Fault tolerance - Retry, CircuitBreaker annotations -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-smallrye-fault-tolerance</artifactId>
    </dependency>

    <!-- Rest Client -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-rest-client-reactive</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-rest-client-reactive-jackson</artifactId>
    </dependency>

    <!-- Hibernate Validator -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-hibernate-validator</artifactId>
    </dependency>

    <!-- Jackson -->
    <dependency>
      <groupId>com.fasterxml.jackson.datatype</groupId>
      <artifactId>jackson-datatype-jsr310</artifactId>
      <version>2.16.0</version>
    </dependency>

    <!-- Kafka - Consume Kafka Messages -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-smallrye-reactive-messaging-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-resteasy-reactive-jackson</artifactId>
    </dependency>

    <!-- Extension required for read avro schemas in Apicurio Schema Registry -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-apicurio-registry-avro</artifactId>
    </dependency>

    <!-- Avro -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-avro</artifactId>
    </dependency>

    <!-- Lombok - Generator of boilerplate code -->
    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
      <version>${lombok.version}</version>
      <scope>provided</scope>
    </dependency>

    <!-- MapStruct -->
    <dependency>
      <groupId>org.mapstruct</groupId>
      <artifactId>mapstruct</artifactId>
      <version>${org.mapstruct.version}</version>
    </dependency>

    <!-- Status -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-smallrye-health</artifactId>
    </dependency>

    <!-- Metrics -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-micrometer</artifactId>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-micrometer-registry-prometheus</artifactId>
    </dependency>
    <dependency>
      <groupId>org.eclipse.microprofile.rest.client</groupId>
      <artifactId>microprofile-rest-client-api</artifactId>
    </dependency>

    <!-- Telemetry -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-opentelemetry</artifactId>
    </dependency>
    <dependency>
      <groupId>io.opentelemetry.instrumentation</groupId>
      <artifactId>opentelemetry-jdbc</artifactId>
    </dependency>

    <!-- Container -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-container-image-jib</artifactId>
    </dependency>

    <!-- Info Endpoint -->
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-info</artifactId>
    </dependency>

Thank you for the answer

[ozangunalp]

In order to use the Apicurio registry serde in your tests, you need to set it up in your tests:

  @BeforeAll
  public static void beforeAll() {
    // Set Apicurio Avro
    vertx = Vertx.vertx();
    RegistryClientFactory.setProvider(new VertxHttpClientProvider(vertx));
  }

That is something we need to mention in https://quarkus.io/guides/kafka-schema-registry-avro#testing-the-application

[cescoffier]

Do not forget to close the vert.x instance if you follow this:

@AfterAll
static void close() {
   vertx.close().toCompletationStage().toCompletableFuture().join();
}