diff --git a/debezium-core/src/main/java/io/debezium/relational/history/DatabaseHistory.java b/debezium-core/src/main/java/io/debezium/relational/history/DatabaseHistory.java
index 5d1311289..b9d49a3f3 100644
--- a/debezium-core/src/main/java/io/debezium/relational/history/DatabaseHistory.java
+++ b/debezium-core/src/main/java/io/debezium/relational/history/DatabaseHistory.java
@@ -61,7 +61,9 @@ public interface DatabaseHistory {
                                                 .withDefault(
                                                     "DROP TEMPORARY TABLE IF EXISTS .+ /\\* generated by server \\*/," +
                                                     // Filter out RDS heartbeat statements, see DBZ-469
-                                                    "INSERT INTO mysql.rds_heartbeat2\\(.*\\) values \\(.*\\) ON DUPLICATE KEY UPDATE value = .*"
+                                                    "INSERT INTO mysql.rds_heartbeat2\\(.*\\) values \\(.*\\) ON DUPLICATE KEY UPDATE value = .*," +
+                                                    "FLUSH RELAY LOGS.*," +
+                                                    "flush relay logs.*"
                                                  )
                                                 .withWidth(Width.LONG)
                                                 .withImportance(Importance.LOW)
diff --git a/debezium-core/src/test/java/io/debezium/config/ConfigurationTest.java b/debezium-core/src/test/java/io/debezium/config/ConfigurationTest.java
index 893c3bbf9..b084b6428 100644
--- a/debezium-core/src/test/java/io/debezium/config/ConfigurationTest.java
+++ b/debezium-core/src/test/java/io/debezium/config/ConfigurationTest.java
@@ -135,4 +135,12 @@ public void defaultDddlFilterShouldFilterOutRdsHeartbeatInsert() {
         Predicate<String> ddlFilter = Predicates.includes(defaultDdlFilter);
         assertThat(ddlFilter.test("INSERT INTO mysql.rds_heartbeat2(id, value) values (1,1510678117058) ON DUPLICATE KEY UPDATE value = 1510678117058")).isTrue();
     }
+
+    @Test
+    @FixFor("DBZ-661")
+    public void defaultDddlFilterShouldFilterOutFlushRelayLogs() {
+        String defaultDdlFilter = Configuration.create().build().getString(DatabaseHistory.DDL_FILTER);
+        Predicate<String> ddlFilter = Predicates.includes(defaultDdlFilter);
+        assertThat(ddlFilter.test("FLUSH RELAY LOGS")).isTrue();
+    }
 }