alexey.rassokhin/tet123
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

DBZ-1164 Misc. improvements;

Browse Source 
* Moving stringToDuration() for the sake of re-use and testability
* Making pattern and durations constants
This commit is contained in:
Gunnar Morling 2019-05-21 09:29:33 +02:00
parent 9b52ac86e8
commit 4751b266f3
3 changed files with 66 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresValueConverter.java
View File

@ -13,13 +13,13 @@
import java.nio.charset.Charset;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.time.OffsetTime;
import java.time.ZoneOffset;
import java.time.Duration;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
@ -30,11 +30,8 @@
import java.util.Map.Entry;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import io.debezium.util.Strings;
import org.apache.kafka.connect.data.Decimal;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
@ -65,6 +62,7 @@
import io.debezium.time.ZonedTime;
import io.debezium.time.ZonedTimestamp;
import io.debezium.util.NumberConversions;
import io.debezium.util.Strings;
/**
* A provider of {@link ValueConverter}s and {@link SchemaBuilder}s for various Postgres specific column types.
@ -107,6 +105,9 @@ public class PostgresValueConverter extends JdbcValueConverters {
.appendPattern("[XXX][XX][X]")
.toFormatter();
private static final Duration ONE_DAY = Duration.ofDays(1);
private static final long NANO_SECONDS_PER_DAY = TimeUnit.DAYS.toNanos(1);
/**
* {@code true} if fields of data type not know should be handle as opaque binary;
* {@code false} if they should be omitted
@ -649,33 +650,32 @@ else if (data instanceof String) {
}
private Object convertTwentyFourHourTime(Column column, Field fieldDefn, Object data) {
final long nanosecondsPerDay = TimeUnit.DAYS.toNanos(1);
long twentyFourHour = nanosecondsPerDay;
long twentyFourHour = NANO_SECONDS_PER_DAY;
if (adaptiveTimeMicrosecondsPrecisionMode) {
twentyFourHour = nanosecondsPerDay / 1_000;
twentyFourHour = NANO_SECONDS_PER_DAY / 1_000;
}
if (adaptiveTimePrecisionMode) {
if (getTimePrecision(column) <= 3) {
twentyFourHour = nanosecondsPerDay / 1_000_000;
twentyFourHour = NANO_SECONDS_PER_DAY / 1_000_000;
}
if (getTimePrecision(column) <= 6) {
twentyFourHour = nanosecondsPerDay / 1_000;
twentyFourHour = NANO_SECONDS_PER_DAY / 1_000;
}
}
// during streaming
if (data instanceof Long) {
if ((Long) data == nanosecondsPerDay) {
if ((Long) data == NANO_SECONDS_PER_DAY) {
return twentyFourHour;
}
return super.converter(column, fieldDefn).convert(data);
}
// during snapshotting
else if (data instanceof String) {
Duration d = stringToDuration((String) data);
Duration d = Strings.asDuration((String) data);
if (d.toNanos() == nanosecondsPerDay) {
if (d.equals(ONE_DAY)) {
return twentyFourHour;
}
return super.converter(column, fieldDefn).convert(d);
@ -684,37 +684,6 @@ else if (data instanceof String) {
return super.converter(column, fieldDefn).convert(data);
}
private static Duration stringToDuration(String timeString) {
final Pattern timeFieldPattern = Pattern.compile("([0-9]*):([0-9]*):([0-9]*)(\\.([0-9]*))?");
Matcher matcher = timeFieldPattern.matcher(timeString);
if (!matcher.matches()) {
throw new RuntimeException("Unexpected format for TIME column: " + timeString);
}
long hours = Long.parseLong(matcher.group(1));
long minutes = Long.parseLong(matcher.group(2));
long seconds = Long.parseLong(matcher.group(3));
long nanoSeconds = 0;
String microSecondsString = matcher.group(5);
if (microSecondsString != null) {
nanoSeconds = Long.parseLong(Strings.justifyLeft(microSecondsString, 9, '0'));
}
if (hours >= 0) {
return Duration.ofHours(hours)
.plusMinutes(minutes)
.plusSeconds(seconds)
.plusNanos(nanoSeconds);
}
else {
return Duration.ofHours(hours)
.minusMinutes(minutes)
.minusSeconds(seconds)
.minusNanos(nanoSeconds);
}
}
private static LocalDateTime nanosToLocalDateTimeUTC(long epocNanos) {
// the pg plugin stores date/time info as microseconds since epoch
BigInteger epochMicrosBigInt = BigInteger.valueOf(epocNanos);

42
debezium-core/src/main/java/io/debezium/util/Strings.java
View File

@ -10,6 +10,7 @@
import java.math.BigDecimal;
import java.math.BigInteger;
import java.text.DecimalFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@ -22,6 +23,7 @@
import java.util.UUID;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
@ -41,6 +43,8 @@
@ThreadSafe
public final class Strings {
private static final Pattern TIME_PATTERN = Pattern.compile("([0-9]*):([0-9]*):([0-9]*)(\\.([0-9]*))?");
/**
* Generate the set of values that are included in the list.
*
@ -682,6 +686,44 @@ public static boolean asBoolean(String value, boolean defaultValue) {
return defaultValue;
}
/**
* Converts the given string (in the format 00:00:00(.0*)) into a {@link Duration}.
*
* @return the given value as {@code Duration} or {@code null} if {@code null} was passed.
*/
public static Duration asDuration(String timeString) {
if (timeString == null) {
return null;
}
Matcher matcher = TIME_PATTERN.matcher(timeString);
if (!matcher.matches()) {
throw new RuntimeException("Unexpected format for TIME column: " + timeString);
}
long hours = Long.parseLong(matcher.group(1));
long minutes = Long.parseLong(matcher.group(2));
long seconds = Long.parseLong(matcher.group(3));
long nanoSeconds = 0;
String microSecondsString = matcher.group(5);
if (microSecondsString != null) {
nanoSeconds = Long.parseLong(Strings.justifyLeft(microSecondsString, 9, '0'));
}
if (hours >= 0) {
return Duration.ofHours(hours)
.plusMinutes(minutes)
.plusSeconds(seconds)
.plusNanos(nanoSeconds);
}
else {
return Duration.ofHours(hours)
.minusMinutes(minutes)
.minusSeconds(seconds)
.minusNanos(nanoSeconds);
}
}
/**
* For the given duration in milliseconds, obtain a readable representation of the form {@code HHH:MM:SS.mmm}, where
* <dl>

12
debezium-core/src/test/java/io/debezium/util/StringsTest.java
View File

@ -11,6 +11,7 @@
import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@ -20,6 +21,7 @@
import org.junit.Test;
import io.debezium.doc.FixFor;
import io.debezium.text.ParsingException;
/**
@ -333,6 +335,16 @@ public void regexSplitWrongEscape() {
Strings.setOfRegex("a,b\\,c\\");
}
@Test
@FixFor("DBZ-1164")
public void asDurationShouldConvertValue() {
assertThat(Strings.asDuration(null)).isNull();
assertThat(Strings.asDuration("24:00:00")).isEqualTo(Duration.parse("PT24H"));
assertThat(Strings.asDuration("18:02:54.123")).isEqualTo(Duration.parse("PT18H2M54.123S"));
assertThat(Strings.asDuration("18:02:54.123456789")).isEqualTo(Duration.parse("PT18H2M54.123456789S"));
assertThat(Strings.asDuration("24:00:01")).isEqualTo(Duration.parse("PT24H1S"));
}
protected void assertReplacement(String before, Map<String, String> replacements, String after) {
String result = Strings.replaceVariables(before, replacements::get);
assertThat(result).isEqualTo(after);