Skip to content

[FLINK-26810][connectors/elasticsearch] Use local timezone for TIMESTAMP_WITH_LOCAL_TIMEZONE fields in dynamic index #19302

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fapaul merged 1 commit into from
Apr 7, 2022
Merged

[FLINK-26810][connectors/elasticsearch] Use local timezone for TIMESTAMP_WITH_LOCAL_TIMEZONE fields in dynamic index #19302

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions ...e/src/main/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
Expand Down Expand Up @@ -133,7 +132,8 @@ public String generate(RowData row) {
final String dateTimeFormat =
indexHelper.extractDateFormat(index, indexFieldLogicalTypeRoot);
DynamicFormatter formatFunction =
createFormatFunction(indexFieldType, indexFieldLogicalTypeRoot);
createFormatFunction(
indexFieldType, indexFieldLogicalTypeRoot, localTimeZoneId);

return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
@Override
Expand Down Expand Up @@ -163,7 +163,9 @@ public String generate(RowData row) {
}

private static DynamicFormatter createFormatFunction(
LogicalType indexFieldType, LogicalTypeRoot indexFieldLogicalTypeRoot) {
LogicalType indexFieldType,
LogicalTypeRoot indexFieldLogicalTypeRoot,
ZoneId localTimeZoneId) {
switch (indexFieldLogicalTypeRoot) {
case DATE:
return (value, dateTimeFormatter) -> {
Expand All @@ -186,7 +188,7 @@ private static DynamicFormatter createFormatFunction(
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return (value, dateTimeFormatter) -> {
TimestampData indexField = (TimestampData) value;
return indexField.toInstant().atZone(ZoneOffset.UTC).format(dateTimeFormatter);
return indexField.toInstant().atZone(localTimeZoneId).format(dateTimeFormatter);
};
default:
throw new TableException(
Expand Down
10 changes: 6 additions & 4 deletions ...java/org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
Expand Down Expand Up @@ -130,7 +129,8 @@ public String generate(RowData row) {
final String dateTimeFormat =
indexHelper.extractDateFormat(index, indexFieldLogicalTypeRoot);
DynamicFormatter formatFunction =
createFormatFunction(indexFieldType, indexFieldLogicalTypeRoot);
createFormatFunction(
indexFieldType, indexFieldLogicalTypeRoot, localTimeZoneId);

return new AbstractTimeIndexGenerator(index, dateTimeFormat) {
@Override
Expand Down Expand Up @@ -160,7 +160,9 @@ public String generate(RowData row) {
}

private static DynamicFormatter createFormatFunction(
LogicalType indexFieldType, LogicalTypeRoot indexFieldLogicalTypeRoot) {
LogicalType indexFieldType,
LogicalTypeRoot indexFieldLogicalTypeRoot,
ZoneId localTimeZoneId) {
switch (indexFieldLogicalTypeRoot) {
case DATE:
return (value, dateTimeFormatter) -> {
Expand All @@ -183,7 +185,7 @@ private static DynamicFormatter createFormatFunction(
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
return (value, dateTimeFormatter) -> {
TimestampData indexField = (TimestampData) value;
return indexField.toInstant().atZone(ZoneOffset.UTC).format(dateTimeFormatter);
return indexField.toInstant().atZone(localTimeZoneId).format(dateTimeFormatter);
};
default:
throw new TableException(
Expand Down
44 changes: 44 additions & 0 deletions ...base/src/test/java/org/apache/flink/connector/elasticsearch/table/IndexGeneratorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,11 +35,14 @@
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.UnsupportedTemporalTypeException;
import java.util.Arrays;
import java.util.List;

import static org.junit.jupiter.api.Assumptions.assumingThat;

/** Suite tests for {@link IndexGenerator}. */
public class IndexGeneratorTest {

Expand All @@ -54,6 +57,7 @@ public class IndexGeneratorTest {
"local_datetime",
"local_date",
"local_time",
"local_timestamp",
"note",
"status");

Expand All @@ -68,6 +72,7 @@ public class IndexGeneratorTest {
DataTypes.TIMESTAMP().bridgedTo(LocalDateTime.class),
DataTypes.DATE().bridgedTo(LocalDate.class),
DataTypes.TIME().bridgedTo(LocalTime.class),
DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(),
DataTypes.STRING(),
DataTypes.BOOLEAN());

Expand All @@ -86,6 +91,10 @@ public class IndexGeneratorTest {
LocalDateTime.of(2020, 3, 18, 12, 12, 14, 1000)),
(int) LocalDate.of(2020, 3, 18).toEpochDay(),
(int) (LocalTime.of(12, 13, 14, 2000).toNanoOfDay() / 1_000_000L),
TimestampData.fromInstant(
LocalDateTime.of(2020, 3, 18, 3, 12, 14, 1000)
.atZone(ZoneId.of("Asia/Shanghai"))
.toInstant()),
"test1",
true),
GenericRowData.of(
Expand All @@ -101,9 +110,44 @@ public class IndexGeneratorTest {
LocalDateTime.of(2020, 3, 19, 12, 22, 14, 1000)),
(int) LocalDate.of(2020, 3, 19).toEpochDay(),
(int) (LocalTime.of(12, 13, 14, 2000).toNanoOfDay() / 1_000_000L),
TimestampData.fromInstant(
LocalDateTime.of(2020, 3, 19, 20, 22, 14, 1000)
.atZone(ZoneId.of("America/Los_Angeles"))
.toInstant()),
"test2",
false));

@Test
public void testDynamicIndexFromTimestampTzUTC() {
assumingThat(
Copy link
Contributor

@matriv matriv Apr 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nice!

ZoneId.systemDefault().equals(ZoneId.of("UTC")),
() -> {
IndexGenerator indexGenerator =
IndexGeneratorFactory.createIndexGenerator(
"{local_timestamp|yyyy_MM_dd_HH-ss}_index",
fieldNames,
dataTypes);
indexGenerator.open();
Assertions.assertEquals(
"2020_03_17_19-14_index", indexGenerator.generate(rows.get(0)));
Assertions.assertEquals(
"2020_03_20_03-14_index", indexGenerator.generate(rows.get(1)));
});
}

@Test
public void testDynamicIndexFromTimestampTzWithSpecificTimezone() {
IndexGenerator indexGenerator =
IndexGeneratorFactory.createIndexGenerator(
Copy link
Contributor

@matriv matriv Apr 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also here you're using the
createIndexGenerator(String index, List<String> fieldNames, List<DataType> dataTypes)
instead of

public static IndexGenerator createIndexGenerator(
            String index,
            List<String> fieldNames,
            List<DataType> dataTypes,
            ZoneId localTimeZoneId) {

were you can pass a custom zone, or also the UTC and be independent of the CI's system default for example.

"{local_timestamp|yyyy_MM_dd_HH-ss}_index",
fieldNames,
dataTypes,
ZoneId.of("Europe/Berlin"));
indexGenerator.open();
Assertions.assertEquals("2020_03_17_20-14_index", indexGenerator.generate(rows.get(0)));
Assertions.assertEquals("2020_03_20_04-14_index", indexGenerator.generate(rows.get(1)));
}

@Test
public void testDynamicIndexFromTimestamp() {
IndexGenerator indexGenerator =
Expand Down
35 changes: 29 additions & 6 deletions .../org/apache/flink/streaming/connectors/elasticsearch/table/IndexGeneratorFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,16 +32,19 @@
import org.junit.Test;

import java.sql.Timestamp;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.UnsupportedTemporalTypeException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assume.assumeThat;

/** Tests for {@link IndexGeneratorFactory}. */
public class IndexGeneratorFactoryTest extends TestLogger {

Expand Down Expand Up @@ -71,7 +74,10 @@ public void prepareData() {
(int) LocalDate.parse("2020-03-18").toEpochDay(),
(int) (LocalTime.parse("12:12:14").toNanoOfDay() / 1_000_000L),
TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-03-18T12:12:14")),
TimestampData.fromInstant(Instant.parse("2020-03-18T12:12:14Z")),
TimestampData.fromInstant(
LocalDateTime.of(2020, 3, 18, 3, 12, 14, 1000)
.atZone(ZoneId.of("Asia/Shanghai"))
.toInstant()),
true));
rows.add(
GenericRowData.of(
Expand All @@ -81,7 +87,10 @@ public void prepareData() {
(int) LocalDate.parse("2020-03-19").toEpochDay(),
(int) (LocalTime.parse("12:22:21").toNanoOfDay() / 1_000_000L),
TimestampData.fromLocalDateTime(LocalDateTime.parse("2020-03-19T12:22:14")),
TimestampData.fromInstant(Instant.parse("2020-03-19T12:12:14Z")),
TimestampData.fromInstant(
LocalDateTime.of(2020, 3, 19, 20, 22, 14, 1000)
.atZone(ZoneId.of("America/Los_Angeles"))
.toInstant()),
false));
}

Expand Down Expand Up @@ -193,12 +202,26 @@ public void testDynamicIndexFromSystemTime() {
}

@Test
public void testDynamicIndexDefaultFormatTimestampWithLocalTimeZone() {
public void testDynamicIndexDefaultFormatTimestampWithLocalTimeZoneUTC() {
assumeThat(ZoneId.systemDefault(), is(ZoneId.of("UTC")));
Copy link
Contributor

@matriv matriv Apr 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not use the same scheme with assumingThat as here: https://github.com/apache/flink/pull/19302/files#diff-718dd4539a392ce4b768bdfcabeca730bcdb4ffa4d09f691fd14c09002a84b95R124 ?

Copy link
Contributor Author

@alpreu alpreu Apr 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The tests under ...streaming.connectors.elasticsearch (for the old SinkFunction based connector) are still using Junit 4. I could port them but not sure if we should do that with tests that we want to remove soon anyway

Copy link
Contributor

@matriv matriv Apr 5, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, then it's fine as is, no need to change.


IndexGenerator indexGenerator =
IndexGeneratorFactory.createIndexGenerator("my-index-{local_timestamp|}", schema);
indexGenerator.open();
Assert.assertEquals("my-index-2020_03_18_12_12_14Z", indexGenerator.generate(rows.get(0)));
Assert.assertEquals("my-index-2020_03_19_12_12_14Z", indexGenerator.generate(rows.get(1)));
Assert.assertEquals("my-index-2020_03_17_19_12_14Z", indexGenerator.generate(rows.get(0)));
Assert.assertEquals("my-index-2020_03_20_03_22_14Z", indexGenerator.generate(rows.get(1)));
}

@Test
public void testDynamicIndexDefaultFormatTimestampWithLocalTimeZoneWithSpecificTimeZone() {
IndexGenerator indexGenerator =
IndexGeneratorFactory.createIndexGenerator(
"my-index-{local_timestamp|}", schema, ZoneId.of("Europe/Berlin"));
indexGenerator.open();
Assert.assertEquals(
"my-index-2020_03_17_20_12_14+01", indexGenerator.generate(rows.get(0)));
Assert.assertEquals(
"my-index-2020_03_20_04_22_14+01", indexGenerator.generate(rows.get(1)));
}

@Test
Expand Down