Hello, I have observed the following strange behaviour When I have a list of kafka sources constructed like this (pseudo code more or less):

List<DataStream<T>> dataStreams = kafkaSourceInfos.stream()
                .map(srcInfo -> {
                          final KafkaSource<T> kafkaSource = buildKafkaSource(srcInfo);
                          return kafkaSource;
                    });
                .toList();

Note: Imagine buildKafkaSource(..) is an auxiliary method which is creating a KafkaSource with specific properties with a specific group.id When I merge all those dataStreams with the method DataStream#union (https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/streaming/api/datastream/DataStream.html#union-org.ap[…]ataStream...-) so to have one and then wire my ProcessFunction, etc. Meaning (pseudo-code more or less):

SingleOutputStreamOperator<T> source = dataStreams
                .stream()
                .skip(1) // cause reduce will already take first element
                .reduce(dataStreams.get(0), DataStream::union)
                .map(i -> i);

....

source.keyBy(..).process(...)

The group.id is never registered (consumer group with this id) - this is cross checked from Burrow, but also from inspecting kafka internals. By not registered, I mean that kafka consumption works as it should, but the offset lag, and other metrics which I should have based on group.id...they are not there, as there is not a consumer group with this group.id. By the time I remove the union function all works fine. Is this a known issue? Has anyone else experienced that before? Thanks in advance!

I’m guessing you haven’t enabled checkpointing?

this was my first guess, but unfortunately checkpointing is enabled

My best guess is that it’s stripped because of the union, but not sure. It would require a dive into the code yes

ok, will proceed. what you want to explain with stripped? as I did not get you 100%

I’m wondering if the groupId is removed (stripped) during the union. Potentially because that value can’t get unified: you can have only one groupId