diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilder.java
index 428936840..a18f8760f 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilder.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilder.java
@@ -31,6 +31,7 @@ import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
 import org.springframework.batch.item.file.transform.FieldExtractor;
 import org.springframework.batch.item.file.transform.FormatterLineAggregator;
 import org.springframework.batch.item.file.transform.LineAggregator;
+import org.springframework.batch.item.file.transform.RecordFieldExtractor;
 import org.springframework.core.io.WritableResource;
 import org.springframework.util.Assert;
 
@@ -290,6 +291,8 @@ public class FlatFileItemWriterBuilder<T> {
 
 		private List<String> names = new ArrayList<>();
 
+		private Class<T> sourceType;
+
 		protected FormattedBuilder(FlatFileItemWriterBuilder<T> parent) {
 			this.parent = parent;
 		}
@@ -336,6 +339,20 @@ public class FlatFileItemWriterBuilder<T> {
 			return this;
 		}
 
+		/**
+		 * Specify the type of items from which fields will be extracted. This is used to
+		 * configure the right {@link FieldExtractor} based on the given type (ie a record
+		 * or a regular class).
+		 * @param sourceType type of items from which fields will be extracted
+		 * @return The current instance of the builder.
+		 * @since 5.0
+		 */
+		public FormattedBuilder<T> sourceType(Class<T> sourceType) {
+			this.sourceType = sourceType;
+
+			return this;
+		}
+
 		/**
 		 * Set the {@link FieldExtractor} to use to extract fields from each item.
 		 * @param fieldExtractor to use to extract fields from each item
@@ -372,15 +389,20 @@ public class FlatFileItemWriterBuilder<T> {
 			formatterLineAggregator.setMaximumLength(this.maximumLength);
 
 			if (this.fieldExtractor == null) {
-				BeanWrapperFieldExtractor<T> beanWrapperFieldExtractor = new BeanWrapperFieldExtractor<>();
-				beanWrapperFieldExtractor.setNames(this.names.toArray(new String[this.names.size()]));
-				try {
-					beanWrapperFieldExtractor.afterPropertiesSet();
+				if (this.sourceType != null && this.sourceType.isRecord()) {
+					this.fieldExtractor = new RecordFieldExtractor<>(this.sourceType);
 				}
-				catch (Exception e) {
-					throw new IllegalStateException("Unable to initialize FormatterLineAggregator", e);
+				else {
+					BeanWrapperFieldExtractor<T> beanWrapperFieldExtractor = new BeanWrapperFieldExtractor<>();
+					beanWrapperFieldExtractor.setNames(this.names.toArray(new String[this.names.size()]));
+					try {
+						beanWrapperFieldExtractor.afterPropertiesSet();
+						this.fieldExtractor = beanWrapperFieldExtractor;
+					}
+					catch (Exception e) {
+						throw new IllegalStateException("Unable to initialize FormatterLineAggregator", e);
+					}
 				}
-				this.fieldExtractor = beanWrapperFieldExtractor;
 			}
 
 			formatterLineAggregator.setFieldExtractor(this.fieldExtractor);
@@ -404,6 +426,8 @@ public class FlatFileItemWriterBuilder<T> {
 
 		private FieldExtractor<T> fieldExtractor;
 
+		private Class<T> sourceType;
+
 		protected DelimitedBuilder(FlatFileItemWriterBuilder<T> parent) {
 			this.parent = parent;
 		}
@@ -419,6 +443,20 @@ public class FlatFileItemWriterBuilder<T> {
 			return this;
 		}
 
+		/**
+		 * Specify the type of items from which fields will be extracted. This is used to
+		 * configure the right {@link FieldExtractor} based on the given type (ie a record
+		 * or a regular class).
+		 * @param sourceType type of items from which fields will be extracted
+		 * @return The current instance of the builder.
+		 * @since 5.0
+		 */
+		public DelimitedBuilder<T> sourceType(Class<T> sourceType) {
+			this.sourceType = sourceType;
+
+			return this;
+		}
+
 		/**
 		 * Names of each of the fields within the fields that are returned in the order
 		 * they occur within the delimited file. These names will be used to create a
@@ -453,15 +491,20 @@ public class FlatFileItemWriterBuilder<T> {
 			}
 
 			if (this.fieldExtractor == null) {
-				BeanWrapperFieldExtractor<T> beanWrapperFieldExtractor = new BeanWrapperFieldExtractor<>();
-				beanWrapperFieldExtractor.setNames(this.names.toArray(new String[this.names.size()]));
-				try {
-					beanWrapperFieldExtractor.afterPropertiesSet();
+				if (this.sourceType != null && this.sourceType.isRecord()) {
+					this.fieldExtractor = new RecordFieldExtractor<>(this.sourceType);
 				}
-				catch (Exception e) {
-					throw new IllegalStateException("Unable to initialize DelimitedLineAggregator", e);
+				else {
+					BeanWrapperFieldExtractor<T> beanWrapperFieldExtractor = new BeanWrapperFieldExtractor<>();
+					beanWrapperFieldExtractor.setNames(this.names.toArray(new String[this.names.size()]));
+					try {
+						beanWrapperFieldExtractor.afterPropertiesSet();
+						this.fieldExtractor = beanWrapperFieldExtractor;
+					}
+					catch (Exception e) {
+						throw new IllegalStateException("Unable to initialize DelimitedLineAggregator", e);
+					}
 				}
-				this.fieldExtractor = beanWrapperFieldExtractor;
 			}
 
 			delimitedLineAggregator.setFieldExtractor(this.fieldExtractor);
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilderTests.java
index 2c175eec8..4f89b1897 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilderTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/file/builder/FlatFileItemWriterBuilderTests.java
@@ -22,12 +22,16 @@ import java.io.InputStreamReader;
 import java.nio.charset.Charset;
 import java.util.Arrays;
 
+import org.junit.Assert;
 import org.junit.Test;
 
 import org.springframework.batch.item.ExecutionContext;
-
 import org.springframework.batch.item.file.FlatFileItemWriter;
+import org.springframework.batch.item.file.transform.BeanWrapperFieldExtractor;
+import org.springframework.batch.item.file.transform.DelimitedLineAggregator;
+import org.springframework.batch.item.file.transform.FormatterLineAggregator;
 import org.springframework.batch.item.file.transform.PassThroughLineAggregator;
+import org.springframework.batch.item.file.transform.RecordFieldExtractor;
 import org.springframework.core.io.FileSystemResource;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.WritableResource;
@@ -236,6 +240,132 @@ public class FlatFileItemWriterBuilderTests {
 		validateBuilderFlags(writer, encoding);
 	}
 
+	@Test
+	public void testSetupDelimitedLineAggregatorWithRecordItemType() throws IOException {
+		// given
+		WritableResource output = new FileSystemResource(File.createTempFile("foo", "txt"));
+		record Person(int id, String name) {
+		}
+
+		// when
+		FlatFileItemWriter<Person> writer = new FlatFileItemWriterBuilder<Person>().name("personWriter")
+				.resource(output).delimited().sourceType(Person.class).names("id", "name").build();
+
+		// then
+		Object lineAggregator = ReflectionTestUtils.getField(writer, "lineAggregator");
+		Assert.assertNotNull(lineAggregator);
+		Assert.assertTrue(lineAggregator instanceof DelimitedLineAggregator);
+		Object fieldExtractor = ReflectionTestUtils.getField(lineAggregator, "fieldExtractor");
+		Assert.assertNotNull(fieldExtractor);
+		Assert.assertTrue(fieldExtractor instanceof RecordFieldExtractor);
+	}
+
+	@Test
+	public void testSetupDelimitedLineAggregatorWithClassItemType() throws IOException {
+		// given
+		WritableResource output = new FileSystemResource(File.createTempFile("foo", "txt"));
+		class Person {
+
+			int id;
+
+			String name;
+
+		}
+
+		// when
+		FlatFileItemWriter<Person> writer = new FlatFileItemWriterBuilder<Person>().name("personWriter")
+				.resource(output).delimited().sourceType(Person.class).names("id", "name").build();
+
+		// then
+		Object lineAggregator = ReflectionTestUtils.getField(writer, "lineAggregator");
+		Assert.assertNotNull(lineAggregator);
+		Assert.assertTrue(lineAggregator instanceof DelimitedLineAggregator);
+		Object fieldExtractor = ReflectionTestUtils.getField(lineAggregator, "fieldExtractor");
+		Assert.assertNotNull(fieldExtractor);
+		Assert.assertTrue(fieldExtractor instanceof BeanWrapperFieldExtractor);
+	}
+
+	@Test
+	public void testSetupDelimitedLineAggregatorWithNoItemType() throws IOException {
+		// given
+		WritableResource output = new FileSystemResource(File.createTempFile("foo", "txt"));
+
+		// when
+		FlatFileItemWriter writer = new FlatFileItemWriterBuilder<>().name("personWriter").resource(output).delimited()
+				.names("id", "name").build();
+
+		// then
+		Object lineAggregator = ReflectionTestUtils.getField(writer, "lineAggregator");
+		Assert.assertNotNull(lineAggregator);
+		Assert.assertTrue(lineAggregator instanceof DelimitedLineAggregator);
+		Object fieldExtractor = ReflectionTestUtils.getField(lineAggregator, "fieldExtractor");
+		Assert.assertNotNull(fieldExtractor);
+		Assert.assertTrue(fieldExtractor instanceof BeanWrapperFieldExtractor);
+	}
+
+	@Test
+	public void testSetupFormatterLineAggregatorWithRecordItemType() throws IOException {
+		// given
+		WritableResource output = new FileSystemResource(File.createTempFile("foo", "txt"));
+		record Person(int id, String name) {
+		}
+
+		// when
+		FlatFileItemWriter<Person> writer = new FlatFileItemWriterBuilder<Person>().name("personWriter")
+				.resource(output).formatted().format("%2s%2s").sourceType(Person.class).names("id", "name").build();
+
+		// then
+		Object lineAggregator = ReflectionTestUtils.getField(writer, "lineAggregator");
+		Assert.assertNotNull(lineAggregator);
+		Assert.assertTrue(lineAggregator instanceof FormatterLineAggregator);
+		Object fieldExtractor = ReflectionTestUtils.getField(lineAggregator, "fieldExtractor");
+		Assert.assertNotNull(fieldExtractor);
+		Assert.assertTrue(fieldExtractor instanceof RecordFieldExtractor);
+	}
+
+	@Test
+	public void testSetupFormatterLineAggregatorWithClassItemType() throws IOException {
+		// given
+		WritableResource output = new FileSystemResource(File.createTempFile("foo", "txt"));
+		class Person {
+
+			int id;
+
+			String name;
+
+		}
+
+		// when
+		FlatFileItemWriter<Person> writer = new FlatFileItemWriterBuilder<Person>().name("personWriter")
+				.resource(output).formatted().format("%2s%2s").sourceType(Person.class).names("id", "name").build();
+
+		// then
+		Object lineAggregator = ReflectionTestUtils.getField(writer, "lineAggregator");
+		Assert.assertNotNull(lineAggregator);
+		Assert.assertTrue(lineAggregator instanceof FormatterLineAggregator);
+		Object fieldExtractor = ReflectionTestUtils.getField(lineAggregator, "fieldExtractor");
+		Assert.assertNotNull(fieldExtractor);
+		Assert.assertTrue(fieldExtractor instanceof BeanWrapperFieldExtractor);
+	}
+
+	@Test
+	public void testSetupFormatterLineAggregatorWithNoItemType() throws IOException {
+		// given
+		WritableResource output = new FileSystemResource(File.createTempFile("foo", "txt"));
+
+		// when
+		FlatFileItemWriter writer = new FlatFileItemWriterBuilder<>().name("personWriter").resource(output).formatted()
+				.format("%2s%2s").names("id", "name").build();
+
+		// then
+		Object lineAggregator = ReflectionTestUtils.getField(writer, "lineAggregator");
+		Assert.assertNotNull(lineAggregator);
+		Assert.assertTrue(lineAggregator instanceof FormatterLineAggregator);
+		Object fieldExtractor = ReflectionTestUtils.getField(lineAggregator, "fieldExtractor");
+		Assert.assertNotNull(fieldExtractor);
+		Assert.assertTrue(fieldExtractor instanceof BeanWrapperFieldExtractor);
+	}
+
 	private void validateBuilderFlags(FlatFileItemWriter<Foo> writer, String encoding) {
 		assertFalse((Boolean) ReflectionTestUtils.getField(writer, "saveState"));
 		assertTrue((Boolean) ReflectionTestUtils.getField(writer, "append"));