diff --git a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurer.java b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurer.java
index 55200c5a3..4ea9e6d11 100644
--- a/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurer.java
+++ b/spring-cloud-task-core/src/main/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurer.java
@@ -56,6 +56,8 @@ public class DefaultTaskConfigurer implements TaskConfigurer {
 
 	private static final Log logger = LogFactory.getLog(DefaultTaskConfigurer.class);
 
+	private TaskProperties taskProperties;
+
 	private TaskRepository taskRepository;
 
 	private TaskExplorer taskExplorer;
@@ -70,6 +72,14 @@ public DefaultTaskConfigurer() {
 		this(TaskProperties.DEFAULT_TABLE_PREFIX);
 	}
 
+	/**
+	 * Initializes the DefaultTaskConfigurer and retrieves table prefix from
+	 * {@link TaskProperties}.
+	 */
+	public DefaultTaskConfigurer(TaskProperties taskProperties) {
+		this(null, null, null, taskProperties);
+	}
+
 	/**
 	 * Initializes the DefaultTaskConfigurer and sets the default table prefix to
 	 * {@link TaskProperties#DEFAULT_TABLE_PREFIX}.
@@ -81,6 +91,19 @@ public DefaultTaskConfigurer(DataSource dataSource) {
 		this(dataSource, TaskProperties.DEFAULT_TABLE_PREFIX, null);
 	}
 
+	/**
+	 * Initializes the DefaultTaskConfigurer and retrieves table prefix from *
+	 * {@link TaskProperties}.
+	 * @param dataSource references the {@link DataSource} to be used as the Task
+	 * repository. If none is provided, a Map will be used (not recommended for production
+	 * use).
+	 * @param taskProperties the task properties used to obtain tablePrefix if not set by
+	 * tablePrefix field.
+	 */
+	public DefaultTaskConfigurer(DataSource dataSource, TaskProperties taskProperties) {
+		this(dataSource, null, null, taskProperties);
+	}
+
 	/**
 	 * Initializes the DefaultTaskConfigurer.
 	 * @param tablePrefix the prefix to apply to the task table names used by task
@@ -90,6 +113,17 @@ public DefaultTaskConfigurer(String tablePrefix) {
 		this(null, tablePrefix, null);
 	}
 
+	/**
+	 * Initializes the DefaultTaskConfigurer.
+	 * @param tablePrefix the prefix to apply to the task table names used by task
+	 * infrastructure.
+	 * @param taskProperties the task properties used to obtain tablePrefix if not set by
+	 * tablePrefix field.
+	 */
+	public DefaultTaskConfigurer(String tablePrefix, TaskProperties taskProperties) {
+		this(null, tablePrefix, null, taskProperties);
+	}
+
 	/**
 	 * Initializes the DefaultTaskConfigurer.
 	 * @param dataSource references the {@link DataSource} to be used as the Task
@@ -100,10 +134,32 @@ public DefaultTaskConfigurer(String tablePrefix) {
 	 * @param context the context to be used.
 	 */
 	public DefaultTaskConfigurer(DataSource dataSource, String tablePrefix, ApplicationContext context) {
+		this(dataSource, tablePrefix, context, null);
+	}
+
+	/**
+	 * Initializes the DefaultTaskConfigurer.
+	 * @param dataSource references the {@link DataSource} to be used as the Task
+	 * repository. If none is provided, a Map will be used (not recommended for production
+	 * use).
+	 * @param tablePrefix the prefix to apply to the task table names used by task
+	 * infrastructure.
+	 * @param context the context to be used.
+	 * @param taskProperties the task properties used to obtain tablePrefix if not set by
+	 * tablePrefix field.
+	 */
+	public DefaultTaskConfigurer(DataSource dataSource, String tablePrefix, ApplicationContext context,
+			TaskProperties taskProperties) {
 		this.dataSource = dataSource;
 		this.context = context;
 
 		TaskExecutionDaoFactoryBean taskExecutionDaoFactoryBean;
+		this.taskProperties = taskProperties;
+
+		if (tablePrefix == null) {
+			tablePrefix = (taskProperties != null && !taskProperties.getTablePrefix().isEmpty())
+					? taskProperties.getTablePrefix() : TaskProperties.DEFAULT_TABLE_PREFIX;
+		}
 
 		if (this.dataSource != null) {
 			taskExecutionDaoFactoryBean = new TaskExecutionDaoFactoryBean(this.dataSource, tablePrefix);
@@ -161,7 +217,6 @@ public PlatformTransactionManager getTransactionManager() {
 				this.transactionManager = new ResourcelessTransactionManager();
 			}
 		}
-
 		return this.transactionManager;
 	}
 
diff --git a/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurerTests.java b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurerTests.java
index 90ba65750..6a54a3ce9 100644
--- a/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurerTests.java
+++ b/spring-cloud-task-core/src/test/java/org/springframework/cloud/task/configuration/DefaultTaskConfigurerTests.java
@@ -24,12 +24,15 @@
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.jdbc.EmbeddedDataSourceConfiguration;
+import org.springframework.cloud.task.repository.support.SimpleTaskRepository;
+import org.springframework.cloud.task.repository.support.TaskExecutionDaoFactoryBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
@@ -59,7 +62,7 @@ public void resourcelessTransactionManagerTest() {
 	}
 
 	@Test
-	public void testDefaultContext() throws Exception {
+	public void testDefaultContext() {
 		AnnotationConfigApplicationContext localContext = new AnnotationConfigApplicationContext();
 		localContext.register(EmbeddedDataSourceConfiguration.class, EntityManagerConfiguration.class);
 		localContext.refresh();
@@ -114,6 +117,48 @@ public void taskDataSource() {
 		assertThat(defaultTaskConfigurer.getTaskDataSource()).isNull();
 	}
 
+	@Test
+	public void taskDataSourceWithProperties() {
+		TaskProperties taskProperties = new TaskProperties();
+		taskProperties.setTablePrefix("foo");
+		DefaultTaskConfigurer defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, taskProperties);
+		assertThat(defaultTaskConfigurer.getTaskDataSource()).isNotNull();
+		String prefix = getPrefix(defaultTaskConfigurer);
+		assertThat(prefix).isEqualTo("foo");
+		System.out.println(prefix);
+		defaultTaskConfigurer = new DefaultTaskConfigurer();
+		validatePrefix(defaultTaskConfigurer, "TASK_");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(taskProperties);
+		validatePrefix(defaultTaskConfigurer, "TASK_");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource);
+		validatePrefix(defaultTaskConfigurer, "TASK_");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, taskProperties);
+		validatePrefix(defaultTaskConfigurer, "foo");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, new TaskProperties());
+		validatePrefix(defaultTaskConfigurer, "TASK_");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, null, null);
+		validatePrefix(defaultTaskConfigurer, "TASK_");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, "bar", null);
+		validatePrefix(defaultTaskConfigurer, "bar");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, "bar", null, null);
+		validatePrefix(defaultTaskConfigurer, "bar");
+		defaultTaskConfigurer = new DefaultTaskConfigurer(this.dataSource, "bar", null, taskProperties);
+		validatePrefix(defaultTaskConfigurer, "bar");
+	}
+
+	private void validatePrefix(DefaultTaskConfigurer defaultTaskConfigurer, String prefix) {
+		String result = getPrefix(defaultTaskConfigurer);
+		assertThat(result).isEqualTo(prefix);
+	}
+
+	private String getPrefix(DefaultTaskConfigurer defaultTaskConfigurer) {
+		SimpleTaskRepository taskRepository = (SimpleTaskRepository) ReflectionTestUtils.getField(defaultTaskConfigurer,
+				"taskRepository");
+		TaskExecutionDaoFactoryBean factoryBean = (TaskExecutionDaoFactoryBean) ReflectionTestUtils
+			.getField(taskRepository, "taskExecutionDaoFactoryBean");
+		return (String) ReflectionTestUtils.getField(factoryBean, "tablePrefix");
+	}
+
 	@Configuration
 	public static class EntityManagerConfiguration {