diff --git a/src/main/java/io/debezium/connector/vitess/VitessValueConverter.java b/src/main/java/io/debezium/connector/vitess/VitessValueConverter.java
index 038c6489..ce8bbf2b 100644
--- a/src/main/java/io/debezium/connector/vitess/VitessValueConverter.java
+++ b/src/main/java/io/debezium/connector/vitess/VitessValueConverter.java
@@ -38,6 +38,7 @@
 public class VitessValueConverter extends JdbcValueConverters {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(VitessValueConverter.class);
+    private static final Logger INVALID_VALUE_LOGGER = LoggerFactory.getLogger(VitessValueConverter.class.getName() + ".invalid_value");
     private static final BigDecimal BIGINT_MAX_VALUE = new BigDecimal("18446744073709551615");
     private static final BigDecimal BIGINT_CORRECTION = BIGINT_MAX_VALUE.add(BigDecimal.ONE);
 
@@ -361,7 +362,7 @@ public static LocalDate stringToLocalDate(String dateString) {
         final int day = Integer.parseInt(matcher.group(3));
 
         if (year == 0 || month == 0 || day == 0) {
-            LOGGER.warn("Invalid value '{}' stored in column converted to empty value", dateString);
+            INVALID_VALUE_LOGGER.warn("Invalid value '{}' stored in column converted to empty value", dateString);
             return null;
         }
         return LocalDate.of(year, month, day);
diff --git a/src/test/java/io/debezium/connector/vitess/VitessValueConverterTest.java b/src/test/java/io/debezium/connector/vitess/VitessValueConverterTest.java
index a8cc03e6..308cdce2 100644
--- a/src/test/java/io/debezium/connector/vitess/VitessValueConverterTest.java
+++ b/src/test/java/io/debezium/connector/vitess/VitessValueConverterTest.java
@@ -23,6 +23,7 @@
 import org.junit.Test;
 
 import io.debezium.connector.vitess.connection.VStreamOutputMessageDecoder;
+import io.debezium.junit.logging.LogInterceptor;
 import io.debezium.relational.Column;
 import io.debezium.relational.Table;
 import io.debezium.relational.ValueConverter;
@@ -256,4 +257,11 @@ public void shouldConvertStringToTimestamp() {
         assertThat(VitessValueConverter.stringToTimestamp("2000-01-01 00:00:00.123456")).isEqualTo(
                 Timestamp.valueOf(LocalDate.of(2000, 1, 1).atStartOfDay().withNano(123456 * 1000)));
     }
+
+    @Test
+    public void shouldConvertInvalidValueToLocalData() {
+        final LogInterceptor logInterceptor = new LogInterceptor(VitessValueConverter.class.getName() + ".invalid_value");
+        assertThat(VitessValueConverter.stringToLocalDate("0000-00-00")).isNull();
+        assertThat(logInterceptor.containsMessage("Invalid value")).isTrue();
+    }
 }