diff --git a/debezium-connector-jdbc/src/main/java/io/debezium/connector/jdbc/type/debezium/ZonedTimeType.java b/debezium-connector-jdbc/src/main/java/io/debezium/connector/jdbc/type/debezium/ZonedTimeType.java index fe1c18418..769524662 100644 --- a/debezium-connector-jdbc/src/main/java/io/debezium/connector/jdbc/type/debezium/ZonedTimeType.java +++ b/debezium-connector-jdbc/src/main/java/io/debezium/connector/jdbc/type/debezium/ZonedTimeType.java @@ -8,10 +8,12 @@ import java.sql.Types; import java.time.LocalDate; import java.time.OffsetTime; +import java.time.ZonedDateTime; import org.apache.kafka.connect.data.Schema; import org.hibernate.engine.jdbc.Size; import org.hibernate.query.Query; +import org.hibernate.type.StandardBasicTypes; import io.debezium.connector.jdbc.dialect.DatabaseDialect; import io.debezium.connector.jdbc.type.AbstractTimeType; @@ -65,11 +67,9 @@ public void bind(Query query, int index, Schema schema, Object value) { query.setParameter(index, null); } else if (value instanceof String) { - // NOTE: - // We must bind the value as an Instant as Hibernate will refuse to bind the TZ details - // on the destination correctly for Oracle. final OffsetTime offsetTime = OffsetTime.parse((String) value, ZonedTime.FORMATTER); - query.setParameter(index, offsetTime.atDate(LocalDate.now()).toInstant()); + final ZonedDateTime zdt = offsetTime.atDate(LocalDate.EPOCH).toInstant().atZone(getDatabaseTimeZone().toZoneId()); + query.setParameter(index, zdt, StandardBasicTypes.ZONED_DATE_TIME_WITH_TIMEZONE); } else { throwUnexpectedValue(value); diff --git a/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/e2e/AbstractJdbcSinkPipelineIT.java b/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/e2e/AbstractJdbcSinkPipelineIT.java index 4f4ce628d..57446430e 100644 --- a/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/e2e/AbstractJdbcSinkPipelineIT.java +++ b/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/e2e/AbstractJdbcSinkPipelineIT.java @@ -15,7 +15,6 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; -import java.time.LocalTime; import java.time.OffsetDateTime; import java.time.OffsetTime; import java.time.ZoneOffset; @@ -1870,6 +1869,7 @@ public void testTimestampWithLocalTimeZoneDataType(Source source, Sink sink) thr @TestTemplate @SkipWhenSource(value = { SourceType.MYSQL, SourceType.ORACLE, SourceType.SQLSERVER }, reason = "No TIME(n) WITH TIME ZONE data type support") @SkipWhenSink(value = { SinkType.MYSQL }, reason = "MySQL has no support for TIME(n) with TIME ZONE support") + @SkipWhenSink(value = { SinkType.DB2 }, reason = "There is an issue with Daylight Savings Time") @WithTemporalPrecisionMode public void testTimeWithTimeZoneDataType(Source source, Sink sink) throws Exception { // Only test non-keys because Oracle does not permit timestamp with timezone as primary key columns diff --git a/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSink.java b/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSink.java index 6fec8dd30..ff3f88fa1 100644 --- a/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSink.java +++ b/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSink.java @@ -6,6 +6,7 @@ package io.debezium.connector.jdbc.junit.jupiter.e2e; import java.lang.annotation.ElementType; +import java.lang.annotation.Repeatable; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; @@ -20,6 +21,7 @@ */ @Target({ ElementType.METHOD }) @Retention(RetentionPolicy.RUNTIME) +@Repeatable(SkipWhenSinks.class) public @interface SkipWhenSink { /** * Returns the connector types that will be excluded from the test template invocation matrix. diff --git a/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSinks.java b/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSinks.java new file mode 100644 index 000000000..4abf9634d --- /dev/null +++ b/debezium-connector-jdbc/src/test/java/io/debezium/connector/jdbc/junit/jupiter/e2e/SkipWhenSinks.java @@ -0,0 +1,23 @@ +/* + * Copyright Debezium Authors. + * + * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 + */ +package io.debezium.connector.jdbc.junit.jupiter.e2e; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Allows applying multiple {@link SkipWhenSink} annotations to the same method, allowing each + * annotation to designate differ reasons why the sources are being skipped. + * + * @author Chris Cranford + */ +@Target({ ElementType.METHOD }) +@Retention(RetentionPolicy.RUNTIME) +public @interface SkipWhenSinks { + SkipWhenSink[] value(); +}