diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverter.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverter.java
index dfefccd358..df1a10b27b 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverter.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverter.java
@@ -32,6 +32,7 @@
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.jdbc.core.mapping.AggregateReference;
+import org.springframework.data.jdbc.repository.config.DialectResolver;
 import org.springframework.data.jdbc.support.JdbcUtil;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.PersistentPropertyPath;
@@ -51,6 +52,7 @@
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -64,6 +66,8 @@
  * @author Jens Schauder
  * @author Christoph Strobl
  * @author Myeonghyeon Lee
+ * @author Mikhail Polivakha
+ *
  * @see MappingContext
  * @see SimpleTypeHolder
  * @see CustomConversions
@@ -170,7 +174,14 @@ private Class<?> getReferenceColumnType(RelationalPersistentProperty property) {
 	 */
 	@Override
 	public int getSqlType(RelationalPersistentProperty property) {
-		return JdbcUtil.sqlTypeFor(getColumnType(property));
+		if (typeFactory instanceof DefaultJdbcTypeFactory) {
+			return JdbcUtil.sqlTypeFor(
+					getColumnType(property),
+					DialectResolver.getDialect(((DefaultJdbcTypeFactory) typeFactory).getOperations())
+			);
+		} else {
+			return JdbcUtil.sqlTypeFor(getColumnType(property));
+		}
 	}
 
 	/*
@@ -196,7 +207,7 @@ private Class<?> doGetColumnType(RelationalPersistentProperty property) {
 			}
 		}
 
-		Class<?> componentColumnType = JdbcColumnTypes.INSTANCE.resolvePrimitiveType(property.getActualType());
+		Class<?> componentColumnType = resolvePropertyPrimitiveType(property);
 
 		while (componentColumnType.isArray()) {
 			componentColumnType = componentColumnType.getComponentType();
@@ -209,6 +220,14 @@ private Class<?> doGetColumnType(RelationalPersistentProperty property) {
 		return componentColumnType;
 	}
 
+	private Class<?> resolvePropertyPrimitiveType(RelationalPersistentProperty property) {
+		if (typeFactory instanceof DefaultJdbcTypeFactory) {
+			return JdbcColumnTypes.INSTANCE.resolvePrimitiveType(property.getActualType(), DialectResolver.getDialect(((DefaultJdbcTypeFactory) typeFactory).getOperations()));
+		} else {
+			return JdbcColumnTypes.INSTANCE.resolvePrimitiveType(property.getActualType());
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.conversion.RelationalConverter#readValue(java.lang.Object, org.springframework.data.util.TypeInformation)
@@ -240,21 +259,6 @@ public Object readValue(@Nullable Object value, TypeInformation<?> type) {
 		return super.readValue(value, type);
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.data.relational.core.conversion.RelationalConverter#writeValue(java.lang.Object, org.springframework.data.util.TypeInformation)
-	 */
-	@Override
-	@Nullable
-	public Object writeValue(@Nullable Object value, TypeInformation<?> type) {
-
-		if (value == null) {
-			return null;
-		}
-
-		return super.writeValue(value, type);
-	}
-
 	private boolean canWriteAsJdbcValue(@Nullable Object value) {
 
 		if (value == null) {
@@ -285,6 +289,7 @@ private boolean canWriteAsJdbcValue(@Nullable Object value) {
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.convert.JdbcConverter#writeValue(java.lang.Object, java.lang.Class, int)
 	 */
+	@NonNull
 	@Override
 	public JdbcValue writeJdbcValue(@Nullable Object value, Class<?> columnType, int sqlType) {
 
@@ -329,8 +334,12 @@ private JdbcValue tryToConvertToJdbcValue(@Nullable Object value) {
 
 	@Override
 	public <T> T mapRow(RelationalPersistentEntity<T> entity, ResultSet resultSet, Object key) {
-		return new ReadingContext<T>(new PersistentPropertyPathExtension(getMappingContext(), entity),
-				new ResultSetAccessor(resultSet), Identifier.empty(), key).mapRow();
+		return new ReadingContext<T>(
+				new PersistentPropertyPathExtension(getMappingContext(), entity),
+				new ResultSetAccessor(resultSet),
+				Identifier.empty(),
+				key
+		).mapRow();
 	}
 
 	@Override
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java
index 33b82a675f..95b4bc375a 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java
@@ -32,6 +32,7 @@
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
+import org.springframework.data.jdbc.repository.config.DialectResolver;
 import org.springframework.data.jdbc.support.JdbcUtil;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
@@ -67,6 +68,8 @@
  * @author Myeonghyeon Lee
  * @author Yunyoung LEE
  * @author Radim Tlusty
+ * @author Mikhail Polivakha
+ *
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -545,13 +548,26 @@ private IdentifierProcessing getIdentifierProcessing() {
 	private void addConvertedPropertyValue(SqlIdentifierParameterSource parameterSource,
 			RelationalPersistentProperty property, @Nullable Object value, SqlIdentifier name) {
 
-		addConvertedValue(parameterSource, value, name, converter.getColumnType(property), converter.getSqlType(property));
+		final Class<?> javaColumnType = converter.getColumnType(property);
+		addConvertedValue(
+				parameterSource,
+				value,
+				name,
+				javaColumnType,
+				JdbcUtil.sqlTypeFor(javaColumnType, DialectResolver.getDialect(operations.getJdbcOperations()))
+		);
 	}
 
 	private void addConvertedPropertyValue(SqlIdentifierParameterSource parameterSource, SqlIdentifier name, Object value,
 			Class<?> javaType) {
 
-		addConvertedValue(parameterSource, value, name, javaType, JdbcUtil.sqlTypeFor(javaType));
+		addConvertedValue(
+				parameterSource,
+				value,
+				name,
+				javaType,
+				JdbcUtil.sqlTypeFor(javaType, DialectResolver.getDialect(operations.getJdbcOperations()))
+		);
 	}
 
 	private void addConvertedValue(SqlIdentifierParameterSource parameterSource, @Nullable Object value,
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultJdbcTypeFactory.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultJdbcTypeFactory.java
index c70287d1ba..95ac39fade 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultJdbcTypeFactory.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultJdbcTypeFactory.java
@@ -19,9 +19,11 @@
 import java.sql.JDBCType;
 
 import org.springframework.data.jdbc.core.dialect.JdbcArrayColumns;
+import org.springframework.data.jdbc.repository.config.DialectResolver;
 import org.springframework.data.jdbc.support.JdbcUtil;
 import org.springframework.jdbc.core.ConnectionCallback;
 import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 
 /**
@@ -30,6 +32,8 @@
  *
  * @author Jens Schauder
  * @author Mark Paluch
+ * @author Mikhail Polivakha
+ *
  * @since 1.1
  */
 public class DefaultJdbcTypeFactory implements JdbcTypeFactory {
@@ -68,11 +72,15 @@ public Array createArray(Object[] value) {
 
 		Class<?> componentType = arrayColumns.getArrayType(value.getClass());
 
-		JDBCType jdbcType = JdbcUtil.jdbcTypeFor(componentType);
+		JDBCType jdbcType = JdbcUtil.jdbcTypeFor(componentType, DialectResolver.getDialect(operations));
 		Assert.notNull(jdbcType, () -> String.format("Couldn't determine JDBCType for %s", componentType));
 		String typeName = arrayColumns.getArrayTypeName(jdbcType);
 
 		return operations.execute((ConnectionCallback<Array>) c -> c.createArrayOf(typeName, value));
 	}
 
+	@NonNull
+	public JdbcOperations getOperations() {
+		return operations;
+	}
 }
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/EntityRowMapper.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/EntityRowMapper.java
index 1a04450957..e7986fae49 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/EntityRowMapper.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/EntityRowMapper.java
@@ -20,6 +20,7 @@
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.jdbc.core.RowMapper;
+import org.springframework.util.Assert;
 
 /**
  * Maps a {@link ResultSet} to an entity of type {@code T}, including entities referenced. This {@link RowMapper} might
@@ -50,6 +51,9 @@ public EntityRowMapper(PersistentPropertyPathExtension path, JdbcConverter conve
 
 	public EntityRowMapper(RelationalPersistentEntity<T> entity, JdbcConverter converter) {
 
+		Assert.notNull(entity, "relationalPersistentEntity must not be null!");
+		Assert.notNull(converter, "jdbcConverter must not be null!");
+
 		this.entity = entity;
 		this.path = null;
 		this.converter = converter;
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcColumnTypes.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcColumnTypes.java
index a5f2cc9136..c849f13cde 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcColumnTypes.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcColumnTypes.java
@@ -23,7 +23,9 @@
 import java.util.Date;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Optional;
 
+import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.util.ClassUtils;
 
 /**
@@ -31,12 +33,19 @@
  * compatible with JDBC drivers.
  *
  * @author Jens Schauder
+ * @author Mikhail Polivakha
  * @since 2.0
  */
 public enum JdbcColumnTypes {
 
 	INSTANCE {
 
+		/**
+		 * @deprecated because this method will resolve the target column type regardless of the
+		 * 			   current client {@link Dialect} - it will search only in common types mappings.
+		 * 			   When possible, please, use {@link #resolvePrimitiveType(Class, Dialect)}
+		 */
+		@Deprecated
 		@SuppressWarnings({ "unchecked", "rawtypes" })
 		public Class<?> resolvePrimitiveType(Class<?> type) {
 
@@ -46,6 +55,13 @@ public Class<?> resolvePrimitiveType(Class<?> type) {
 					.findFirst() //
 					.orElseGet(() -> (Class) ClassUtils.resolvePrimitiveIfNecessary(type));
 		}
+
+		@SuppressWarnings({ "unchecked", "rawtypes", "deprecation" })
+		public Class<?> resolvePrimitiveType(Class<?> type, Dialect dialect) {
+			return Optional
+					.ofNullable(dialect.getCustomJdbcColumnsMappings().get(type))
+					.orElse((Class) this.resolvePrimitiveType(type));
+		}
 	};
 
 	private static final Map<Class<?>, Class<?>> javaToDbType = new LinkedHashMap<>();
@@ -53,11 +69,19 @@ public Class<?> resolvePrimitiveType(Class<?> type) {
 	static {
 
 		javaToDbType.put(Enum.class, String.class);
-		javaToDbType.put(ZonedDateTime.class, String.class);
+		javaToDbType.put(ZonedDateTime.class, ZonedDateTime.class);
 		javaToDbType.put(OffsetDateTime.class, OffsetDateTime.class);
 		javaToDbType.put(LocalDateTime.class, LocalDateTime.class);
 		javaToDbType.put(Temporal.class, Timestamp.class);
 	}
 
+	/**
+	 * @deprecated because this method will resolve the target column type regardless of the
+	 * 			   current client {@link Dialect} - it will search only in common types mappings.
+	 * 			   When possible, please, use {@link #resolvePrimitiveType(Class, Dialect)}
+	 */
+	@Deprecated
 	public abstract Class<?> resolvePrimitiveType(Class<?> type);
+
+	public abstract Class<?> resolvePrimitiveType(Class<?> type, Dialect dialect);
 }
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcConverter.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcConverter.java
index b0a835ca65..34d5228369 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcConverter.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/JdbcConverter.java
@@ -18,10 +18,12 @@
 import java.sql.ResultSet;
 
 import org.springframework.data.relational.core.conversion.RelationalConverter;
+import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.util.TypeInformation;
+import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 
 /**
@@ -42,6 +44,7 @@ public interface JdbcConverter extends RelationalConverter {
 	 * @param sqlType the type constant from {@link java.sql.Types} to be used if non is specified by a converter.
 	 * @return The converted value wrapped in a {@link JdbcValue}. Guaranteed to be not {@literal null}.
 	 */
+	@NonNull
 	JdbcValue writeJdbcValue(@Nullable Object value, Class<?> type, int sqlType);
 
 	/**
@@ -72,7 +75,7 @@ public interface JdbcConverter extends RelationalConverter {
 	 * top-level array type (e.g. {@code String[][]} returns {@code String[]}).
 	 *
 	 * @return a {@link Class} that is suitable for usage with JDBC drivers.
-	 * @see org.springframework.data.jdbc.support.JdbcUtil#sqlTypeFor(Class)
+	 * @see org.springframework.data.jdbc.support.JdbcUtil#sqlTypeFor(Class, Dialect)
 	 * @since 2.0
 	 */
 	Class<?> getColumnType(RelationalPersistentProperty property);
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/Jsr310TimestampBasedConverters.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/Jsr310TimestampBasedConverters.java
index a331c8183d..0ea81bb526 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/Jsr310TimestampBasedConverters.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/Jsr310TimestampBasedConverters.java
@@ -24,8 +24,11 @@
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
+import java.time.OffsetDateTime;
 import java.time.Period;
 import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -43,6 +46,7 @@
  *
  * @see org.springframework.data.convert.Jsr310Converters
  * @author Jens Schauder
+ * @author Mikhail Polivakha
  * @since 2.2
  */
 public abstract class Jsr310TimestampBasedConverters {
@@ -66,6 +70,8 @@ public abstract class Jsr310TimestampBasedConverters {
 		converters.add(LocalTimeToTimestampConverter.INSTANCE);
 		converters.add(TimestampToInstantConverter.INSTANCE);
 		converters.add(InstantToTimestampConverter.INSTANCE);
+		converters.add(TimestampToZonedDateTimeConverter.INSTANCE);
+		converters.add(TimestampToOffsetDateTimeConverter.INSTANCE);
 
 		return converters;
 	}
@@ -165,4 +171,26 @@ public Timestamp convert(Instant source) {
 			return Timestamp.from(source);
 		}
 	}
+
+	@ReadingConverter
+	public enum TimestampToZonedDateTimeConverter implements Converter<Timestamp, ZonedDateTime> {
+
+		INSTANCE;
+
+		@Override
+		public ZonedDateTime convert(Timestamp source) {
+			return ZonedDateTime.ofInstant(source.toInstant(), ZoneId.of("UTC"));
+		}
+	}
+
+	@ReadingConverter
+	public enum TimestampToOffsetDateTimeConverter implements Converter<Timestamp, OffsetDateTime> {
+
+		INSTANCE;
+
+		@Override
+		public OffsetDateTime convert(Timestamp source) {
+			return OffsetDateTime.ofInstant(source.toInstant(), ZoneId.of("UTC"));
+		}
+	}
 }
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToOffsetDateTimeConverter.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToOffsetDateTimeConverter.java
index 7687676462..586c4db18d 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToOffsetDateTimeConverter.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToOffsetDateTimeConverter.java
@@ -31,8 +31,7 @@
  * @since 2.7
  */
 @ReadingConverter
-public enum H2TimestampWithTimeZoneToOffsetDateTimeConverter
-		implements Converter<TimestampWithTimeZone, OffsetDateTime> {
+public enum H2TimestampWithTimeZoneToOffsetDateTimeConverter implements Converter<TimestampWithTimeZone, OffsetDateTime> {
 
 	INSTANCE;
 
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToZonedDateTimeConverter.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToZonedDateTimeConverter.java
new file mode 100644
index 0000000000..1f6f5c1eef
--- /dev/null
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/H2TimestampWithTimeZoneToZonedDateTimeConverter.java
@@ -0,0 +1,46 @@
+package org.springframework.data.jdbc.core.dialect;
+
+import org.h2.api.TimestampWithTimeZone;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.ReadingConverter;
+
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+
+/**
+ * Converter converting from an H2 internal representation of a timestamp with time zone to an {@link java.time.ZonedDateTime}.
+ *
+ * Only required for H2 versions < 2.0
+ *
+ * @author Mikhail Polivakha
+ */
+@ReadingConverter
+public enum H2TimestampWithTimeZoneToZonedDateTimeConverter implements Converter<TimestampWithTimeZone, ZonedDateTime> {
+
+    INSTANCE;
+
+    @Override
+    public ZonedDateTime convert(TimestampWithTimeZone source) {
+
+        long nanosInSecond = 1_000_000_000;
+        long nanosInMinute = nanosInSecond * 60;
+        long nanosInHour = nanosInMinute * 60;
+
+        long hours = (source.getNanosSinceMidnight() / nanosInHour);
+
+        long nanosInHours = hours * nanosInHour;
+        long nanosLeft = source.getNanosSinceMidnight() - nanosInHours;
+        long minutes = nanosLeft / nanosInMinute;
+
+        long nanosInMinutes = minutes * nanosInMinute;
+        nanosLeft -= nanosInMinutes;
+        long seconds = nanosLeft / nanosInSecond;
+
+        long nanosInSeconds = seconds * nanosInSecond;
+        nanosLeft -= nanosInSeconds;
+        ZoneOffset offset = ZoneOffset.ofTotalSeconds(source.getTimeZoneOffsetSeconds());
+
+        return ZonedDateTime.of(source.getYear(), source.getMonth(), source.getDay(), (int) hours, (int) minutes,
+                (int) seconds, (int) nanosLeft, offset);
+    }
+}
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcDb2Dialect.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcDb2Dialect.java
index 45a8c58eca..789cd572aa 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcDb2Dialect.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcDb2Dialect.java
@@ -16,21 +16,26 @@
 package org.springframework.data.jdbc.core.dialect;
 
 import java.sql.Timestamp;
+import java.sql.Types;
 import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.jdbc.core.convert.Jsr310TimestampBasedConverters;
 import org.springframework.data.relational.core.dialect.Db2Dialect;
+import org.springframework.lang.NonNull;
 
 /**
  * {@link Db2Dialect} that registers JDBC specific converters.
  *
  * @author Jens Schauder
  * @author Christoph Strobl
+ * @author Mikhail Polivakha
  * @since 2.3
  */
 public class JdbcDb2Dialect extends Db2Dialect {
@@ -39,16 +44,48 @@ public class JdbcDb2Dialect extends Db2Dialect {
 
 	protected JdbcDb2Dialect() {}
 
+	@NonNull
 	@Override
 	public Collection<Object> getConverters() {
-
 		List<Object> converters = new ArrayList<>(super.getConverters());
 		converters.add(OffsetDateTimeToTimestampConverter.INSTANCE);
+		converters.add(ZonedDateTimeToTimestampConverter.INSTANCE);
 		converters.add(Jsr310TimestampBasedConverters.LocalDateTimeToTimestampConverter.INSTANCE);
-
 		return converters;
 	}
 
+	/**
+	 * Unfortunately, DB2 jdbc driver does not support Js310 (new Java date/time API) classes. Therefore for this
+	 * dialect we need to convert these classes into {@link Timestamp}.
+	 *
+	 * @see OffsetDateTimeToTimestampConverter
+	 * @see ZonedDateTimeToTimestampConverter
+	 */
+	@NonNull
+    @Override
+    public Map<Class<?>, Class<?>> getCustomJdbcColumnsMappings() {
+		final Map<Class<?>, Class<?>> db2CustomJdbcColumnsMappings = super.getCustomJdbcColumnsMappings();
+		db2CustomJdbcColumnsMappings.put(ZonedDateTime.class, Timestamp.class);
+		db2CustomJdbcColumnsMappings.put(OffsetDateTime.class, Timestamp.class);
+		return db2CustomJdbcColumnsMappings;
+	}
+
+	/**
+	 * Unfortunately, DB2 jdbc driver does not support {@link Types#TIMESTAMP_WITH_TIMEZONE} as a type. Therefore for this
+	 * dialect we need to use {@link Types#TIMESTAMP} and pass {@link Timestamp} instead of Jsr310 classes
+	 *
+	 * @see OffsetDateTimeToTimestampConverter
+	 * @see ZonedDateTimeToTimestampConverter
+	 */
+	@NonNull
+	@Override
+	public Map<Class<?>, Integer> getCustomSqlCodesMappings() {
+		final Map<Class<?>, Integer> db2CustomSqlCodesMappings = super.getCustomSqlCodesMappings();
+		db2CustomSqlCodesMappings.put(ZonedDateTime.class, Types.TIMESTAMP);
+		db2CustomSqlCodesMappings.put(OffsetDateTime.class, Types.TIMESTAMP);
+		return db2CustomSqlCodesMappings;
+	}
+
 	/**
 	 * {@link WritingConverter} from {@link OffsetDateTime} to {@link Timestamp}. The conversion preserves the
 	 * {@link java.time.Instant} represented by {@link OffsetDateTime}
@@ -66,4 +103,22 @@ public Timestamp convert(OffsetDateTime source) {
 			return Timestamp.from(source.toInstant());
 		}
 	}
+
+	/**
+	 * {@link WritingConverter} from {@link ZonedDateTime} to {@link Timestamp}. The conversion preserves the
+	 * {@link java.time.Instant} represented by {@link ZonedDateTime}
+	 *
+	 * @author Mikhail Polivakha
+	 * @since 3.0
+	 */
+	@WritingConverter
+	enum ZonedDateTimeToTimestampConverter implements Converter<ZonedDateTime, Timestamp> {
+
+		INSTANCE;
+
+		@Override
+		public Timestamp convert(ZonedDateTime source) {
+			return Timestamp.from(source.toInstant());
+		}
+	}
 }
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcH2Dialect.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcH2Dialect.java
index 15807ff4e1..20067d370a 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcH2Dialect.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcH2Dialect.java
@@ -27,6 +27,8 @@
  *
  * @author Jens Schauder
  * @author Christoph Strobl
+ * @author Mikhail Polivakha
+ *
  * @since 2.3
  */
 public class JdbcH2Dialect extends H2Dialect {
@@ -40,17 +42,18 @@ public Collection<Object> getConverters() {
 
 		final Collection<Object> originalConverters = super.getConverters();
 
-		if (isH2belowVersion2()) {
+		if (doesH2TimestampWithTimeZoneClassPresent()) {
 
 			List<Object> converters = new ArrayList<>(originalConverters);
 			converters.add(H2TimestampWithTimeZoneToOffsetDateTimeConverter.INSTANCE);
+			converters.add(H2TimestampWithTimeZoneToZonedDateTimeConverter.INSTANCE);
 			return converters;
 		}
 
 		return originalConverters;
 	}
 
-	static boolean isH2belowVersion2() {
+	static boolean doesH2TimestampWithTimeZoneClassPresent() {
 
 		try {
 
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcPostgresDialect.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcPostgresDialect.java
index 6c7bc82945..ed1a39c454 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcPostgresDialect.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcPostgresDialect.java
@@ -17,13 +17,21 @@
 
 import java.sql.JDBCType;
 import java.sql.SQLType;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Collection;
 
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.relational.core.dialect.PostgresDialect;
+import org.springframework.lang.NonNull;
 
 /**
  * JDBC specific Postgres Dialect.
  *
  * @author Jens Schauder
+ * @author Mikhail Polivakha
  * @since 2.3
  */
 public class JdbcPostgresDialect extends PostgresDialect implements JdbcDialect {
@@ -37,6 +45,14 @@ public JdbcArrayColumns getArraySupport() {
 		return ARRAY_COLUMNS;
 	}
 
+	@NonNull
+	@Override
+	public Collection<Object> getConverters() {
+		final Collection<Object> converters = new ArrayList<>(super.getConverters());
+		converters.add(ZonedDateTimeToOffsetDateTimeWritingConverter.INSTANCE);
+		return converters;
+	}
+
 	static class JdbcPostgresArrayColumns extends PostgresArrayColumns implements JdbcArrayColumns {
 
 		@Override
@@ -57,4 +73,19 @@ public String getArrayTypeName(SQLType jdbcType) {
 			return jdbcType.getName();
 		}
 	}
+
+	/**
+	 * Unfortunately, PostgresSQL jdbc driver can accept {@link OffsetDateTime} as
+	 * {@link java.sql.Types#TIMESTAMP_WITH_TIMEZONE}, but not {@link ZonedDateTime}
+	 */
+	@WritingConverter
+	enum ZonedDateTimeToOffsetDateTimeWritingConverter implements Converter<ZonedDateTime, OffsetDateTime> {
+
+		INSTANCE;
+
+		@Override
+		public OffsetDateTime convert(ZonedDateTime source) {
+			return source.toOffsetDateTime();
+		}
+	}
 }
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcSqlServerDialect.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcSqlServerDialect.java
index 9883c23a46..c4e4a72665 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcSqlServerDialect.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/dialect/JdbcSqlServerDialect.java
@@ -18,30 +18,37 @@
 import microsoft.sql.DateTimeOffset;
 
 import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.convert.ReadingConverter;
+import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.relational.core.dialect.SqlServerDialect;
+import org.springframework.lang.NonNull;
 
 /**
  * {@link SqlServerDialect} that registers JDBC specific converters.
  *
  * @author Jens Schauder
  * @author Christoph Strobl
+ * @author Mikhail Polivakha
+ *
  * @since 2.3
  */
 public class JdbcSqlServerDialect extends SqlServerDialect {
 
 	public static JdbcSqlServerDialect INSTANCE = new JdbcSqlServerDialect();
 
+	@NonNull
 	@Override
 	public Collection<Object> getConverters() {
-
 		List<Object> converters = new ArrayList<>(super.getConverters());
 		converters.add(DateTimeOffsetToOffsetDateTimeConverter.INSTANCE);
+		converters.add(ZonedDateTimeToOffsetDateTimeConverter.INSTANCE);
+		converters.add(DateTimeOffsetToZonedDateTimeConverter.INSTANCE);
 		return converters;
 	}
 
@@ -55,4 +62,30 @@ public OffsetDateTime convert(DateTimeOffset source) {
 			return source.getOffsetDateTime();
 		}
 	}
+
+	@ReadingConverter
+	enum DateTimeOffsetToZonedDateTimeConverter implements Converter<DateTimeOffset, ZonedDateTime> {
+
+		INSTANCE;
+
+		@Override
+		public ZonedDateTime convert(DateTimeOffset source) {
+			return source.getOffsetDateTime().toZonedDateTime();
+		}
+	}
+
+	/**
+	 * Unfortunately, SQl Server jdbc driver can accept {@link OffsetDateTime} as
+	 * {@link java.sql.Types#TIMESTAMP_WITH_TIMEZONE}, but not {@link ZonedDateTime}
+	 */
+	@WritingConverter
+	enum ZonedDateTimeToOffsetDateTimeConverter implements Converter<ZonedDateTime, OffsetDateTime> {
+
+		INSTANCE;
+
+		@Override
+		public OffsetDateTime convert(ZonedDateTime source) {
+			return source.toOffsetDateTime();
+		}
+	}
 }
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/AbstractJdbcQuery.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/AbstractJdbcQuery.java
index 2d7df924ed..cb049b166c 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/AbstractJdbcQuery.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/AbstractJdbcQuery.java
@@ -47,7 +47,7 @@
 public abstract class AbstractJdbcQuery implements RepositoryQuery {
 
 	private final JdbcQueryMethod queryMethod;
-	private final NamedParameterJdbcOperations operations;
+	protected final NamedParameterJdbcOperations operations;
 
 	/**
 	 * Creates a new {@link AbstractJdbcQuery} for the given {@link JdbcQueryMethod} and
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/StringBasedJdbcQuery.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/StringBasedJdbcQuery.java
index 6f832525df..f382814fd2 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/StringBasedJdbcQuery.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/query/StringBasedJdbcQuery.java
@@ -15,18 +15,15 @@
  */
 package org.springframework.data.jdbc.repository.query;
 
-import static org.springframework.data.jdbc.repository.query.JdbcQueryExecution.*;
-
-import java.lang.reflect.Constructor;
-import java.sql.JDBCType;
-
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.jdbc.core.convert.JdbcColumnTypes;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
 import org.springframework.data.jdbc.core.convert.JdbcValue;
+import org.springframework.data.jdbc.repository.config.DialectResolver;
 import org.springframework.data.jdbc.support.JdbcUtil;
+import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.repository.query.RelationalParameterAccessor;
 import org.springframework.data.relational.repository.query.RelationalParametersParameterAccessor;
@@ -43,6 +40,11 @@
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
+import java.lang.reflect.Constructor;
+import java.sql.JDBCType;
+
+import static org.springframework.data.jdbc.repository.query.JdbcQueryExecution.ResultProcessingConverter;
+
 /**
  * A query to be executed based on a repository method, it's annotated SQL query and the arguments provided to the
  * method.
@@ -53,6 +55,7 @@
  * @author Maciej Walkowiak
  * @author Mark Paluch
  * @author Hebert Coelho
+ * @author Mikhail Polivakha
  * @since 2.0
  */
 public class StringBasedJdbcQuery extends AbstractJdbcQuery {
@@ -158,10 +161,13 @@ private void convertAndAddParameter(MapSqlParameterSource parameters, Parameter
 		String parameterName = p.getName().orElseThrow(() -> new IllegalStateException(PARAMETER_NEEDS_TO_BE_NAMED));
 
 		Class<?> parameterType = queryMethod.getParameters().getParameter(p.getIndex()).getType();
-		Class<?> conversionTargetType = JdbcColumnTypes.INSTANCE.resolvePrimitiveType(parameterType);
+		Class<?> conversionTargetType = JdbcColumnTypes.INSTANCE.resolvePrimitiveType(parameterType, DialectResolver.getDialect(operations.getJdbcOperations()));
 
-		JdbcValue jdbcValue = converter.writeJdbcValue(value, conversionTargetType,
-				JdbcUtil.sqlTypeFor(conversionTargetType));
+		JdbcValue jdbcValue = converter.writeJdbcValue(
+				value,
+				conversionTargetType,
+				JdbcUtil.sqlTypeFor(conversionTargetType, DialectResolver.getDialect(operations.getJdbcOperations()))
+		);
 
 		JDBCType jdbcType = jdbcValue.getJdbcType();
 		if (jdbcType == null) {
diff --git a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/support/JdbcUtil.java b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/support/JdbcUtil.java
index e5d0c291f4..d8865b22af 100644
--- a/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/support/JdbcUtil.java
+++ b/spring-data-jdbc/src/main/java/org/springframework/data/jdbc/support/JdbcUtil.java
@@ -23,9 +23,12 @@
 import java.sql.Timestamp;
 import java.sql.Types;
 import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
 
+import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.jdbc.support.JdbcUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -35,35 +38,37 @@
  *
  * @author Jens Schauder
  * @author Thomas Lang
+ * @author Mikhail Polivakha
  */
 public final class JdbcUtil {
 
-	private static final Map<Class<?>, Integer> sqlTypeMappings = new HashMap<>();
+	private static final Map<Class<?>, Integer> commonSqlTypeMappings = new HashMap<>();
 
 	static {
 
-		sqlTypeMappings.put(String.class, Types.VARCHAR);
-		sqlTypeMappings.put(BigInteger.class, Types.BIGINT);
-		sqlTypeMappings.put(BigDecimal.class, Types.DECIMAL);
-		sqlTypeMappings.put(Byte.class, Types.TINYINT);
-		sqlTypeMappings.put(byte.class, Types.TINYINT);
-		sqlTypeMappings.put(Short.class, Types.SMALLINT);
-		sqlTypeMappings.put(short.class, Types.SMALLINT);
-		sqlTypeMappings.put(Integer.class, Types.INTEGER);
-		sqlTypeMappings.put(int.class, Types.INTEGER);
-		sqlTypeMappings.put(Long.class, Types.BIGINT);
-		sqlTypeMappings.put(long.class, Types.BIGINT);
-		sqlTypeMappings.put(Double.class, Types.DOUBLE);
-		sqlTypeMappings.put(double.class, Types.DOUBLE);
-		sqlTypeMappings.put(Float.class, Types.REAL);
-		sqlTypeMappings.put(float.class, Types.REAL);
-		sqlTypeMappings.put(Boolean.class, Types.BIT);
-		sqlTypeMappings.put(boolean.class, Types.BIT);
-		sqlTypeMappings.put(byte[].class, Types.VARBINARY);
-		sqlTypeMappings.put(Date.class, Types.DATE);
-		sqlTypeMappings.put(Time.class, Types.TIME);
-		sqlTypeMappings.put(Timestamp.class, Types.TIMESTAMP);
-		sqlTypeMappings.put(OffsetDateTime.class, Types.TIMESTAMP_WITH_TIMEZONE);
+		commonSqlTypeMappings.put(String.class, Types.VARCHAR);
+		commonSqlTypeMappings.put(BigInteger.class, Types.BIGINT);
+		commonSqlTypeMappings.put(BigDecimal.class, Types.DECIMAL);
+		commonSqlTypeMappings.put(Byte.class, Types.TINYINT);
+		commonSqlTypeMappings.put(byte.class, Types.TINYINT);
+		commonSqlTypeMappings.put(Short.class, Types.SMALLINT);
+		commonSqlTypeMappings.put(short.class, Types.SMALLINT);
+		commonSqlTypeMappings.put(Integer.class, Types.INTEGER);
+		commonSqlTypeMappings.put(int.class, Types.INTEGER);
+		commonSqlTypeMappings.put(Long.class, Types.BIGINT);
+		commonSqlTypeMappings.put(long.class, Types.BIGINT);
+		commonSqlTypeMappings.put(Double.class, Types.DOUBLE);
+		commonSqlTypeMappings.put(double.class, Types.DOUBLE);
+		commonSqlTypeMappings.put(Float.class, Types.REAL);
+		commonSqlTypeMappings.put(float.class, Types.REAL);
+		commonSqlTypeMappings.put(Boolean.class, Types.BIT);
+		commonSqlTypeMappings.put(boolean.class, Types.BIT);
+		commonSqlTypeMappings.put(byte[].class, Types.VARBINARY);
+		commonSqlTypeMappings.put(Date.class, Types.DATE);
+		commonSqlTypeMappings.put(Time.class, Types.TIME);
+		commonSqlTypeMappings.put(Timestamp.class, Types.TIMESTAMP);
+		commonSqlTypeMappings.put(OffsetDateTime.class, Types.TIMESTAMP_WITH_TIMEZONE);
+		commonSqlTypeMappings.put(ZonedDateTime.class, Types.TIMESTAMP_WITH_TIMEZONE);
 	}
 
 	private JdbcUtil() {
@@ -76,18 +81,52 @@ private JdbcUtil() {
 	 *
 	 * @param type The type of value to be bound to a {@link java.sql.PreparedStatement}.
 	 * @return One of the values defined in {@link Types} or {@link JdbcUtils#TYPE_UNKNOWN}.
+	 * @deprecated because this method will make lookup only in common sql types map and therefore
+	 * 			   will not take into account dialect specific sql types codes mappings. When possible,
+	 * 			   please, use {@link #sqlTypeFor(Class, Dialect)}
 	 */
+	@Deprecated
 	public static int sqlTypeFor(Class<?> type) {
 
 		Assert.notNull(type, "Type must not be null.");
 
-		return sqlTypeMappings.keySet().stream() //
+		return commonSqlTypeMappings.keySet().stream() //
 				.filter(k -> k.isAssignableFrom(type)) //
 				.findFirst() //
-				.map(sqlTypeMappings::get) //
+				.map(commonSqlTypeMappings::get) //
 				.orElse(JdbcUtils.TYPE_UNKNOWN);
 	}
 
+	/**
+	 * Returns the {@link Types} value suitable for passing a value of the provided type to a
+	 * {@link java.sql.PreparedStatement} giving regards to passed dialect specific sql codes
+	 * mappings
+	 *
+	 * The main motivation for this method is, in general, the fact that we cannot assign the same int code for the same java class for
+	 * each dialect. For example, currently, there MySQL Driver cannot handle {@link Types#TIMESTAMP_WITH_TIMEZONE}
+	 * to be set by the means of {@link java.sql.PreparedStatement#setObject(int, Object, int)}. If we We need to instead set
+	 * it be the mean
+	 *
+	 * @param type The type of value to be bound to a {@link java.sql.PreparedStatement}.
+	 * @param dialect represents the dialect, in the context of which the int sql code must be derived
+	 * @return the int code of sql type in regards to custom dialect mappings
+	 *
+	 * @see Types
+	 */
+	public static int sqlTypeFor(Class<?> type, Dialect dialect) {
+
+		Assert.notNull(type, "Type must not be null.");
+
+		return Optional
+				.ofNullable(dialect.getCustomSqlCodesMappings().get(type))
+				.orElse(commonSqlTypeMappings.keySet().stream()
+						.filter(k -> k.isAssignableFrom(type))
+						.findFirst()
+						.map(commonSqlTypeMappings::get)
+						.orElse(JdbcUtils.TYPE_UNKNOWN)
+				);
+	}
+
 	/**
 	 * Converts a {@link JDBCType} to an {@code int} value as defined in {@link Types}.
 	 *
@@ -121,9 +160,33 @@ public static JDBCType jdbcTypeFor(int sqlType) {
 	 *
 	 * @param type The type of value to be bound to a {@link java.sql.PreparedStatement}.
 	 * @return a matching {@link JDBCType} instance or {@literal null}.
+	 * @deprecated because this method will make lookup only in common sql types map and therefore
+	 * 			   will not take into account dialect specific sql types codes mappings. When possible,
+	 * 			   please, use {@link #jdbcTypeFor(Class, Dialect)}
 	 */
 	@Nullable
+	@Deprecated
 	public static JDBCType jdbcTypeFor(Class<?> type) {
 		return jdbcTypeFor(sqlTypeFor(type));
 	}
+
+
+	/**
+	 * Returns the {@link JDBCType} suitable for passing a value of the provided type to a
+	 * {@link java.sql.PreparedStatement} giving regards to passed dialect specific sql codes
+	 * mappings
+	 *
+	 * The main motivation for this method is, in general, the fact that we cannot assign the same int code for the same java class for
+	 * each dialect. For example, currently, there MySQL Driver cannot handle {@link Types#TIMESTAMP_WITH_TIMEZONE}
+	 * to be set by the means of {@link java.sql.PreparedStatement#setObject(int, Object, int)}. If we We need to instead set
+	 * it be the mean
+	 *
+	 * @param type The type of value to be bound to a {@link java.sql.PreparedStatement}.
+	 * @param dialect represents the dialect, in the context of which the int sql code must be derived
+	 * @return a matching {@link JDBCType} instance or {@literal null}.
+	 */
+	@Nullable
+	public static JDBCType jdbcTypeFor(Class<?> type, Dialect dialect) {
+		return jdbcTypeFor(sqlTypeFor(type, dialect));
+	}
 }
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverterUnitTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverterUnitTests.java
index cb60ed38fa..339889b721 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverterUnitTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/BasicJdbcConverterUnitTests.java
@@ -74,7 +74,7 @@ public void testTargetTypesForPropertyType() {
 		checkTargetType(softly, entity, "localDateTime", LocalDateTime.class);
 		checkTargetType(softly, entity, "localDate", Timestamp.class);
 		checkTargetType(softly, entity, "localTime", Timestamp.class);
-		checkTargetType(softly, entity, "zonedDateTime", String.class);
+		checkTargetType(softly, entity, "zonedDateTime", ZonedDateTime.class);
 		checkTargetType(softly, entity, "offsetDateTime", OffsetDateTime.class);
 		checkTargetType(softly, entity, "instant", Timestamp.class);
 		checkTargetType(softly, entity, "date", Date.class);
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategyUnitTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategyUnitTests.java
index 5098823c29..fc11792bc0 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategyUnitTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategyUnitTests.java
@@ -45,6 +45,7 @@
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.jdbc.core.ConnectionCallback;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
@@ -84,6 +85,11 @@ public void before() {
 
 		converter = new BasicJdbcConverter(context, relationResolver, new JdbcCustomConversions(),
 				new DefaultJdbcTypeFactory(jdbcOperations), dialect.getIdentifierProcessing());
+
+		when(jdbcOperations.execute(any(ConnectionCallback.class))).thenReturn(dialect);
+
+		when(namedJdbcOperations.getJdbcOperations()).thenReturn(jdbcOperations);
+
 		accessStrategy = new DefaultDataAccessStrategy( //
 				new SqlGeneratorSource(context, converter, dialect), //
 				context, //
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryCustomConversionIntegrationTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryCustomConversionIntegrationTests.java
index 790154e7dd..2309811617 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryCustomConversionIntegrationTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryCustomConversionIntegrationTests.java
@@ -15,14 +15,6 @@
  */
 package org.springframework.data.jdbc.repository;
 
-import static java.util.Arrays.*;
-import static org.assertj.core.api.Assertions.*;
-import static org.assertj.core.api.SoftAssertions.*;
-import static org.springframework.test.context.TestExecutionListeners.MergeMode.*;
-
-import java.math.BigDecimal;
-import java.sql.JDBCType;
-import java.util.Date;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -36,6 +28,8 @@
 import org.springframework.data.convert.WritingConverter;
 import org.springframework.data.jdbc.core.convert.JdbcCustomConversions;
 import org.springframework.data.jdbc.core.convert.JdbcValue;
+import org.springframework.data.jdbc.core.dialect.H2TimestampWithTimeZoneToOffsetDateTimeConverter;
+import org.springframework.data.jdbc.core.dialect.H2TimestampWithTimeZoneToZonedDateTimeConverter;
 import org.springframework.data.jdbc.repository.support.JdbcRepositoryFactory;
 import org.springframework.data.jdbc.testing.AssumeFeatureTestExecutionListener;
 import org.springframework.data.jdbc.testing.TestConfiguration;
@@ -45,6 +39,15 @@
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.math.BigDecimal;
+import java.sql.JDBCType;
+import java.util.Date;
+
+import static java.util.Arrays.asList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.SoftAssertions.assertSoftly;
+import static org.springframework.test.context.TestExecutionListeners.MergeMode.MERGE_WITH_DEFAULTS;
+
 /**
  * Tests storing and retrieving data types that get processed by custom conversions.
  *
@@ -69,21 +72,29 @@ Class<?> testClass() {
 		}
 
 		@Bean
-		EntityWithBooleanRepository repository() {
+		EntityWithBooleanRepository entityWithBooleanRepository() {
 			return factory.getRepository(EntityWithBooleanRepository.class);
 		}
 
 		@Bean
 		JdbcCustomConversions jdbcCustomConversions() {
-			return new JdbcCustomConversions(asList(StringToBigDecimalConverter.INSTANCE, BigDecimalToString.INSTANCE,
-					CustomIdReadingConverter.INSTANCE, CustomIdWritingConverter.INSTANCE));
+			return new JdbcCustomConversions(
+					asList(
+							StringToBigDecimalConverter.INSTANCE,
+							BigDecimalToString.INSTANCE,
+							CustomIdReadingConverter.INSTANCE,
+							CustomIdWritingConverter.INSTANCE,
+							H2TimestampWithTimeZoneToZonedDateTimeConverter.INSTANCE,
+							H2TimestampWithTimeZoneToOffsetDateTimeConverter.INSTANCE
+					)
+			);
 		}
 	}
 
 	@Autowired EntityWithBooleanRepository repository;
 
 	/**
-	 * In PostrgreSQL this fails if a simple converter like the following is used.
+	 * In PostgreSQL this fails if a simple converter like the following is used.
 	 *
 	 * <pre class="code">
 	 *
@@ -146,7 +157,6 @@ public void saveAndLoadAnEntityWithReference() {
 	interface EntityWithBooleanRepository extends CrudRepository<EntityWithStringyBigDecimal, CustomId> {}
 
 	private static class EntityWithStringyBigDecimal {
-
 		@Id CustomId id;
 		String stringyNumber;
 		OtherEntity reference;
@@ -213,5 +223,4 @@ public CustomId convert(Number source) {
 			return new CustomId(source.longValue());
 		}
 	}
-
-}
+}
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryIntegrationTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryIntegrationTests.java
index 66201877e3..2e79a61b1a 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryIntegrationTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryIntegrationTests.java
@@ -30,12 +30,16 @@
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
 import org.junit.jupiter.api.extension.ExtendWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
@@ -87,6 +91,8 @@ public class JdbcRepositoryIntegrationTests {
 	@Autowired NamedParameterJdbcTemplate template;
 	@Autowired DummyEntityRepository repository;
 	@Autowired MyEventListener eventListener;
+	@Autowired EntityWithZonedDateTimeRepository entityWithZonedDateTimeRepository;
+	@Autowired EntityWithOffsetDateTimeRepository entityWithOffsetDateTimeRepository;
 
 	private static DummyEntity createDummyEntity() {
 
@@ -96,6 +102,17 @@ private static DummyEntity createDummyEntity() {
 		return entity;
 	}
 
+	private static class EntityWithZonedDateTime {
+		@Id private Long id;
+		private ZonedDateTime createdAt;
+	}
+
+	private static class EntityWithOffsetDateTime {
+		@Id private Long id;
+		private OffsetDateTime createdAt;
+	}
+
+
 	@BeforeEach
 	public void before() {
 
@@ -549,6 +566,37 @@ void queryByAggregateReference() {
 		assertThat(result).extracting(e -> e.idProp).containsExactly(two.idProp);
 	}
 
+	/**
+	 * DATAJDBC-1089
+	 */
+	@Test
+	public void testZonedDateTimeToTimestampConversion() {
+		EntityWithZonedDateTime entity = new EntityWithZonedDateTime();
+		entity.createdAt = ZonedDateTime.now(ZoneOffset.ofHours(3));
+
+		final EntityWithZonedDateTime persistedEntity = entityWithZonedDateTimeRepository.save(entity);
+		final Optional<EntityWithZonedDateTime> foundEntity = entityWithZonedDateTimeRepository.findById(persistedEntity.id);
+
+		assertThat(foundEntity).isPresent();
+		assertThat(persistedEntity.createdAt.toEpochSecond()).isEqualTo(foundEntity.get().createdAt.toEpochSecond());
+	}
+
+	/**
+	 * DATAJDBC-1089
+	 */
+	@Test
+	public void testOffsetDateTimeToTimestampConversion() {
+		EntityWithOffsetDateTime entity = new EntityWithOffsetDateTime();
+		entity.createdAt = OffsetDateTime.now(ZoneOffset.ofHours(3));
+
+		final EntityWithOffsetDateTime persistedEntity = entityWithOffsetDateTimeRepository.save(entity);
+		final Optional<EntityWithOffsetDateTime> foundEntity = entityWithOffsetDateTimeRepository.findById(persistedEntity.id);
+
+		assertThat(foundEntity).isPresent();
+		assertThat(persistedEntity.createdAt.toEpochSecond()).isEqualTo(foundEntity.get().createdAt.toEpochSecond());
+	}
+
+
 	private Instant createDummyBeforeAndAfterNow() {
 
 		Instant now = Instant.now();
@@ -572,6 +620,10 @@ private Instant createDummyBeforeAndAfterNow() {
 		return now;
 	}
 
+	interface EntityWithZonedDateTimeRepository extends CrudRepository<EntityWithZonedDateTime, Long> {};
+
+	interface EntityWithOffsetDateTimeRepository extends CrudRepository<EntityWithOffsetDateTime, Long> {};
+
 	interface DummyEntityRepository extends CrudRepository<DummyEntity, Long> {
 
 		List<DummyEntity> findAllByNamedQuery();
@@ -643,6 +695,16 @@ DummyEntityRepository dummyEntityRepository() {
 			return factory.getRepository(DummyEntityRepository.class);
 		}
 
+		@Bean
+		EntityWithZonedDateTimeRepository entityWithZonedDateTimeRepository() {
+			return factory.getRepository(EntityWithZonedDateTimeRepository.class);
+		}
+
+		@Bean
+		EntityWithOffsetDateTimeRepository entityWithOffsetDateTimeRepository() {
+			return factory.getRepository(EntityWithOffsetDateTimeRepository.class);
+		}
+
 		@Bean
 		NamedQueries namedQueries() throws IOException {
 
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/SimpleJdbcRepositoryEventsUnitTests.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/SimpleJdbcRepositoryEventsUnitTests.java
index 31d51e8663..2fd71cbfae 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/SimpleJdbcRepositoryEventsUnitTests.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/SimpleJdbcRepositoryEventsUnitTests.java
@@ -27,6 +27,8 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Optional;
+import java.util.function.BiFunction;
 
 import org.assertj.core.groups.Tuple;
 import org.junit.jupiter.api.BeforeEach;
@@ -45,14 +47,18 @@
 import org.springframework.data.jdbc.core.convert.JdbcCustomConversions;
 import org.springframework.data.jdbc.core.convert.SqlGeneratorSource;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
+import org.springframework.data.jdbc.repository.config.DialectResolver;
+import org.springframework.data.jdbc.repository.config.DialectResolver.DefaultDialectProvider;
 import org.springframework.data.jdbc.repository.support.JdbcRepositoryFactory;
 import org.springframework.data.jdbc.repository.support.SimpleJdbcRepository;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.H2Dialect;
 import org.springframework.data.relational.core.dialect.HsqlDbDialect;
+import org.springframework.data.relational.core.dialect.PostgresDialect;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.event.*;
 import org.springframework.data.repository.PagingAndSortingRepository;
+import org.springframework.jdbc.core.ConnectionCallback;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.core.namedparam.SqlParameterSource;
@@ -80,11 +86,17 @@ public class SimpleJdbcRepositoryEventsUnitTests {
 	public void before() {
 
 		RelationalMappingContext context = new JdbcMappingContext();
+
 		NamedParameterJdbcOperations operations = createIdGeneratingOperations();
+
+		JdbcOperations mockJdbcOperations = operations.getJdbcOperations();
+
+		when(mockJdbcOperations.execute(any(ConnectionCallback.class))).thenReturn(PostgresDialect.INSTANCE);
+
 		DelegatingDataAccessStrategy delegatingDataAccessStrategy = new DelegatingDataAccessStrategy();
 		Dialect dialect = HsqlDbDialect.INSTANCE;
 		JdbcConverter converter = new BasicJdbcConverter(context, delegatingDataAccessStrategy, new JdbcCustomConversions(),
-				new DefaultJdbcTypeFactory(operations.getJdbcOperations()), dialect.getIdentifierProcessing());
+				new DefaultJdbcTypeFactory(mockJdbcOperations), dialect.getIdentifierProcessing());
 		SqlGeneratorSource generatorSource = new SqlGeneratorSource(context, converter, dialect);
 
 		this.dataAccessStrategy = spy(new DefaultDataAccessStrategy(generatorSource, context, converter, operations));
@@ -290,10 +302,12 @@ private static NamedParameterJdbcOperations createIdGeneratingOperations() {
 			return 1;
 		};
 
+		JdbcOperations mockJdbcOperations = mock(JdbcOperations.class);
+
 		NamedParameterJdbcOperations operations = mock(NamedParameterJdbcOperations.class);
 		when(operations.update(anyString(), any(SqlParameterSource.class), any(KeyHolder.class)))
 				.thenAnswer(setIdInKeyHolder);
-		when(operations.getJdbcOperations()).thenReturn(mock(JdbcOperations.class));
+		when(operations.getJdbcOperations()).thenReturn(mockJdbcOperations);
 		return operations;
 	}
 
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MariaDBDataSourceConfiguration.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MariaDBDataSourceConfiguration.java
index d703bf68c4..2c7c74db8e 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MariaDBDataSourceConfiguration.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MariaDBDataSourceConfiguration.java
@@ -71,10 +71,8 @@ protected DataSource createDataSource() {
 
 	@Override
 	public void afterPropertiesSet() throws Exception {
-
 		try (Connection connection = createDataSource().getConnection()) {
-			ScriptUtils.executeSqlScript(connection,
-					new ByteArrayResource("DROP DATABASE test;CREATE DATABASE test;".getBytes()));
+			ScriptUtils.executeSqlScript(connection, new ByteArrayResource("DROP DATABASE IF EXISTS test;CREATE DATABASE IF NOT EXISTS test;".getBytes()));
 		}
 	}
 }
diff --git a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MySqlDataSourceConfiguration.java b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MySqlDataSourceConfiguration.java
index 8da00a7744..75a20351be 100644
--- a/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MySqlDataSourceConfiguration.java
+++ b/spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/MySqlDataSourceConfiguration.java
@@ -73,8 +73,7 @@ protected DataSource createDataSource() {
 	public void afterPropertiesSet() throws Exception {
 
 		try (Connection connection = createDataSource().getConnection()) {
-			ScriptUtils.executeSqlScript(connection,
-					new ByteArrayResource("DROP DATABASE test;CREATE DATABASE test;".getBytes()));
+			ScriptUtils.executeSqlScript(connection, new ByteArrayResource("DROP DATABASE IF EXISTS test; CREATE DATABASE IF NOT EXISTS test;".getBytes()));
 		}
 	}
 
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-db2.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-db2.sql
index e75592d0bc..328e3b1d42 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-db2.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-db2.sql
@@ -2,4 +2,4 @@ DROP TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL;
 DROP TABLE OTHER_ENTITY;
 
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( id BIGINT GENERATED BY DEFAULT AS IDENTITY(START WITH 1) PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID  BIGINT GENERATED BY DEFAULT AS IDENTITY(START WITH 1) PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
+CREATE TABLE OTHER_ENTITY ( ID  BIGINT GENERATED BY DEFAULT AS IDENTITY(START WITH 1) PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-h2.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-h2.sql
index d383545694..7eece207e8 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-h2.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-h2.sql
@@ -1,2 +1,2 @@
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( id IDENTITY PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID IDENTITY PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
+CREATE TABLE OTHER_ENTITY ( ID IDENTITY PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-hsql.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-hsql.sql
index 78d9c930b7..7eece207e8 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-hsql.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-hsql.sql
@@ -1,3 +1,2 @@
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( id IDENTITY PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID IDENTITY PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
-
+CREATE TABLE OTHER_ENTITY ( ID IDENTITY PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mariadb.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mariadb.sql
index e89bb0d951..50fb20ce10 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mariadb.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mariadb.sql
@@ -1,2 +1,2 @@
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( id BIGINT AUTO_INCREMENT PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
+CREATE TABLE OTHER_ENTITY ( ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mssql.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mssql.sql
index 34b2982692..6eb1afcadc 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mssql.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mssql.sql
@@ -2,4 +2,4 @@ DROP TABLE OTHER_ENTITY;
 DROP TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL;
 
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( id BIGINT IDENTITY PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID BIGINT IDENTITY PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
+CREATE TABLE OTHER_ENTITY ( ID BIGINT IDENTITY PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mysql.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mysql.sql
index 34776fc5a8..ddd20eabba 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mysql.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-mysql.sql
@@ -1,2 +1,2 @@
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( ID BIGINT AUTO_INCREMENT PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
+CREATE TABLE OTHER_ENTITY ( ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-oracle.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-oracle.sql
index 1b02ef7214..8ec5b011e7 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-oracle.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-oracle.sql
@@ -10,5 +10,4 @@ CREATE TABLE OTHER_ENTITY (
     ID  NUMBER GENERATED BY DEFAULT ON NULL AS IDENTITY PRIMARY KEY,
     CREATED DATE,
     ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER
-);
-
+);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-postgres.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-postgres.sql
index c376dcf03f..36b46c27d9 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-postgres.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryCustomConversionIntegrationTests-postgres.sql
@@ -1,2 +1,2 @@
 CREATE TABLE ENTITY_WITH_STRINGY_BIG_DECIMAL ( id SERIAL PRIMARY KEY, Stringy_number DECIMAL(20,10));
-CREATE TABLE OTHER_ENTITY ( ID SERIAL PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
+CREATE TABLE OTHER_ENTITY ( ID SERIAL PRIMARY KEY, CREATED DATE, ENTITY_WITH_STRINGY_BIG_DECIMAL INTEGER);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-db2.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-db2.sql
index 34be74ec51..d58d89f383 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-db2.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-db2.sql
@@ -1,4 +1,6 @@
 DROP TABLE dummy_entity;
+DROP TABLE ENTITY_WITH_ZONED_DATE_TIME;
+DROP TABLE ENTITY_WITH_OFFSET_DATE_TIME;
 
 CREATE TABLE dummy_entity
 (
@@ -9,3 +11,6 @@ CREATE TABLE dummy_entity
     FLAG             BOOLEAN,
     REF              BIGINT
 );
+
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (ID BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY, CREATED_AT TIMESTAMP);
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (ID BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY, CREATED_AT TIMESTAMP);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-h2.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-h2.sql
index b3b93bc744..b5964a8179 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-h2.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-h2.sql
@@ -7,3 +7,6 @@ CREATE TABLE dummy_entity
     FLAG             BOOLEAN,
     REF              BIGINT
 );
+
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED_AT TIMESTAMP WITH TIME ZONE);
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED_AT TIMESTAMP WITH TIME ZONE);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-hsql.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-hsql.sql
index b3b93bc744..1aabce50ab 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-hsql.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-hsql.sql
@@ -7,3 +7,5 @@ CREATE TABLE dummy_entity
     FLAG             BOOLEAN,
     REF              BIGINT
 );
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (id IDENTITY PRIMARY KEY, CREATED_AT TIMESTAMP WITH TIME ZONE);
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (id IDENTITY PRIMARY KEY, CREATED_AT TIMESTAMP WITH TIME ZONE);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mariadb.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mariadb.sql
index 949e626399..0a48e46fb9 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mariadb.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mariadb.sql
@@ -7,3 +7,6 @@ CREATE TABLE dummy_entity
     FLAG             BOOLEAN,
     REF              BIGINT
 );
+
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED_AT TIMESTAMP(3));
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (ID BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED_AT TIMESTAMP(3));
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mssql.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mssql.sql
index 15f8881327..ff5077ec36 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mssql.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mssql.sql
@@ -1,4 +1,7 @@
 DROP TABLE IF EXISTS dummy_entity;
+DROP TABLE IF EXISTS ENTITY_WITH_ZONED_DATE_TIME;
+DROP TABLE IF EXISTS ENTITY_WITH_OFFSET_DATE_TIME;
+
 CREATE TABLE dummy_entity
 (
     id_Prop          BIGINT IDENTITY PRIMARY KEY,
@@ -8,3 +11,6 @@ CREATE TABLE dummy_entity
     FLAG             BIT,
     REF              BIGINT
 );
+
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (id BIGINT IDENTITY PRIMARY KEY, CREATED_AT DATETIMEOFFSET);
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (id BIGINT IDENTITY PRIMARY KEY, CREATED_AT DATETIMEOFFSET);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mysql.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mysql.sql
index e3baa94602..34d318a479 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mysql.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-mysql.sql
@@ -10,3 +10,6 @@ CREATE TABLE DUMMY_ENTITY
     FLAG             BIT(1),
     REF              BIGINT
 );
+
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (id BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED_AT TIMESTAMP(3));
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (id BIGINT AUTO_INCREMENT PRIMARY KEY, CREATED_AT TIMESTAMP(3));
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-oracle.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-oracle.sql
index e71eb63286..41621cfef9 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-oracle.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-oracle.sql
@@ -1,4 +1,6 @@
 DROP TABLE DUMMY_ENTITY CASCADE CONSTRAINTS PURGE;
+DROP TABLE ENTITY_WITH_ZONED_DATE_TIME CASCADE CONSTRAINTS PURGE;
+DROP TABLE ENTITY_WITH_OFFSET_DATE_TIME CASCADE CONSTRAINTS PURGE;
 
 CREATE TABLE DUMMY_ENTITY
 (
@@ -9,3 +11,13 @@ CREATE TABLE DUMMY_ENTITY
     FLAG             NUMBER(1,0),
     REF              NUMBER
 );
+
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (
+  id NUMBER GENERATED BY DEFAULT ON NULL AS IDENTITY PRIMARY KEY,
+  CREATED_AT TIMESTAMP WITH TIME ZONE
+);
+
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (
+  id NUMBER GENERATED BY DEFAULT ON NULL AS IDENTITY PRIMARY KEY,
+  CREATED_AT TIMESTAMP WITH TIME ZONE
+);
\ No newline at end of file
diff --git a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-postgres.sql b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-postgres.sql
index 97fc78c9da..37e485fa13 100644
--- a/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-postgres.sql
+++ b/spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.repository/JdbcRepositoryIntegrationTests-postgres.sql
@@ -1,4 +1,7 @@
 DROP TABLE dummy_entity;
+DROP TABLE ENTITY_WITH_ZONED_DATE_TIME;
+DROP TABLE ENTITY_WITH_OFFSET_DATE_TIME;
+
 CREATE TABLE dummy_entity
 (
     id_Prop          SERIAL PRIMARY KEY,
@@ -8,3 +11,5 @@ CREATE TABLE dummy_entity
     FLAG             BOOLEAN,
     REF              BIGINT
 );
+CREATE TABLE ENTITY_WITH_ZONED_DATE_TIME (id SERIAL PRIMARY KEY, CREATED_AT TIMESTAMPTZ);
+CREATE TABLE ENTITY_WITH_OFFSET_DATE_TIME (id SERIAL PRIMARY KEY, CREATED_AT TIMESTAMPTZ);
\ No newline at end of file
diff --git a/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/Dialect.java b/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/Dialect.java
index 5febb8c52f..ad01222de1 100644
--- a/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/Dialect.java
+++ b/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/Dialect.java
@@ -15,13 +15,18 @@
  */
 package org.springframework.data.relational.core.dialect;
 
+import java.sql.Timestamp;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Set;
 
+import org.jetbrains.annotations.NotNull;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.relational.core.sql.render.SelectRenderContext;
+import org.springframework.lang.NonNull;
 
 /**
  * Represents a dialect that is implemented by a particular database. Please note that not all features are supported by
@@ -120,4 +125,23 @@ default Set<Class<?>> simpleTypes() {
 	default InsertRenderContext getInsertRenderContext() {
 		return InsertRenderContexts.DEFAULT;
 	}
+
+	/**
+	 * @return the map of custom mappings from java classes to sql codes (integer sql codes present in {@link java.sql.Types})
+	 *         For example, for most of the drivers {@link java.sql.Types#TIMESTAMP_WITH_TIMEZONE} is OK, but for the others,
+	 *         like MySQL, we have to use {@link java.sql.Types#TIMESTAMP} and pass OffsetDateTime or ZonedDateTime by the means
+	 *         of {@link java.sql.PreparedStatement#setTimestamp(int, Timestamp)}.
+	 * @since 3.0
+	 */
+	@NonNull
+	default Map<Class<?>, Integer> getCustomSqlCodesMappings() { return new HashMap<>(); }
+
+	/**
+	 * @return the map of custom mappings from source java classes to appropriate jdbc columns java classes in the scope of current dialect.
+	 * 	          For example, most of the drivers can handle Jsr310 types directly, like {@link java.time.OffsetDateTime} or {@link java.time.ZonedDateTime},
+	 * 	          and we can just set them as is, but some others, like DB2 for example, require these types to be mapped/converted
+	 * 	          into {@link Timestamp} before passing to the underlying {@link java.sql.PreparedStatement}
+	 */
+	@NonNull
+	default Map<Class<?>, Class<?>> getCustomJdbcColumnsMappings() { return new HashMap<>(); }
 }
diff --git a/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/HsqlDbDialect.java b/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/HsqlDbDialect.java
index cf534de202..b877c8e3ef 100644
--- a/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/HsqlDbDialect.java
+++ b/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/HsqlDbDialect.java
@@ -15,11 +15,20 @@
  */
 package org.springframework.data.relational.core.dialect;
 
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.convert.WritingConverter;
+
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
+import java.util.ArrayList;
+import java.util.Collection;
+
 /**
  * A {@link Dialect} for HsqlDb.
  *
  * @author Jens Schauder
  * @author Myeonghyeon Lee
+ * @author Mikhail Polivakha
  */
 public class HsqlDbDialect extends AbstractDialect {
 
@@ -59,4 +68,26 @@ public Position getClausePosition() {
 			return Position.AFTER_ORDER_BY;
 		}
 	};
+
+	@Override
+	public Collection<Object> getConverters() {
+		Collection<Object> converters = new ArrayList<>(super.getConverters());
+		converters.add(ZonedDateTimeOffsetDateTimeWritingConverter.INSTANCE);
+		return converters;
+	}
+
+	/**
+	 * Unfortunately, HSqlDb jdbc driver can accept {@link OffsetDateTime} as
+	 * {@link java.sql.Types#TIMESTAMP_WITH_TIMEZONE}, but not {@link ZonedDateTime}
+	 */
+	@WritingConverter
+	enum ZonedDateTimeOffsetDateTimeWritingConverter implements Converter<ZonedDateTime, OffsetDateTime> {
+
+		INSTANCE;
+
+		@Override
+		public OffsetDateTime convert(ZonedDateTime source) {
+			return source.toOffsetDateTime();
+		}
+	}
 }
diff --git a/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/MySqlDialect.java b/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/MySqlDialect.java
index 6032582186..944a74195d 100644
--- a/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/MySqlDialect.java
+++ b/spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/MySqlDialect.java
@@ -15,13 +15,18 @@
  */
 package org.springframework.data.relational.core.dialect;
 
+import java.sql.Types;
+import java.time.OffsetDateTime;
+import java.time.ZonedDateTime;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Map;
 
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.LockOptions;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
+import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 
 /**
@@ -169,4 +174,17 @@ public IdentifierProcessing getIdentifierProcessing() {
 	public Collection<Object> getConverters() {
 		return Collections.singletonList(TimestampAtUtcToOffsetDateTimeConverter.INSTANCE);
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#getCustomSqlCodesMappings()
+	 */
+	@NonNull
+    @Override
+    public Map<Class<?>, Integer> getCustomSqlCodesMappings() {
+		Map<Class<?>, Integer> customSqlCodesMappings = super.getCustomSqlCodesMappings();
+		customSqlCodesMappings.put(OffsetDateTime.class, Types.TIMESTAMP);
+		customSqlCodesMappings.put(ZonedDateTime.class, Types.TIMESTAMP);
+		return customSqlCodesMappings;
+    }
 }