This is an automated email from the ASF dual-hosted git repository. zehnder pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 19430b78bdf6aa4508d28b7dda84cf8562f75e17 Author: Philipp Zehnder <[email protected]> AuthorDate: Tue Dec 7 09:49:00 2021 +0100 [hotfix] Fix timestamp format for data lake csv export --- .../dataexplorer/DataLakeManagementV4.java | 56 ++++++++++++---------- 1 file changed, 31 insertions(+), 25 deletions(-) diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java index 3ae4241..f9853fb 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataLakeManagementV4.java @@ -40,6 +40,11 @@ import java.io.IOException; import java.io.OutputStream; import java.text.ParseException; import java.text.SimpleDateFormat; +import java.time.*; +import java.time.format.DateTimeFormatter; +import java.time.temporal.ChronoField; +import java.time.temporal.ChronoUnit; +import java.time.temporal.TemporalAccessor; import java.util.List; import java.util.Map; import java.util.TimeZone; @@ -48,6 +53,10 @@ import static org.apache.streampipes.dataexplorer.v4.SupportedDataLakeQueryParam public class DataLakeManagementV4 { + private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern( + "yyyy-MM-dd'T'HH:mm:ss[.SSS]'Z'") + .withZone(ZoneId.of("UTC")); + public List<DataLakeMeasure> getAllMeasurements() { return DataExplorerUtils.getInfos(); } @@ -64,8 +73,6 @@ public class DataLakeManagementV4 { if (!params.has(QP_LIMIT)) { params.update(QP_LIMIT, 500000); } - SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"); - formatter.setTimeZone(TimeZone.getTimeZone("UTC")); SpQueryResult dataResult; //JSON @@ -100,11 +107,7 @@ public class DataLakeManagementV4 { } isFirstElementInRow = false; if (i1 == 0) { - try { - element = formatter.parse(element.toString()).getTime(); - } catch (ParseException e) { - element = element.toString(); - } + element = parseTime(element.toString()); } //produce json e.g. "name": "Pipes" or "load": 42 outputStream.write(toBytes("\"" + dataResult.getHeaders().get(i1) + "\": " @@ -157,11 +160,7 @@ public class DataLakeManagementV4 { } isFirstInRow = false; if (i1 == 0) { - try { - element = formatter.parse(element.toString()).getTime(); - } catch (ParseException e) { - element = element.toString(); - } + element = parseTime(element.toString()); } if (element == null) { outputStream.write(toBytes("")); @@ -195,10 +194,7 @@ public class DataLakeManagementV4 { if (measure.getMeasureName().equals(measurementID)) { QueryResult queryResult = new DeleteDataQuery(new DataLakeMeasure(measurementID, null)).executeQuery(); - if (queryResult.hasError() || queryResult.getResults().get(0).getError() != null) { - return false; - } - return true; + return !queryResult.hasError() && queryResult.getResults().get(0).getError() == null; } } return false; @@ -232,11 +228,11 @@ public class DataLakeManagementV4 { Integer batchSize = config.getBatchSize(); Integer flushDuration = config.getFlushDuration(); - /** - * TODO: - * - Implementation of parameter update for batchSize and flushDuration - * - Updating multiple retention policies - */ + // + // TODO: + // - Implementation of parameter update for batchSize and flushDuration + // - Updating multiple retention policies + // String operation = "CREATE"; if (existingRetentionPolicies.size() > 1) { @@ -247,10 +243,10 @@ public class DataLakeManagementV4 { } public List<DataLakeRetentionPolicy> getAllExistingRetentionPolicies() { - /** - * TODO: - * - Implementation of parameter return for batchSize and flushDuration - */ + // + // TODO: + // - Implementation of parameter return for batchSize and flushDuration + // return new ShowRetentionPolicyQuery(RetentionPolicyQueryParams.from("", "0s")).executeQuery(); } @@ -278,4 +274,14 @@ public class DataLakeManagementV4 { private byte[] toBytes(String value) { return value.getBytes(); } + + private static Long parseTime(String v) { + TemporalAccessor temporalAccessor = formatter.parseBest(v, + ZonedDateTime::from, + LocalDateTime::from, + LocalDate::from); + + Instant instant = Instant.from(temporalAccessor); + return Instant.EPOCH.until(instant, ChronoUnit.MILLIS); + } }
