Ability to create new bindings at runtime via the BindingLifecycleController

[jmannix-ot]

We discussed this previously on issues #3071 and #3077.

In summary, we have a library which needs to both modify existing Kafka bindings and generate new Kafka bindings at runtime during application initialisation.

Previously we were fetching the list of bindings by auto-wiring KafkaExtendedBindingProperties and calling getBindings(). Then we modified and added to the list of bindings before updating them by calling setBindings().

We encountered an issue when upgrading to spring-cloud-stream 4.1.4 caused by the change for #2974. The addition of @validated to AbstractExtendedBindingProperties broke the auto-wiring of KafkaExtendedBindingProperties.

After some discussion we agreed that the better way to handle our use case was to use BindingsLifecycleController, and @olegz made some changes for us to make it possible to fetch and modify bindings.

However, we still need a way to generate new bindings at runtime. And that is currently not supported by BindingsLifecycleController.

[olegz]

@jmannix-ot I just pushed an initial bits for the feature you are requesting.
I'll add the docs and polish code some more, but please give it a try as soon as you can as we have a milestone release coming.

BindingsLifecycleController controller = context.getBean(BindingsLifecycleController.class);
controller.defineBinding("myBinidng", false);
KafkaProducerProperties extensionProperties = controller.getExtensionProperties("myBinidng-out-0");

The boolean flag in defineBinding(..) method is for input or output

Let me know, I am keeping it open for now

[olegz]

I did few enhancements, so the following test pretty much explains what you can do

@Test
void testDefiningNewBindingAndSettingItsProperties() throws Exception {
	BindingsLifecycleController controller = context.getBean(BindingsLifecycleController.class);
	KafkaConsumerProperties consumerProperties = controller.defineInputBinding("test-input-binding");
	boolean isAutoRebalanceEnabled = consumerProperties.isAutoRebalanceEnabled();
	assertThat(isAutoRebalanceEnabled).isTrue();
	consumerProperties.setAutoRebalanceEnabled(false);
	consumerProperties = controller.getExtensionProperties("test-input-binding-in-0");
	isAutoRebalanceEnabled = consumerProperties.isAutoRebalanceEnabled();
	assertThat(isAutoRebalanceEnabled).isFalse();
}