kafka: refactor around datetime and timestamp

timestamp is always in UTC

Author: ffffwh

driver/kafka/kafka2.go

@@ -50,7 +50,9 @@ const (
 )
 
 const (
-	LAYOUT = "2006-01-02 15:04:05"
+	MySQLDateTimeFormat = "2006-01-02 15:04:05"
+	MySQLDateFormat     = "2006-01-02"
+	dbzTimestampFormat  = "2006-01-02T15:04:05Z"
 )
 
 var (
@@ -465,7 +467,7 @@ func NewDateTimeField(optional bool, field string, defaultValue interface{}, loc
 	}
 }
 func DateTimeValue(dateTime string, loc *time.Location) int64 {
-	tm2, err := time.ParseInLocation(LAYOUT, dateTime, loc)
+	tm2, err := time.ParseInLocation(MySQLDateTimeFormat, dateTime, loc)
 	if err != nil {
 		return 0
 	}
@@ -477,17 +479,16 @@ func DateTimeValue(dateTime string, loc *time.Location) int64 {
 	return tm2.Unix()*1000 + ms
 }
 func DateValue(date string) int64 {
-	tm2, err := time.Parse(LAYOUT, date+" 00:00:00")
+	tm2, err := time.Parse(MySQLDateFormat, date)
 	if err != nil {
 		return 0
 	}
 	return tm2.Unix() / 60 / 60 / 24
 }
-func TimeStamp(timestamp string, loc *time.Location) string {
-	tm2, _ := time.ParseInLocation(LAYOUT, timestamp, loc)
-	value := tm2.In(time.UTC).Format(LAYOUT)
-	timestamp = value[:10] + "T" + value[11:] + "Z"
-	return timestamp
+func TimeStamp(timestamp string) string {
+	// dtle always read timestamp in UTC
+	tm2, _ := time.ParseInLocation(MySQLDateTimeFormat, timestamp, time.UTC)
+	return tm2.In(time.UTC).Format(dbzTimestampFormat)
 }
 
 func NewJsonField(optional bool, field string) *Schema {
@@ -556,13 +557,12 @@ func NewSetField(theType SchemaType, optional bool, field string, allowed string
 		Version: 1,
 	}
 }
-func NewTimeStampField(optional bool, field string, defaultValue interface{}, loc *time.Location) *Schema {
+func NewTimeStampField(optional bool, field string, defaultValue interface{}) *Schema {
 	if defaultValue == "CURRENT_TIMESTAMP" {
 		defaultValue = "1970-01-01T00:00:00Z"
 	} else if defaultValue != nil {
-		tm2, _ := time.ParseInLocation(LAYOUT, defaultValue.(string), loc)
-		value := tm2.In(time.UTC).Format(LAYOUT)
-		defaultValue = value[:10] + "T" + value[11:] + "Z"
+		tm2, _ := time.ParseInLocation(MySQLDateTimeFormat, defaultValue.(string), time.UTC)
+		defaultValue = tm2.In(time.UTC).Format(dbzTimestampFormat)
 	}
 	return &Schema{
 		Field:    field,

driver/kafka/kafka3.go

@@ -762,8 +762,9 @@ func (kr *KafkaRunner) kafkaTransformSnapshotData(
 					value = TimeValue(valueStr)
 				case mysqlconfig.TimestampColumnType:
 					if valueStr != "" {
-						value = TimeStamp(valueStr, kr.location)
+						value = TimeStamp(valueStr)
 					} else {
+						// TODO what?
 						value = TimeValue(valueStr)
 					}
 				case mysqlconfig.BinaryColumnType:
@@ -790,13 +791,14 @@ func (kr *KafkaRunner) kafkaTransformSnapshotData(
 						valueStr = "char(255)"
 					}
 					value = base64.StdEncoding.EncodeToString([]byte(valueStr))
-				case mysqlconfig.DateColumnType, mysqlconfig.DateTimeColumnType:
-					if valueStr != "" && columnList[i].ColumnType == "datetime" {
+				case mysqlconfig.DateTimeColumnType:
+					if valueStr != "" {
 						value = DateTimeValue(valueStr, kr.location)
-					} else if valueStr != "" {
+					}
+				case mysqlconfig.DateColumnType:
+					 if valueStr != "" {
 						value = DateValue(valueStr)
 					}
-
 				case mysqlconfig.YearColumnType:
 					if valueStr != "" {
 						value = YearValue(valueStr)
@@ -1072,18 +1074,22 @@ func (kr *KafkaRunner) kafkaConvertArg(column *mysqlconfig.Column, theValue inte
 				theValue = int64(theValue.(uint64))
 			}
 		}
-	case mysqlconfig.TimeColumnType, mysqlconfig.TimestampColumnType:
-		if theValue != nil && column.ColumnType == "timestamp" {
-			theValue = TimeStamp(theValue.(string), kr.location)
-		} else if theValue != nil {
+	case mysqlconfig.TimeColumnType:
+		if theValue != nil {
 			theValue = TimeValue(theValue.(string))
 		}
-	case mysqlconfig.DateColumnType, mysqlconfig.DateTimeColumnType:
-		if theValue != nil && column.ColumnType == "datetime" {
-			theValue = DateTimeValue(theValue.(string), kr.location)
-		} else if theValue != nil {
+	case mysqlconfig.TimestampColumnType:
+		if theValue != nil {
+			theValue = TimeStamp(theValue.(string))
+		}
+	case mysqlconfig.DateColumnType:
+		if theValue != nil {
 			theValue = DateValue(theValue.(string))
 		}
+	case mysqlconfig.DateTimeColumnType:
+		if theValue != nil {
+			theValue = DateTimeValue(theValue.(string), kr.location)
+		}
 	case mysqlconfig.VarbinaryColumnType:
 		if theValue != nil {
 			theValue = encodeStringInterfaceToBase64String(theValue)
@@ -1294,19 +1300,15 @@ func kafkaColumnListToColDefs(colList *common.ColumnList, loc *time.Location) (v
 		case mysqlconfig.DecimalColumnType:
 			field = NewDecimalField(cols[i].Precision, cols[i].Scale, optional, fieldName, defaultValue)
 		case mysqlconfig.DateColumnType:
-			if cols[i].ColumnType == "datetime" {
-				field = NewDateTimeField(optional, fieldName, defaultValue, loc)
-			} else {
-				field = NewDateField(SCHEMA_TYPE_INT32, optional, fieldName, defaultValue)
-			}
+			field = NewDateField(SCHEMA_TYPE_INT32, optional, fieldName, defaultValue)
 		case mysqlconfig.YearColumnType:
 			field = NewYearField(SCHEMA_TYPE_INT32, optional, fieldName, defaultValue)
 		case mysqlconfig.DateTimeColumnType:
 			field = NewDateTimeField(optional, fieldName, defaultValue, loc)
 		case mysqlconfig.TimeColumnType:
 			field = NewTimeField(optional, fieldName, defaultValue)
 		case mysqlconfig.TimestampColumnType:
-			field = NewTimeStampField(optional, fieldName, defaultValue, loc)
+			field = NewTimeStampField(optional, fieldName, defaultValue)
 		case mysqlconfig.JSONColumnType:
 			field = NewJsonField(optional, fieldName)
 		default: