Added header-filter function and processor.

README.adoc

@@ -30,61 +30,81 @@ The following are the various components of this repository.
 |link:functions/supplier/cdc-debezium-supplier/README.adoc[CDC Debezium]
 |link:functions/function/aggregator-function/README.adoc[Aggregator]
 |link:functions/consumer/analytics-consumer/README.adoc[Analytics]
+
 |link:functions/supplier/file-supplier/README.adoc[File]
 |link:functions/function/filter-function/README.adoc[Filter]
 |link:functions/consumer/cassandra-consumer/README.adoc[Cassandra]
+
 |link:functions/supplier/ftp-supplier/README.adoc[FTP]
 |link:functions/function/header-enricher-function/README.adoc[Header-Enricher]
 |link:functions/consumer/elasticsearch-consumer/README.adoc[Elasticsearch]
 |
-|link:functions/function/http-request-function/README.adoc[HTTP Request]
+|link:functions/function/header-filter-function/README.adoc[Header-Filter]
 |link:functions/consumer/file-consumer/README.adoc[File]
+
 |link:functions/supplier/http-supplier/README.adoc[HTTP]
-|link:functions/function/image-recognition-function/README.adoc[Image Recognition(Tensorflow)]
+|link:functions/function/http-request-function/README.adoc[HTTP Request]
 |link:functions/consumer/ftp-consumer/README.adoc[FTP]
+
 |link:functions/supplier/jdbc-supplier/README.adoc[JDBC]
-|link:functions/function/object-detection-function/README.adoc[Object Detection(Tensorflow)]
+|link:functions/function/image-recognition-function/README.adoc[Image Recognition(Tensorflow)]
 |
 |link:functions/supplier/jms-supplier/README.adoc[JMS]
-|link:functions/function/semantic-segmentation-function/README.adoc[Semantic Segmentation(Tensorflow)]
+
+|link:functions/function/object-detection-function/README.adoc[Object Detection(Tensorflow)]
 |link:functions/consumer/jdbc-consumer/README.adoc[JDBC]
+
 |link:functions/supplier/mail-supplier/README.adoc[Mail]
-|link:functions/function/spel-function/README.adoc[SpEL]
+
+|link:functions/function/semantic-segmentation-function/README.adoc[Semantic Segmentation(Tensorflow)]
 |link:functions/consumer/log-consumer/README.adoc[Log]
+
 |link:functions/supplier/mongodb-supplier/README.adoc[MongoDB]
-|link:functions/function/splitter-function/README.adoc[Splitter]
+
+|link:functions/function/spel-function/README.adoc[SpEL]
 |link:functions/consumer/mongodb-consumer/README.adoc[MongoDB]
+
 |link:functions/supplier/mqtt-supplier/README.adoc[MQTT]
-|link:functions/function/task-launch-request-function/README.adoc[Task Launch Request]
+
+|link:functions/function/splitter-function/README.adoc[Splitter]
 |link:functions/consumer/mqtt-consumer/README.adoc[MQTT]
+
 |link:functions/supplier/rabbit-supplier/README.adoc[RabbitMQ]
-|link:functions/function/tasklauncher-function/README.adoc[Task Launcher]
+|link:functions/function/task-launch-request-function/README.adoc[Task Launch Request]
 |link:functions/consumer/rabbit-consumer/README.adoc[RabbitMQ]
+
 |link:functions/supplier/s3-supplier/README.adoc[AWS S3]
-|link:functions/function/twitter-function/README.adoc[Twitter]
+|link:functions/function/tasklauncher-function/README.adoc[Task Launcher]
 |link:functions/consumer/redis-consumer/README.adoc[Redis]
+
 |link:functions/supplier/sftp-supplier/README.adoc[SFTP]
 |
 |link:functions/consumer/rsocket-consumer/README.adoc[RSocket]
+
 |link:functions/supplier/syslog-supplier/README.adoc[Syslog]
 |
 |link:functions/consumer/s3-consumer/README.adoc[AWS S3]
+
 |link:functions/supplier/tcp-supplier/README.adoc[TCP]
 |
 |link:functions/consumer/sftp-consumer/README.adoc[SFTP]
+
 |link:functions/supplier/time-supplier/README.adoc[Time]
 |
 |link:functions/consumer/tcp-consumer/README.adoc[TCP]
+
 |link:functions/supplier/twitter-supplier/README.adoc[Twitter]
-|
+|link:functions/function/twitter-function/README.adoc[Twitter]
 |link:functions/consumer/twitter-consumer/README.adoc[Twitter]
+
 |link:functions/supplier/websocket-supplier/README.adoc[Websocket]
 |
 |link:functions/consumer/websocket-consumer/README.adoc[Websocket]
-|link:functions/supplier/xmpp-supplier/README.adoc[XMPP]
+
+|
 |
 |link:functions/consumer/wavefront-consumer/README.adoc[Wavefront]
-|
+|link:functions/supplier/xmpp-supplier/README.adoc[XMPP]
 |
 |link:functions/consumer/xmpp-consumer/README.adoc[XMPP]
 |===

applications/processor/header-filter-processor/README.adoc

@@ -0,0 +1,17 @@
+//tag::ref-doc[]
+= Header Filter Processor
+
+Use the header-filter app to remove message headers.
+
+The names of headers to remove are presented as a comma or space separated list. For example `--remove='foo,bar,pf-*'`.
+
+== Options
+
+The **$$header-filter$$** $$processor$$ has the following options:
+
+//tag::configuration-properties[]
+$$header.filter.delete-all$$:: $$Indicates the need to remove all headers.$$ *($$Boolean$$, default: `$$false$$`)*
+$$header.filter.remove$$:: $$Remove all headers named. A comma, space separated list of header names. The names may contain patterns.$$ *($$String$$, default: `$$<none>$$`)*
+//end::configuration-properties[]
+
+//end::ref-doc[]

applications/processor/header-filter-processor/pom.xml

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>header-filter-processor</artifactId>
+    <name>header-filter-processor</name>
+    <description>header-filter processor apps</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.stream.app</groupId>
+        <artifactId>stream-applications-core</artifactId>
+        <version>4.0.0-SNAPSHOT</version>
+        <relativePath>../../stream-applications-core/pom.xml</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>header-filter-function</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test-support</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-deploy-plugin</artifactId>
+                <version>3.0.0</version>
+                <configuration>
+                    <skip>false</skip>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.springframework.cloud</groupId>
+                <artifactId>spring-cloud-dataflow-apps-docs-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.springframework.cloud</groupId>
+                <artifactId>spring-cloud-dataflow-apps-generator-plugin</artifactId>
+                <configuration>
+                    <application>
+                        <name>header-filter</name>
+                        <type>processor</type>
+                        <version>${project.version}</version>
+                        <configClass>org.springframework.cloud.fn.header.filter.HeaderFilterFunctionConfiguration.class</configClass>
+                        <functionDefinition>headerFilterFunction</functionDefinition>
+                        <maven>
+                            <dependencies>
+                                <dependency>
+                                    <groupId>org.springframework.cloud.fn</groupId>
+                                    <artifactId>header-filter-function</artifactId>
+                                </dependency>
+                            </dependencies>
+                        </maven>
+                    </application>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

applications/processor/header-filter-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties

@@ -0,0 +1,2 @@
+configuration-properties.classes=org.springframework.cloud.fn.header.filter.HeaderFilterFunctionProperties
+

applications/processor/header-filter-processor/src/main/resources/META-INF/dataflow-configuration-metadata.properties

@@ -0,0 +1,2 @@
+configuration-properties.classes=org.springframework.cloud.fn.header.filter.HeaderFilterFunctionProperties
+

applications/processor/header-filter-processor/src/test/java/org/springframework/cloud/stream/app/processor/header/filter/HeaderFilterProcessorTests.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.app.processor.header.filter;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.fn.header.filter.HeaderFilterFunctionConfiguration;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Christian Tzolov
+ * @author Soby Chacko
+ */
+public class HeaderFilterProcessorTests {
+
+	@Test
+	public void testHeaderFilterProcessor() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(HeaderEnricherProcessorTestApplication.class))
+			.web(WebApplicationType.NONE)
+			.run("--spring.cloud.function.definition=headerFilterFunction",
+				"--header.filter.remove=foo,bar")) {
+
+			InputDestination processorInput = context.getBean(InputDestination.class);
+			OutputDestination processorOutput = context.getBean(OutputDestination.class);
+
+			final Message<?> message = MessageBuilder.withPayload("hello")
+				.setHeader("bar", "foo")
+				.setHeader("foo", "bar")
+				.setHeader("foo-bar", "fubar")
+				.build();
+			processorInput.send(message);
+			Message<byte[]> results = processorOutput.receive(10000);
+			Set<String> headers = getNonReadOnlyHeaders(results);
+			assertThat(headers).isEqualTo(Set.of("foo-bar", "contentType", "target-protocol"));
+		}
+	}
+
+	@NotNull
+	private static Set<String> getNonReadOnlyHeaders(Message<byte[]> message) {
+		var headers = new HashSet<>(message.getHeaders().keySet());
+		var accessor = new IntegrationMessageHeaderAccessor(message);
+		headers.removeIf(accessor::isReadOnly);
+		return headers;
+	}
+
+	@Test
+	public void testHeaderFilterRemoveAllProcessor() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+			TestChannelBinderConfiguration.getCompleteConfiguration(HeaderEnricherProcessorTestApplication.class))
+			.web(WebApplicationType.NONE)
+			.run("--spring.cloud.function.definition=headerFilterFunction",
+				"--header.filter.delete-all=true")) {
+
+			InputDestination processorInput = context.getBean(InputDestination.class);
+			OutputDestination processorOutput = context.getBean(OutputDestination.class);
+
+			final Message<?> message = MessageBuilder.withPayload("hello")
+				.setHeader("bar", "foo")
+				.setHeader("foo", "bar")
+				.setHeader("foo-bar", "fubar")
+				.build();
+			processorInput.send(message);
+			Message<byte[]> result = processorOutput.receive(10000);
+			var headers = getNonReadOnlyHeaders(result);
+			assertThat(headers).isEqualTo(Set.of("contentType"));
+		}
+	}
+
+	@SpringBootApplication
+	@Import({HeaderFilterFunctionConfiguration.class})
+	public static class HeaderEnricherProcessorTestApplication {
+
+		@Bean
+		public String value() {
+			return "beanValue";
+		}
+	}
+
+}

applications/processor/pom.xml

@@ -15,6 +15,7 @@
         <module>filter-processor</module>
         <module>groovy-processor</module>
         <module>header-enricher-processor</module>
+        <module>header-filter-processor</module>
         <module>http-request-processor</module>
         <module>splitter-processor</module>
         <module>transform-processor</module>

build-apps.sh

@@ -0,0 +1,35 @@
+#!/bin/bash
+SCDIR=$(dirname "$(readlink -f "${BASH_SOURCE[0]}")")
+SCDIR=$(realpath "$SCDIR")
+(return 0 2>/dev/null) && sourced=1 || sourced=0
+function check_env() {
+  eval ev='$'$1
+  if [ "$ev" == "" ]; then
+    echo "$1 not defined"
+    if (( sourced != 0 )); then
+      return 1
+    else
+      exit 1
+    fi
+  fi
+}
+
+if [ "$1" = "" ]; then
+  MAVEN_GOAL="install"
+else
+  MAVEN_GOAL="$*"
+fi
+
+$SCDIR/create-matrices.sh
+PROCESSORS=$(jq -c '.processors | .[]' matrix.json | sed 's/\"//g')
+SINKS=$(jq -c '.sinks | .[]' matrix.json | sed 's/\"//g')
+SOURCES=$(jq -c '.sources | .[]' matrix.json | sed 's/\"//g')
+for app in $PROCESSORS; do
+  $SCDIR/build-app.sh . "applications/processor/$app"
+done
+for app in $SOURCES; do
+  $SCDIR/build-app.sh . "applications/source/$app"
+done
+for app in $SINKS; do
+  $SCDIR/build-app.sh . "applications/sink/$app"
+done

build-release-train.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+SCDIR=$(dirname "$(readlink -f "${BASH_SOURCE[0]}")")
+SCDIR=$(realpath "$SCDIR")
+(return 0 2>/dev/null) && sourced=1 || sourced=0
+function check_env() {
+  eval ev='$'$1
+  if [ "$ev" == "" ]; then
+    echo "$1 not defined"
+    if (( sourced != 0 )); then
+      return 1
+    else
+      exit 1
+    fi
+  fi
+}
+
+if [ "$1" = "" ]; then
+  MAVEN_GOAL="install"
+else
+  MAVEN_GOAL="$*"
+fi
+
+$SCDIR/build-folder.sh stream-applications-release-train "$MAVEN_GOAL"

functions/function-dependencies/pom.xml

@@ -227,6 +227,11 @@
                 <artifactId>header-enricher-function</artifactId>
                <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.springframework.cloud.fn</groupId>
+                <artifactId>header-filter-function</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>org.springframework.cloud.fn</groupId>
                 <artifactId>http-request-function</artifactId>

functions/function/header-filter-function/README.adoc

@@ -0,0 +1,25 @@
+= Header Enricher Function
+
+This module provides a header enricher function that can be reused and composed in other applications.
+
+== Beans for injection
+
+You can import the `HeaderEnricherFunctionConfiguration` in a Spring Boot application and then inject the following bean.
+
+`headerFilterFunction`
+
+You can use `headerFilterFunction` as a qualifier when injecting.
+
+Once injected, you can use the `apply` method of the `Function` to invoke it and get the result.
+
+== Configuration Options
+
+For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/header/filter/HeaderFilterFunctionProperties.java[HeaderFilterFunctionProperties.java]
+
+== Tests
+
+See this link:src/test/java/org/springframework/cloud/fn/header/filter/HeaderFilterFunctionApplicationTests.java[test suite] for examples of how this function is used.
+
+== Other usage
+
+See this link:../../../applications/processor/header-filter-processor/README.adoc[README] where this function is used to create a Spring Cloud Stream application.

functions/function/header-filter-function/pom.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>4.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent/pom.xml</relativePath>
+    </parent>
+
+    <artifactId>header-filter-function</artifactId>
+    <name>header-filter-function</name>
+    <description>Spring Native Function for applying message filters</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>payload-converter-function</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

functions/function/header-filter-function/src/main/java/org/springframework/cloud/fn/header/filter/HeaderFilterFunctionConfiguration.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.header.filter;
+
+import java.util.HashSet;
+import java.util.function.Function;
+
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.transformer.HeaderFilter;
+import org.springframework.messaging.Message;
+import org.springframework.util.StringUtils;
+
+/**
+ * Configure a function using {@link HeaderFilter}.
+ *
+ * @author Corneil du Plessis
+ */
+@AutoConfiguration
+@EnableConfigurationProperties(HeaderFilterFunctionProperties.class)
+@ConditionalOnExpression("'${header.filter.remove}'!='' or '${header.filter.delete-all}' != ''")
+public class HeaderFilterFunctionConfiguration {
+	private final HeaderFilterFunctionProperties properties;
+	public HeaderFilterFunctionConfiguration(HeaderFilterFunctionProperties properties) {
+		this.properties = properties;
+	}
+
+	@Bean
+	public Function<Message<?>, Message<?>> headerFilterFunction() {
+		if (properties.isDeleteAll()) {
+			return (message) -> {
+				var  accessor = new IntegrationMessageHeaderAccessor(message);
+				var headers = new HashSet<>(message.getHeaders().keySet());
+				headers.removeIf(accessor::isReadOnly);
+				HeaderFilter filter = new HeaderFilter(headers.toArray(new String[0]));
+				return filter.transform(message);
+			};
+		}
+		else {
+			return headerFilter()::transform;
+		}
+	}
+
+	@Bean
+	public HeaderFilter headerFilter() {
+		if (properties.getRemove() != null) {
+			String[] remove = StringUtils.tokenizeToStringArray(properties.getRemove(), ", ", true, true);
+			HeaderFilter filter = new HeaderFilter(remove);
+			if (properties.getRemove().contains("*")) {
+				filter.setPatternMatch(true);
+			}
+			return filter;
+		}
+		else {
+			return new HeaderFilter("");
+		}
+	}
+
+}

functions/function/header-filter-function/src/main/java/org/springframework/cloud/fn/header/filter/HeaderFilterFunctionProperties.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.header.filter;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Properties for configuration of header-filter-function.
+ * @author Corneil du Plessis
+ */
+@ConfigurationProperties("header.filter")
+public class HeaderFilterFunctionProperties {
+	/**
+	 * Indicates the need to remove all headers.
+	 */
+	private boolean deleteAll = false;
+
+	/**
+	 * Remove all headers named. A comma, space separated list of header names.
+	 * The names may contain patterns.
+	 */
+	private String remove;
+
+	public boolean isDeleteAll() {
+		return deleteAll;
+	}
+
+	public void setDeleteAll(boolean deleteAll) {
+		this.deleteAll = deleteAll;
+	}
+
+	public String getRemove() {
+		return remove;
+	}
+
+	public void setRemove(String remove) {
+		this.remove = remove;
+	}
+}

functions/function/header-filter-function/src/test/java/org/springframework/cloud/fn/header/filter/HeaderFilterFunctionApplicationDeleteAllTests.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.header.filter;
+
+import java.util.Set;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(classes = {
+	HeaderFilterFunctionApplicationDeleteAllTests.HeaderFilterFunctionTestApplication.class,
+	HeaderFilterFunctionConfiguration.class
+},
+	properties = {"header.filter.delete-all=true"}
+)
+public class HeaderFilterFunctionApplicationDeleteAllTests {
+	@Autowired
+	protected Function<Message<?>, Message<?>> headerFilter;
+
+	@Test
+	public void testRemoveLeavesIdTimestampAll() {
+		// given
+		final Message<?> message = MessageBuilder.withPayload("hello")
+			.setHeader("foo", "bar")
+			.setHeader("bar", "foo")
+			.build();
+		Message<?> result = headerFilter.apply(message);
+		var headers = result.getHeaders().keySet();
+		assertThat(headers).isEqualTo(Set.of("id", "timestamp"));
+	}
+
+	@SpringBootApplication
+	static class HeaderFilterFunctionTestApplication {
+		public static void main(String[] args) throws Exception {
+			SpringApplication.main(args);
+		}
+	}
+
+}

functions/function/header-filter-function/src/test/java/org/springframework/cloud/fn/header/filter/HeaderFilterFunctionApplicationTests.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.header.filter;
+
+import java.util.Set;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest(classes = {
+	HeaderFilterFunctionApplicationTests.HeaderFilterFunctionTestApplication.class,
+	HeaderFilterFunctionConfiguration.class
+},
+	properties = {"header.filter.remove=foo,bar,pf-*"}
+)
+public class HeaderFilterFunctionApplicationTests {
+	@Autowired
+	protected Function<Message<?>, Message<?>> headerFilter;
+
+	@Test
+	public void testRemoveAll() {
+		// given
+		final Message<?> message = MessageBuilder.withPayload("hello")
+			.setHeader("foo", "bar")
+			.setHeader("bar", "foo")
+			.build();
+		Message<?> result = headerFilter.apply(message);
+		var headers = HeaderUtils.getNonReadOnlyHeaders(result);
+		assertThat(headers).isEmpty();
+	}
+
+	@Test
+	public void testRemoveSome() {
+		// given
+		final Message<?> message = MessageBuilder.withPayload("hello")
+			.setHeader("foo", "bar")
+			.setHeader("foo-bar", "bar")
+			.setHeader("bar", "foo")
+			.build();
+		Message<?> result = headerFilter.apply(message);
+		var headers = HeaderUtils.getNonReadOnlyHeaders(result);
+		assertThat(headers).isEqualTo(Set.of("foo-bar"));
+	}
+
+	@Test
+	public void testRemoveSomeWithWildcard() {
+		// given
+		final Message<?> message = MessageBuilder.withPayload("hello")
+			.setHeader("foo", "bar")
+			.setHeader("pf-foo", "bar")
+			.setHeader("pf-bar", "bar")
+			.setHeader("pfBar", "bar")
+			.setHeader("bar", "foo")
+			.build();
+		Message<?> result = headerFilter.apply(message);
+		var headers = HeaderUtils.getNonReadOnlyHeaders(result);
+		assertThat(result.getHeaders().keySet()).isEqualTo(Set.of("pfBar"));
+	}
+
+	@Test
+	public void testRemoveLeavesIdTimestampAll() {
+		// given
+		final Message<?> message = MessageBuilder.withPayload("hello")
+			.setHeader("foo", "bar")
+			.setHeader("bar", "foo")
+			.build();
+		Message<?> result = headerFilter.apply(message);
+
+		assertThat(result.getHeaders().keySet()).isEqualTo(Set.of("id", "timestamp"));
+	}
+
+	@SpringBootApplication
+	static class HeaderFilterFunctionTestApplication {
+		public static void main(String[] main) {
+
+		}
+	}
+
+}

functions/function/header-filter-function/src/test/java/org/springframework/cloud/fn/header/filter/HeaderUtils.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2023-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.header.filter;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.jetbrains.annotations.NotNull;
+
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.messaging.Message;
+
+final public class HeaderUtils {
+	private HeaderUtils() {
+	}
+
+	@NotNull
+	public static Set<String> getNonReadOnlyHeaders(Message<?> message) {
+		var headers = new HashSet<>(message.getHeaders().keySet());
+		var accessor = new IntegrationMessageHeaderAccessor(message);
+		headers.removeIf(accessor::isReadOnly);
+		return headers;
+	}
+}

functions/function/header-filter-function/src/test/resources/test.properties

@@ -0,0 +1 @@
+logging.level.root=debug

functions/function/pom.xml

@@ -14,6 +14,7 @@
         <module>aggregator-function</module>
         <module>filter-function</module>
         <module>header-enricher-function</module>
+        <module>header-filter-function</module>
         <module>http-request-function</module>
         <module>spel-function</module>
         <module>payload-converter-function</module>