Introduce HeaderFilterSpec to streamline DSL API (#8636)

* Introduce HeaderFilterSpec to streamline DSL API

The concern has been driven by the discussion from: https://github.com/spring-projects/spring-integration/issues/8625
The point is that Java method arguments are not so descriptive when we read the code.
Therefore, it is better to design DSL the way it would be cleaner from reading perspective.
Plus less choice of methods to chain would give a better end-user experience from coding.

* Add a `HeaderFilterSpec` which can accept `headersToRemove` and `patternMatch` as individual
options instead of top-level deprecated `headerFilter(headersToRemove, patternMatch)` `IntegrationFlow` method.
This way Kotlin and Groovy DSLs get a gain from their "inner section" style.
* Such a `Consumer<HeaderFilterSpec>` way to configure an endpoint is similar to already
existing `aggregate(Consumer<AggregatorSpec>)`, `resequence(Consumer<ResequencerSpec>)` etc.
In other words those components which has a dedicated `ConsumerEndpointSpec` extension are OK
from an idiomatic DSL style perspective
* Expose a `HeaderFilter.setHeadersToRemove()` to make it working smoothly with this new
DSL requirements
* Apply a new `headerFilter()` style into Kotlin and Groovy DSLs

This is just an initial work to surface an idea.
If it is OK, I'll slow continue with others to realign and simplify the paradox of choice.

* * Fix asterisk imports in the `KotlinIntegrationFlowDefinition`

spring-integration-groovy/src/main/groovy/org/springframework/integration/groovy/dsl/GroovyIntegrationFlowDefinition.groovy

@@ -35,6 +35,7 @@ import org.springframework.integration.dsl.FilterEndpointSpec
 import org.springframework.integration.dsl.GatewayEndpointSpec
 import org.springframework.integration.dsl.GenericEndpointSpec
 import org.springframework.integration.dsl.HeaderEnricherSpec
+import org.springframework.integration.dsl.HeaderFilterSpec
 import org.springframework.integration.dsl.IntegrationFlow
 import org.springframework.integration.dsl.IntegrationFlowDefinition
 import org.springframework.integration.dsl.MessageChannelSpec
@@ -818,8 +819,10 @@ class GroovyIntegrationFlowDefinition {
 	 * {@link HeaderFilter}.
 	 * @param headerFilter the {@link HeaderFilter} to use.
 	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
+	 * @deprecated since 6.2 in favor of {@link #headerFilter(groovy.lang.Closure)}
 	 * @see GenericEndpointSpec
 	 */
+	@Deprecated(since = "6.2", forRemoval = true)
 	GroovyIntegrationFlowDefinition headerFilter(
 			String headersToRemove,
 			boolean patternMatch = true,
@@ -834,6 +837,21 @@ class GroovyIntegrationFlowDefinition {
 		this
 	}
 
+	/**
+	 * Populate {@link HeaderFilter} based on the options from a {@link HeaderFilterSpec}.
+	 * @param endpointConfigurer the {@link Consumer} to provide {@link HeaderFilter} and its endpoint options.
+	 * @see HeaderFilterSpec
+	 * @since 6.2
+	 */
+	GroovyIntegrationFlowDefinition headerFilter(
+			@DelegatesTo(value = HeaderFilterSpec, strategy = Closure.DELEGATE_FIRST)
+			@ClosureParams(value = SimpleType.class, options = 'org.springframework.integration.dsl.HeaderFilterSpec')
+					Closure<?> headerFilterConfigurer) {
+
+		this.delegate.headerFilter createConfigurerIfAny(headerFilterConfigurer)
+		this
+	}
+
 	/**
 	 * Populate the {@link MessageTransformingHandler} for the
 	 * {@link org.springframework.integration.transformer.ClaimCheckInTransformer} with provided {@link MessageStore}.

spring-integration-groovy/src/test/groovy/org/springframework/integration/groovy/dsl/test/GroovyDslTests.groovy

@@ -189,11 +189,18 @@ class GroovyDslTests {
 	@Test
 	void 'flow from lambda'() {
 		def replyChannel = new QueueChannel()
-		def message = MessageBuilder.withPayload('test').setReplyChannel(replyChannel).build()
+		def message =
+				MessageBuilder.withPayload('test')
+						.setHeader('headerToRemove', 'no value')
+						.setReplyChannel(replyChannel)
+						.build()
 
 		this.flowLambdaInput.send message
 
-		assert replyChannel.receive(10_000)?.payload == 'TEST'
+		def receive = replyChannel.receive(10_000)
+
+		assert receive?.payload == 'TEST'
+		assert !receive?.headers?.containsKey('headerToRemove')
 		assert this.wireTapChannel.receive(10_000)?.payload == 'test'
 	}
 
@@ -300,6 +307,10 @@ class GroovyDslTests {
 		flowLambda() {
 			integrationFlow {
 				filter String, { it == 'test' }, { id 'filterEndpoint' }
+				headerFilter {
+					patternMatch false
+					headersToRemove "notAHeader", "headerToRemove"
+				}
 				wireTap integrationFlow {
 					channel { queue 'wireTapChannel' }
 				}