diff --git a/applications/processor/aggregator-processor/README.adoc b/applications/processor/aggregator-processor/README.adoc
index 0507a27a..298793c2 100644
--- a/applications/processor/aggregator-processor/README.adoc
+++ b/applications/processor/aggregator-processor/README.adoc
@@ -9,6 +9,10 @@ Change kafka to rabbit if you want to run it against RabbitMQ.
 
 === Payload
 
+If an input payload is a `byte[]` and content-type header is a JSON, then `JsonBytesToMap` function tries to deserialize this payload to a `Map` for better data representation on the output of the aggregator function.
+Also, such a `Map` data representation makes it easy to access to the payload content from SpEL expressions mentioned below.
+Otherwise(including a deserialization error), the input payload is left as is - and it is the target application configuration to convert it into a desired form.
+
 == Options
 
 //tag::configuration-properties[]
diff --git a/applications/processor/aggregator-processor/pom.xml b/applications/processor/aggregator-processor/pom.xml
index ec55aa3b..c92a7ad4 100644
--- a/applications/processor/aggregator-processor/pom.xml
+++ b/applications/processor/aggregator-processor/pom.xml
@@ -32,7 +32,7 @@
                         <type>processor</type>
                         <version>${project.version}</version>
                         <configClass>org.springframework.cloud.fn.aggregator.AggregatorFunctionConfiguration.class</configClass>
-                        <functionDefinition>aggregatorFunction</functionDefinition>
+                        <functionDefinition>jsonBytesToMap|aggregatorFunction</functionDefinition>
                         <maven>
                             <dependencies>
                                 <dependency>
diff --git a/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java b/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java
index 1589f1c7..5a6475c6 100644
--- a/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java
+++ b/applications/processor/aggregator-processor/src/test/java/org/springframework/cloud/fn/aggregator/AggregatorProcessorTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-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.
@@ -16,6 +16,11 @@
 
 package org.springframework.cloud.fn.aggregator;
 
+import java.io.IOException;
+import java.util.List;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.WebApplicationType;
@@ -34,41 +39,58 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class AggregatorProcessorTests {
 
 	@Test
-	public void testWithJdbcMessageStore() {
+	public void testWithJdbcMessageStore() throws IOException {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
 				TestChannelBinderConfiguration.getCompleteConfiguration(AggregatorProcessorTestApplication.class))
 				.web(WebApplicationType.NONE)
-				.run("--spring.cloud.function.definition=aggregatorFunction",
-						"--aggregator.message-store-type=jdbc")) {
+				.run("--spring.cloud.function.definition=jsonBytesToMap|aggregatorFunction",
+						"--aggregator.message-store-type=jdbc",
+						"--aggregator.release=size()==2 or one.payload instanceof T(java.util.Map)")) {
 
 			InputDestination processorInput = context.getBean(InputDestination.class);
 			OutputDestination processorOutput = context.getBean(OutputDestination.class);
 
-			processorInput.send(
-					MessageBuilder.withPayload("2")
-							.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
-							.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 2)
-							.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
-							.build());
-			processorInput.send(
-					MessageBuilder.withPayload("1")
-							.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
-							.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 1)
-							.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 2)
-							.build());
+			processorInput.send(createMessage("2", 2, 2));
 
-			Message<byte[]> receive = processorOutput.receive(10_000, "aggregatorFunction-out-0");
+			processorInput.send(createMessage("1", 1, 2));
 
-			assertThat(receive).isNotNull()
+			Message<byte[]> receive = processorOutput.receive(10_000, "jsonBytesToMapaggregatorFunction-out-0");
+
+			assertThat(receive)
 					.extracting(Message::getPayload)
 					.extracting(String::new)
 					.isEqualTo("[\"2\",\"1\"]");
+
+			ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
+
+			Person person = new Person("First1 Last1", "St. #1");
+			processorInput.send(createMessage(objectMapper.writeValueAsBytes(person), 2, 2));
+
+			receive = processorOutput.receive(10_000, "jsonBytesToMapaggregatorFunction-out-0");
+
+			assertThat(receive).isNotNull();
+			List<Person> result =
+					objectMapper.readValue(receive.getPayload(),
+							objectMapper.constructType(new TypeReference<List<Person>>() { }));
+
+			assertThat(result).containsOnly(person);
 		}
 	}
 
+	private static Message<?> createMessage(Object payload, int sequenceNumber, int sequenceSize) {
+		return MessageBuilder.withPayload(payload)
+				.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
+				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber)
+				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize)
+				.build();
+	}
+
 	@SpringBootApplication
 	public static class AggregatorProcessorTestApplication {
 
 	}
 
+	private record Person(String name, String address) {
+	}
+
 }
diff --git a/functions/function/aggregator-function/pom.xml b/functions/function/aggregator-function/pom.xml
index ce11a214..57a24503 100644
--- a/functions/function/aggregator-function/pom.xml
+++ b/functions/function/aggregator-function/pom.xml
@@ -15,6 +15,11 @@
     <description>Spring Native Function for Aggregator</description>
 
     <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>payload-converter-function</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.integration</groupId>
             <artifactId>spring-integration-core</artifactId>
diff --git a/functions/function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java b/functions/function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java
new file mode 100644
index 00000000..a1bdc9df
--- /dev/null
+++ b/functions/function/payload-converter-function/src/main/java/functions/JsonBytesToMap.java
@@ -0,0 +1,80 @@
+/*
+ * 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 functions;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.function.Function;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.springframework.core.log.LogAccessor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeTypeUtils;
+
+/**
+ * The {@link Function} to deserialize {@code byte[]} payload into a Map
+ * if {@link MessageHeaders#CONTENT_TYPE} header is JSON.
+ * Otherwise, the message is returned as is.
+ *
+ * @author Artem Bilan
+ *
+ * @since 4.0
+ */
+public class JsonBytesToMap implements Function<Message<?>, Message<?>> {
+
+	private static final LogAccessor logger = new LogAccessor(JsonBytesToMap.class);
+
+	private final ObjectMapper objectMapper;
+
+	public JsonBytesToMap(ObjectMapper objectMapper) {
+		this.objectMapper = objectMapper;
+	}
+
+	@Override
+	public Message<?> apply(Message<?> message) {
+		if (message.getPayload() instanceof byte[] payload) {
+			MessageHeaders headers = message.getHeaders();
+			String contentType =
+					headers.containsKey(MessageHeaders.CONTENT_TYPE)
+							? headers.get(MessageHeaders.CONTENT_TYPE).toString()
+							: MimeTypeUtils.APPLICATION_JSON_VALUE;
+
+			if (contentType.contains("json")) {
+				message = MessageBuilder.withPayload(payloadToMapIfCan(payload))
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+		}
+
+		return message;
+	}
+
+	private Object payloadToMapIfCan(byte[] payload) {
+		try {
+			return this.objectMapper.readValue(payload, Map.class);
+		}
+		catch (IOException ex) {
+			logger.trace(ex, "Cannot deserialize to Map");
+			// Was not able to construct the map from byte[] -- returning as is
+			return payload;
+		}
+	}
+
+}