The call to applyGroupProtocol() in the constructor is always a no-op — this.groupProtocol is always null at construction time. See my review on that method.

Thanks for raising this. I have removed the method and added the property check in start() method.
I see that there is also a createInstance() method which returns a streamsBuilder, Do we need to add this in that method as well?

if (this.groupProtocol != null) guard never passes since it is called from the ctor. The setter sets the field later, but there is no path from the setter back to applying it to this.properties. This means the feature silently has no effect. Also, applyGroupProtocol() should not be a public method — it is an internal implementation detail and should not be part of the API surface. However, I think that the right fix is to remove applyGroupProtocol() entirely and apply the value inline in start(), just before initializing KafkaStreams:

  if (this.groupProtocol != null) {
      this.properties.setProperty(ConsumerConfig.GROUP_PROTOCOL_CONFIG,
          this.groupProtocol.name().toLowerCase(Locale.ROOT));
  }
  this.kafkaStreams = this.kafkaStreamsCustomizer.initKafkaStreams(
      this.topology, this.properties, this.clientSupplier
  );

Updated this as per the recommendation.

The test only verifies that getGroupProtocol() returns the correct enum value (the setter/getter round-trip), but never asserts that group.protocol actually ends up in the properties passed to KafkaStreams at start() time. A meaningful assertion would be:

  assertThat(streamsBuilderFactoryBean.getStreamsConfiguration())
      .containsEntry(ConsumerConfig.GROUP_PROTOCOL_CONFIG,
          testGroupProtocol.toLowerCase(Locale.ROOT));

Test case is updated, Can you please check comment on same ? #4373 (comment)
Please suggest if we need to update implementation for createInstance/createStreamsBuilder.

The setter should accept GroupProtocol directly rather than a String:

public void setGroupProtocol(GroupProtocol groupProtocol) {
    Assert.notNull(groupProtocol, "'groupProtocol' must not be null");
    this.groupProtocol = groupProtocol;
}

A typo like fb.setGroupProtocol("strems") compiles fine today but blows up at runtime with an IllegalArgumentException. With the typed setter, fb.setGroupProtocol(GroupProtocol.STREMS)
simply does not compile. The valid options are also immediately discoverable via IDE auto-complete.
The typical usage would then be:

@Bean
public StreamsBuilderFactoryBeanConfigurer groupProtocolConfigurer() {
    return fb -> fb.setGroupProtocol(GroupProtocol.STREAMS);
}

I have updated implementation to use GroupProtcol enum for Streams. For all the methods.

wrong package for GroupProtocol.

I have updated the comment.

Might want to consider adding an extra anchor - something like [[streams-group-protocol] and then use it to link from whats-new entry.

Updated whats-new entry with an anchor to stream doc

There is real value to get access to this property.
Please, consider to remove the getter, unless you have a strong argument why do we need it.

Removed getter.

I think we need to emit the WARN log message in the case when property is provided already in the properties and we are going to override it with this our groupProtocol.
And probably only when they are not equal 😄

I have added a warning, can you please check if wording is correct?

Thanks.
First of all that is too many this.properties.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG) call.
Consider to extract it into a local variable.

I think the wording in the warning message should be like:

The 'group.protocol=%s' property is overridden with '%s'."

Nothing more.
Also, consider to use LogMessage.format() instead of string.
That would be unfortunate to calculate string for nothing when warn is not enabled.

too many this.properties.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG)

was not addressed.

There is no need in the Locale since we turn pre-compiled enum value to string, not opposite to cause some vulnerability.

Removed the Locale usage.

Well, this is a bug.
We must not mutate shared instance.
And looks like that is exactly what happens here: we provide shared kafkaStreamsConfiguration and then we mutate its properties adding a new one in the StreamsBuilderFactoryBean.
I think KafkaStreamsConfiguration must return a fresh copy of its internal state.
We also have to make a copy in the setStreamsConfiguration(Properties streamsConfig).
And must return a copy from the getStreamsConfiguration().
All of that to avoid bug on mutation of shared objects.

However I believe that has to be done in the separate PR and back-ported.
And then when you rebase this one, all should be good in this test.

Although I still have concerns for the purpose of this test.
I believe we should aim for asserting whatever ConsumerConfig.GROUP_PROTOCOL_CONFIG is in the end for the StreamsBuilder, not streamsBuilderFactoryBean.

Right now we are adding group.protocol to KafkaStreams and not to StreamsBuilder. Might have to add group protocol property through createInstance/createStreamsBuilder in case we need to add it for streamsbuilder. I am not sure though if streamsbuilder can use group.protocol as it seems to mainly handle topology configurations?

We just must not modify shared instance.
That's the point.
An internal properties in the StreamsBuilderFactoryBean has to be isolated and modified before it is propagated down to the Kafka Streams API.

StreamsBuilderFactoryBean creates kafkaStreams in start() so Ideally the test case is just ensuring that the passed down properties when creating KafkaStreams are correct.
As we are not able to directly check KafkaStreams configuration properties, better option would be to drop the test case assuming that KafkaStreams will use properties correctly.
Please share if you have an idea on how can we handle this better.

As for mutable shared instance of properties I will create a separate PR/issue it.

Well, what I see so far:

1. StreamsBuilderFactoryBean manages a KafkaStreams object after its start.
2. The KafkaStreams is based on the provided properties (including our custom ConsumerConfig.GROUP_PROTOCOL_CONFIG)
3. Those props are stored in the KafkaStreams.applicationConfigs .
4. So, in our test we should check that provided ConsumerConfig.GROUP_PROTOCOL_CONFIG is really propagated down there:

KafkaStreams kafkaStreams = streamsBuilderFactoryBean.getKafkaStreams();
StreamsConfig streamsConfig = KafkaTestUtils.getPropertyValue(kafkaStreams, "applicationConfigs", StreamsConfig.class);
streamsConfig.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG);

What do I miss?

Thanks.
First of all that is too many this.properties.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG) call.
Consider to extract it into a local variable.

I think the wording in the warning message should be like:

The 'group.protocol=%s' property is overridden with '%s'."

Nothing more.
Also, consider to use LogMessage.format() instead of string.
That would be unfortunate to calculate string for nothing when warn is not enabled.

too many this.properties.get(ConsumerConfig.GROUP_PROTOCOL_CONFIG)

was not addressed.

I'm not why have you decided to use this version if we asked explicitly for 4.1

Why do you spy if you don't verify eventually?

Please, no unprofessional foo/bar language in new code.
I know there is a number of them around to be fixed one day, but that doesn't mean we should continue following such a bad habit.