diff --git a/data/avro-schema/src/main/resources/avro/net/corda/data/flow/event/mapper/ExecuteCleanup.avsc b/data/avro-schema/src/main/resources/avro/net/corda/data/flow/event/mapper/ExecuteCleanup.avsc
index c04b43197d..e5e1595715 100644
--- a/data/avro-schema/src/main/resources/avro/net/corda/data/flow/event/mapper/ExecuteCleanup.avsc
+++ b/data/avro-schema/src/main/resources/avro/net/corda/data/flow/event/mapper/ExecuteCleanup.avsc
@@ -4,5 +4,13 @@
   "docs": "When this event is processed the flow mapper state should be set to null",
   "namespace": "net.corda.data.flow.event.mapper",
   "fields": [
+    {
+      "name": "ids",
+      "docs": "A list of IDs of mapper states that should be deleted",
+      "type": {
+        "type": "array",
+        "items": "string"
+      }
+    }
   ]
 }
diff --git a/data/topic-schema/src/main/java/net/corda/schema/Schemas.java b/data/topic-schema/src/main/java/net/corda/schema/Schemas.java
index 342ae10cdc..33d6590895 100644
--- a/data/topic-schema/src/main/java/net/corda/schema/Schemas.java
+++ b/data/topic-schema/src/main/java/net/corda/schema/Schemas.java
@@ -103,6 +103,7 @@ private Flow() {
         public static final String FLOW_MAPPER_EVENT_TOPIC = "flow.mapper.event";
         public static final String FLOW_MAPPER_EVENT_STATE_TOPIC = getStateAndEventStateTopic(FLOW_MAPPER_EVENT_TOPIC);
         public static final String FLOW_MAPPER_EVENT_DLQ_TOPIC = getDLQTopic(FLOW_MAPPER_EVENT_TOPIC);
+        public static final String FLOW_MAPPER_CLEANUP_TOPIC = "flow.mapper.cleanup";
     }
 
     /**
@@ -267,5 +268,6 @@ public static final class ScheduledTask {
         private ScheduledTask() {}
 
         public static final String SCHEDULED_TASK_DB_PROCESSOR = "scheduled.task.db.processor";
+        public static final String SCHEDULED_TASK_MAPPER_PROCESSOR = "scheduled.task.mapper.processor";
     }
 }
diff --git a/data/topic-schema/src/main/resources/net/corda/schema/Flow.yaml b/data/topic-schema/src/main/resources/net/corda/schema/Flow.yaml
index 04a0e5236c..0b29c6fce0 100644
--- a/data/topic-schema/src/main/resources/net/corda/schema/Flow.yaml
+++ b/data/topic-schema/src/main/resources/net/corda/schema/Flow.yaml
@@ -80,3 +80,10 @@ topics:
       min.compaction.lag.ms: 60000
       max.compaction.lag.ms: 604800000
       min.cleanable.dirty.ratio: 0.5
+  FlowMapperCleanupTopic:
+    name: flow.mapper.cleanup
+    consumers:
+      - flowMapper
+    producers:
+      - flowMapper
+    config:
diff --git a/data/topic-schema/src/main/resources/net/corda/schema/ScheduledTask.yaml b/data/topic-schema/src/main/resources/net/corda/schema/ScheduledTask.yaml
index e0a19daa79..fa45e6401c 100644
--- a/data/topic-schema/src/main/resources/net/corda/schema/ScheduledTask.yaml
+++ b/data/topic-schema/src/main/resources/net/corda/schema/ScheduledTask.yaml
@@ -5,3 +5,9 @@ topics:
       - db
     producers:
       - db
+  ScheduledTaskFlowMapperProcessorTopic:
+    name: scheduled.task.mapper.processor
+    consumers:
+      - flowMapper
+    producers:
+      - db
diff --git a/gradle.properties b/gradle.properties
index 10fdad95d7..d2908e1b60 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -9,7 +9,7 @@ cordaProductVersion = 5.1.0
 # NOTE: update this each time this module contains a breaking change
 ## NOTE: currently this is a top level revision, so all API versions will line up, but this could be moved to
 ##   a per module property in which case module versions can change independently.
-cordaApiRevision = 24
+cordaApiRevision = 25
 
 # Main
 kotlinVersion = 1.8.21