diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
index c4f658c..8b464d9 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigValidator.java
@@ -24,6 +24,8 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Set;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.statefun.flink.core.exceptions.StatefulFunctionsInvalidConfigException;
@@ -40,6 +42,7 @@
 
   static void validate(Configuration configuration) {
     validateParentFirstClassloaderPatterns(configuration);
+    validateNoHeapBackedTimers(configuration);
   }
 
   private static void validateParentFirstClassloaderPatterns(Configuration configuration) {
@@ -61,4 +64,18 @@
     }
     return parentFirstClassloaderPatterns;
   }
+
+  private static final ConfigOption<String> TIMER_SERVICE_FACTORY =
+      ConfigOptions.key("state.backend.rocksdb.timer-service.factory")
+          .stringType()
+          .defaultValue("rocksdb");
+
+  private static void validateNoHeapBackedTimers(Configuration configuration) {
+    final String timerFactory = configuration.getString(TIMER_SERVICE_FACTORY);
+    if (!timerFactory.equalsIgnoreCase("rocksdb")) {
+      throw new StatefulFunctionsInvalidConfigException(
+          TIMER_SERVICE_FACTORY,
+          "StateFun only supports non-heap timers with a rocksdb state backend.");
+    }
+  }
 }