Cannot handle multiple group-ids with same topic

[sn-intrasoft]

Describe the bug
When there multiple consumer group-id consuming the same topic with different payloads, ClassLevelKafkaListenerScanner throws an exception while scanning due to duplicate key of topics.

Dependencies and versions used
springwolf-kafka version 0.7.0.

Stack trace and error logs
i.g.s.s.asyncapi.DefaultChannelsService : An error was encountered during channel scanning with io.github.stavshamir.springwolf.asyncapi.scanners.channels.ClassLevelKafkaListenerScanner@1dd76982: Duplicate key CUSTOMER_PRODUCT_ADPT (attempted merging values ChannelItem(description=null, subscribe=null, publish=Operation(operationId=null, summary=null, description=null, tags=null, externalDocs=null, bindings={kafka=KafkaOperationBinding(groupId=GroupA, clientId=null, bindingVersion=null)}, traits=null, message={oneOf=[Message(name=java.lang.Object, title=Object, payload=io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.PayloadReference@1), Message(name=com.example.PayloadA, title=PayloadA, payload=io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.PayloadReference@1)]}), parameters=null, bindings={kafka=KafkaChannelBinding()}) and ChannelItem(description=null, subscribe=null, publish=Operation(operationId=null, summary=null, description=null, tags=null, externalDocs=null, bindings={kafka=KafkaOperationBinding(groupId=GroupB, clientId=null, bindingVersion=null)}, traits=null, message={oneOf=[Message(name=java.lang.Object, title=Object, payload=io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.PayloadReference@1), Message(name=com.example.PayloadB, title=PayloadB, payload=io.github.stavshamir.springwolf.asyncapi.types.channel.operation.message.PayloadReference@1)]}), parameters=null, bindings={kafka=KafkaChannelBinding()}))

[stavshamir]

Thanks for reporting this issue! I am swamped lately but will try to get to researching this in the coming weeks.

[sn-intrasoft]

Thanks. More specifically, this case please

@Test
public void scan_componentWithMultipleKafkaListenersAndHandlers() {
    // Given multiple @KafkaListener annotated classes with method(s) annotated with @KafkaHandler
    setClassToScan(KafkaListenerClassWithOneKafkaHandler.class, KafkaListenerClassWithMultipleKafkaHandler.class);

    // When scan is called
    Map<String, ChannelItem> actualChannels = methodLevelKafkaListenerScanner.scan();

    // Then the returned collection contains the channel with message set to oneOf
    Message fooMessage = Message.builder()
            .name(SimpleFoo.class.getName())
            .title(SimpleFoo.class.getSimpleName())
            .payload(PayloadReference.fromModelName(SimpleFoo.class.getSimpleName()))
            .build();

    Message barMessage = Message.builder()
            .name(SimpleBar.class.getName())
            .title(SimpleBar.class.getSimpleName())
            .payload(PayloadReference.fromModelName(SimpleBar.class.getSimpleName()))
            .build();

    Operation operation = Operation.builder()
            .bindings(ImmutableMap.of("kafka", new KafkaOperationBinding()))
            .message(ImmutableMap.of(ONE_OF, ImmutableSet.of(fooMessage, barMessage)))
            .build();

    ChannelItem expectedChannel = ChannelItem.builder()
            .bindings(ImmutableMap.of("kafka", new KafkaChannelBinding()))
            .publish(operation)
            .build();

    assertThat(actualChannels)
            .containsExactly(Maps.immutableEntry(TOPIC, expectedChannel));
}

[sn-intrasoft]

Thanks @stavshamir 🙏