diff --git a/cdc/log.cc b/cdc/log.cc
index b1ddcd18ae..5fa927ab8a 100644
--- a/cdc/log.cc
+++ b/cdc/log.cc
@@ -160,7 +160,7 @@ public:
         });
     }
 
-    void on_before_create_column_family(const schema& schema, std::vector<mutation>& mutations, api::timestamp_type timestamp) override {
+    void on_before_create_column_family(const keyspace_metadata& ksm, const schema& schema, std::vector<mutation>& mutations, api::timestamp_type timestamp) override {
         if (schema.cdc_options().enabled()) {
             auto& db = _ctxt._proxy.get_db().local();
             auto logname = log_name(schema.cf_name());
diff --git a/service/migration_listener.hh b/service/migration_listener.hh
index 6df9be2e6f..e731dbf1f7 100644
--- a/service/migration_listener.hh
+++ b/service/migration_listener.hh
@@ -73,7 +73,12 @@ public:
     // The callback runs inside seastar thread
     // called before adding/updating/dropping column family. 
     // listener can add additional type altering mutations if he knows what he is doing. 
-    virtual void on_before_create_column_family(const schema&, std::vector<mutation>&, api::timestamp_type) {}
+    //
+    // The `on_before_create_column_family` method is different as it doesn't assume the existence
+    // of the column family's keyspace. The reason for this is that we sometimes create a keyspace
+    // and its column families together. Therefore, listeners can't load the keyspace from the
+    // database. Instead, they should use the `ksm` parameter if needed.
+    virtual void on_before_create_column_family(const keyspace_metadata& ksm, const schema&, std::vector<mutation>&, api::timestamp_type) {}
     virtual void on_before_update_column_family(const schema& new_schema, const schema& old_schema, std::vector<mutation>&, api::timestamp_type) {}
     virtual void on_before_drop_column_family(const schema&, std::vector<mutation>&, api::timestamp_type) {}
     virtual void on_before_drop_keyspace(const sstring& keyspace_name, std::vector<mutation>&, api::timestamp_type) {}
@@ -139,7 +144,7 @@ public:
     future<> drop_function(const db::functions::function_name& fun_name, const std::vector<data_type>& arg_types);
     future<> drop_aggregate(const db::functions::function_name& fun_name, const std::vector<data_type>& arg_types);
 
-    void before_create_column_family(const schema&, std::vector<mutation>&, api::timestamp_type);
+    void before_create_column_family(const keyspace_metadata& ksm, const schema&, std::vector<mutation>&, api::timestamp_type);
     void before_update_column_family(const schema& new_schema, const schema& old_schema, std::vector<mutation>&, api::timestamp_type);
     void before_drop_column_family(const schema&, std::vector<mutation>&, api::timestamp_type);
     void before_drop_keyspace(const sstring& keyspace_name, std::vector<mutation>&, api::timestamp_type);
diff --git a/service/migration_manager.cc b/service/migration_manager.cc
index b0eb85011d..4a3a55bd51 100644
--- a/service/migration_manager.cc
+++ b/service/migration_manager.cc
@@ -614,11 +614,11 @@ future<> migration_notifier::drop_aggregate(const db::functions::function_name&
     });
 }
 
-void migration_notifier::before_create_column_family(const schema& schema,
-        std::vector<mutation>& mutations, api::timestamp_type timestamp) {
-    _listeners.thread_for_each([&mutations, &schema, timestamp] (migration_listener* listener) {
+void migration_notifier::before_create_column_family(const keyspace_metadata& ksm,
+        const schema& schema, std::vector<mutation>& mutations, api::timestamp_type timestamp) {
+    _listeners.thread_for_each([&ksm, &schema, &mutations, timestamp] (migration_listener* listener) {
         // allow exceptions. so a listener can effectively kill a create-table
-        listener->on_before_create_column_family(schema, mutations, timestamp);
+        listener->on_before_create_column_family(ksm, schema, mutations, timestamp);
     });
 }
 
@@ -691,9 +691,9 @@ static future<std::vector<mutation>> do_prepare_new_column_family_announcement(s
 
     mlogger.info("Create new ColumnFamily: {}", cfm);
 
-    return seastar::async([&db, cfm, timestamp] {
+    return seastar::async([&db, &ksm, cfm, timestamp] {
         auto mutations = db::schema_tables::make_create_table_mutations(cfm, timestamp);
-        db.get_notifier().before_create_column_family(*cfm, mutations, timestamp);
+        db.get_notifier().before_create_column_family(ksm, *cfm, mutations, timestamp);
         return mutations;
     }).then([&sp, &ksm](std::vector<mutation> mutations) {
         return include_keyspace(sp, ksm, std::move(mutations));
diff --git a/service/tablet_allocator.cc b/service/tablet_allocator.cc
index 6bee031a77..f41c5fe952 100644
--- a/service/tablet_allocator.cc
+++ b/service/tablet_allocator.cc
@@ -824,7 +824,7 @@ public:
         co_return co_await lb.make_plan();
     }
 
-    void on_before_create_column_family(const schema& s, std::vector<mutation>& muts, api::timestamp_type ts) override {
+    void on_before_create_column_family(const keyspace_metadata& ksm, const schema& s, std::vector<mutation>& muts, api::timestamp_type ts) override {
         keyspace& ks = _db.find_keyspace(s.ks_name());
         auto&& rs = ks.get_replication_strategy();
         if (auto&& tablet_rs = rs.maybe_as_tablet_aware()) {