Le nœud de ce problème est que le producteur utilise spring-cloud-stream pour envoyer des messages à Kafka, mais le consommateur utilise spring-kaka. Les raisons en sont:

• Le système existant est déjà bien établi et utilise spring-cloud-stream
• Un nouveau consommateur doit écouter plusieurs sujets en utilisant la même méthode, en se liant uniquement sur une liste csv de noms de sujets
• Il est nécessaire de consommer une collection de messages à la fois, plutôt qu'individuellement, afin que leur contenu puisse être écrit en masse dans une base de données.

Spring-cloud-stream ne permet pas actuellement au consommateur de lier un auditeur à plusieurs sujets, et il n'y a aucun moyen de consommer une collection de messages à la fois (sauf erreur de ma part). p>

J'ai trouvé une solution qui ne nécessite aucune modification du code producteur qui utilise spring-cloud-stream pour publier des messages sur Kafka. Spring-cloud-stream utilise un MessageConverter pour gérer la sérialisation et la désérialisation. Dans le AbstractAvroMessageConverter , il existe des méthodes: convertFromInternal et convertToInternal qui gèrent la transformation vers / depuis un tableau d'octets. Ma solution était d'étendre ce code (en créant une classe qui étend AvroSchemaRegistryClientMessageConverter ), afin que je puisse réutiliser une grande partie de la fonctionnalité spring-cloud-stream, mais avec une interface accessible depuis mon spring-kafka KafkaListener . J'ai ensuite modifié mon TopicListener pour utiliser cette classe pour effectuer la conversion:

Le convertisseur:

@Slf4j
@Component
@RequiredArgsConstructor
public class TopicListener {

  private final AvroKafkaMessageConverter messageConverter;

  @KafkaListener(topics = {"test-request"})
  public void listenForMessage(ConsumerRecord<?, ?> consumerRecord) {
    log.info("listenForMessage. got a message: {}", consumerRecord);
    Request request = messageConverter.convertFromInternal(
        consumerRecord, Request.class, MimeType.valueOf("application/vnd.*+avr"));
    log.info("request message: {}", request.getMessage());
  }
}

Le TopicListener modifié:

@Component
@Slf4j
public class AvroKafkaMessageConverter extends AvroSchemaRegistryClientMessageConverter {

  public AvroKafkaMessageConverter(SchemaRegistryClient schemaRegistryClient) {
    super(schemaRegistryClient, new NoOpCacheManager());
  }

  public <T> T convertFromInternal(ConsumerRecord<?, ?> consumerRecord, Class<T> targetClass,
      Object conversionHint) {
    T result;
    try {
      byte[] payload = (byte[]) consumerRecord.value();

      Map<String, String> headers = new HashMap<>();
      consumerRecord.headers().forEach(header -> headers.put(header.key(), asString(header.value())));

      MimeType mimeType = messageMimeType(conversionHint, headers);
      if (mimeType == null) {
        return null;
      }

      Schema writerSchema = resolveWriterSchemaForDeserialization(mimeType);
      Schema readerSchema = resolveReaderSchemaForDeserialization(targetClass);

      @SuppressWarnings("unchecked")
      DatumReader<Object> reader = getDatumReader((Class<Object>) targetClass, readerSchema, writerSchema);
      Decoder decoder = DecoderFactory.get().binaryDecoder(payload, null);
      result = (T) reader.read(null, decoder);
    }
    catch (IOException e) {
      throw new RuntimeException("Failed to read payload", e);
    }
    return result;
  }

  private MimeType messageMimeType(Object conversionHint, Map<String, String> headers) {
    MimeType mimeType;
    try {
      String contentType = headers.get(MessageHeaders.CONTENT_TYPE);
      log.debug("contentType: {}", contentType);
      mimeType = MimeType.valueOf(contentType);
    } catch (InvalidMimeTypeException e) {
      log.error("Exception getting object MimeType from contentType header", e);
      if (conversionHint instanceof MimeType) {
        mimeType = (MimeType) conversionHint;
      }
      else {
        return null;
      }
    }
    return mimeType;
  }

  private String asString(byte[] byteArray) {
    String theString = new String(byteArray, Charset.defaultCharset());
    return theString.replace("\"", "");
  }
}

Cette solution ne consomme qu’un message à la fois mais peut être facilement modifiée pour consommer des lots de messages.

La solution complète est ici: https://github.com/robjwilkins/avro-example/tree/develop

Merci d'avoir sauvé ma journée en utilisant le codage natif et le printemps: nuage: stream:

Propriétés de liaison générique

  kafka:
    bindings:
      input:
        consumer:
          configuration:
            value.deserializer: com.example.FooDeserializer
      output:
        producer:
          configuration:
            value.serializer: com.example.FooSerializer

Propriétés de liaison spécifiques à Kafka

  bindings:
    input:
      consumer:
        use-native-decoding: true
      destination: so54448732
      group: so54448732
    output:
      destination: so54448732
      producer:
        use-native-encoding: true