diff --git a/pkg/transformer/debezium/message.go b/pkg/transformer/debezium/message.go
index 220c2cdb..0a42c4ca 100644
--- a/pkg/transformer/debezium/message.go
+++ b/pkg/transformer/debezium/message.go
@@ -133,7 +133,7 @@ func convertDebeziumDate(days int) string {
 	ts := FromUnixMilli(int64(days * 86400 * 1000))
 	ts = ts.UTC()
 	return fmt.Sprintf(
-		"%d-%02d-%02d",
+		"%04d-%02d-%02d",
 		ts.Year(), ts.Month(), ts.Day(),
 	)
 }
@@ -149,7 +149,7 @@ func convertDebeziumMilliseconds(ms int, length int) string {
 	ts = ts.UTC()
 
 	result := fmt.Sprintf(
-		"%d-%02d-%02d %02d:%02d:%02d",
+		"%04d-%02d-%02d %02d:%02d:%02d",
 		ts.Year(), ts.Month(), ts.Day(),
 		ts.Hour(), ts.Minute(), ts.Second(),
 	)
@@ -177,7 +177,7 @@ func convertDebeziumMicroseconds(us int, length int) string {
 	ts = ts.UTC()
 
 	result := fmt.Sprintf(
-		"%d-%02d-%02d %02d:%02d:%02d",
+		"%04d-%02d-%02d %02d:%02d:%02d",
 		ts.Year(), ts.Month(), ts.Day(),
 		ts.Hour(), ts.Minute(), ts.Second(),
 	)
diff --git a/pkg/transformer/debezium/message_test.go b/pkg/transformer/debezium/message_test.go
index 0bfdec10..bd3af928 100644
--- a/pkg/transformer/debezium/message_test.go
+++ b/pkg/transformer/debezium/message_test.go
@@ -91,6 +91,72 @@ func TestConvertDebeziumFormattedTime(t *testing.T) {
 			sourceLength:  "3",
 			formattedTime: "2020-10-15 04:31:57.708",
 		},
+		{
+			name:          "test11: DATE epoch",
+			value:         "0",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "1970-01-01",
+		},
+		{
+			name:          "test12: DATE recent",
+			value:         "19181",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "2022-07-08",
+		},
+		{
+			name:          "test13: DATE max year",
+			value:         "2932896",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "9999-12-31",
+		},
+		// Years < 1000 must be zero-padded to 4 digits, else Redshift
+		// COPY rejects the value (Invalid Date Format - length must be
+		// 10 or more) and the loader retries the batch forever.
+		{
+			name:          "test14: DATE year < 1000 is zero-padded",
+			value:         "-645643",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "0202-04-17",
+		},
+		{
+			name:          "test15: DATE 3 digit year is zero-padded",
+			value:         "-361910",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "0979-02-15",
+		},
+		{
+			name:          "test16: DATE last 3 digit year",
+			value:         "-354286",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "0999-12-31",
+		},
+		{
+			name:          "test17: DATE first 4 digit year",
+			value:         "-354285",
+			sourceType:    "DATE",
+			sourceLength:  "",
+			formattedTime: "1000-01-01",
+		},
+		{
+			name:          "test18: DATETIME year < 1000 is zero-padded",
+			value:         "-55704758400000",
+			sourceType:    "DATETIME",
+			sourceLength:  "",
+			formattedTime: "0204-10-15 00:00:00",
+		},
+		{
+			name:          "test19: DATETIME(6) year < 1000 is zero-padded",
+			value:         "-55805894999243000",
+			sourceType:    "DATETIME",
+			sourceLength:  "6",
+			formattedTime: "0201-08-01 10:30:00.757000",
+		},
 	}
 
 	for _, tc := range tests {