Merge pull request #37 from RADAR-base/release-0.5.3

Release 0.5.3

Author: blootsvoets

README.md

@@ -8,7 +8,7 @@ Data streamed to HDFS using the [RADAR HDFS sink connector](https://github.com/R
 
 This package is available as docker image [`radarbase/radar-hdfs-restructure`](https://hub.docker.com/r/radarbase/radar-hdfs-restructure). The entrypoint of the image is the current application. So in all of the commands listed in usage, replace `radar-hdfs-restructure` with for example:
 ```shell
-docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-hdfs-restructure:0.5.2 -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-hdfs-restructure:0.5.3 -n hdfs-namenode -o /output /myTopic
 ```
 if your docker cluster is running in the `hadoop` network and your output directory should be `./output`.
 
@@ -23,7 +23,7 @@ This package requires at least Java JDK 8. Build the distribution with
 and install the package into `/usr/local` with for example
 ```shell
 sudo mkdir -p /usr/local
-sudo tar -xzf build/distributions/radar-hdfs-restructure-0.5.2.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-hdfs-restructure-0.5.3.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-hdfs-restructure` command should be available.

build.gradle

@@ -7,7 +7,7 @@ plugins {
 }
 
 group 'org.radarcns'
-version '0.5.2'
+version '0.5.3'
 mainClassName = 'org.radarcns.hdfs.Application'
 
 sourceCompatibility = '1.8'
@@ -179,5 +179,5 @@ bintray {
 }
 
 wrapper {
-    gradleVersion '4.9'
+    gradleVersion '4.10.2'
 }

gradle/wrapper/gradle-wrapper.jar

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.9-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.10.2-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

gradle/wrapper/gradle-wrapper.properties

@@ -19,11 +19,13 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Instant;
+import javax.annotation.Nonnull;
 import org.apache.avro.generic.GenericRecord;
 
 public class ObservationKeyPathFactory extends RecordPathFactory {
+    @Nonnull
     @Override
-    public Path getRelativePath(String topic, GenericRecord key, GenericRecord value, Instant time, int attempt) {
+    public Path getRelativePath(@Nonnull String topic, GenericRecord key, GenericRecord value, Instant time, int attempt) {
         String projectId = sanitizeId(key.get("projectId"), "unknown-project");
         String userId = sanitizeId(key.get("userId"), "unknown-user");
 
@@ -33,8 +35,9 @@ public Path getRelativePath(String topic, GenericRecord key, GenericRecord value
         return Paths.get(projectId, userId, topic, outputFileName);
     }
 
+    @Nonnull
     @Override
-    public String getCategory(GenericRecord key, GenericRecord value) {
+    public String getCategory(@Nonnull GenericRecord key, @Nonnull GenericRecord value) {
         return sanitizeId(key.get("sourceId"), "unknown-source");
     }
 }

src/main/java/org/radarcns/hdfs/ObservationKeyPathFactory.java

@@ -20,9 +20,16 @@
 
 import java.nio.file.Path;
 import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
 import java.util.regex.Pattern;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import org.apache.avro.Schema;
+import org.apache.avro.Schema.Field;
+import org.apache.avro.Schema.Type;
 import org.apache.avro.generic.GenericRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,7 +44,17 @@ public abstract class RecordPathFactory implements Plugin {
     private Path root;
     private String extension;
 
-    public RecordOrganization getRecordOrganization(String topic, GenericRecord record, int attempt) {
+    /**
+     * Get the organization of given record in given topic.
+     * @param topic Kafka topic name
+     * @param record record with possible key and value fields containing records
+     * @param attempt number of previous attempts to write given record. This increases if previous
+     *                paths already existed and are incompatible.
+     * @return organization of given record
+     */
+    @Nonnull
+    public RecordOrganization getRecordOrganization(@Nonnull String topic,
+            @Nonnull GenericRecord record, int attempt) {
         GenericRecord keyField = (GenericRecord) record.get("key");
         GenericRecord valueField = (GenericRecord) record.get("value");
 
@@ -54,8 +71,27 @@ public RecordOrganization getRecordOrganization(String topic, GenericRecord reco
         return new RecordOrganization(outputPath, category, time);
     }
 
-    public abstract Path getRelativePath(String topic, GenericRecord key, GenericRecord value, Instant time, int attempt);
+    /**
+     * Get the relative path corresponding to given record on given topic.
+     * @param topic Kafka topic name
+     * @param key record key
+     * @param value record value
+     * @param time time contained in the record
+     * @param attempt number of previous attempts to write given record. This increases if previous
+     *                paths already existed and are incompatible.
+     * @return relative path corresponding to given parameters.
+     */
+    @Nonnull
+    public abstract Path getRelativePath(@Nonnull String topic, @Nullable GenericRecord key,
+            @Nullable GenericRecord value, @Nullable Instant time, int attempt);
 
+    /**
+     * Get the category of a record, representing a partitioning for a given topic and user.
+     * @param key record key
+     * @param value record value
+     * @return category name.
+     */
+    @Nonnull
     public abstract String getCategory(GenericRecord key, GenericRecord value);
 
     public Path getRoot() {
@@ -82,48 +118,148 @@ public DateTimeFormatter getTimeBinFormat() {
         return HOURLY_TIME_BIN_FORMAT;
     }
 
-    public String getTimeBin(Instant time) {
+    @Nonnull
+    public String getTimeBin(@Nullable Instant time) {
         return time == null ? "unknown_date" : getTimeBinFormat().format(time);
     }
 
+    /**
+     * Organization of a record.
+     */
     public static class RecordOrganization {
         private final Path path;
         private final Instant time;
         private final String category;
 
-        public RecordOrganization(Path path, String category, Instant time) {
+        /**
+         * Organization of a record.
+         *
+         * @param path path that the record should be stored in.
+         * @param category category or partition that the record belongs to
+         * @param time time contained in the record, if any
+         */
+        public RecordOrganization(@Nonnull Path path, @Nonnull String category,
+                @Nullable Instant time) {
             this.path = path;
             this.time = time;
             this.category = category;
         }
 
+        @Nonnull
         public Path getPath() {
             return path;
         }
 
+        @Nullable
         public Instant getTime() {
             return time;
         }
 
+        @Nonnull
         public String getCategory() {
             return category;
         }
     }
 
-    public static Instant getDate(GenericRecord keyField, GenericRecord valueField) {
-        Schema.Field timeField = valueField.getSchema().getField("time");
-        if (timeField != null) {
-            double time = (Double) valueField.get(timeField.pos());
-            // Convert from millis to date and apply dateFormat
-            return Instant.ofEpochMilli((long) (time * 1000d));
+    /**
+     * Get the date contained in given records
+     * @param keyField key field of the record
+     * @param valueField value field of the record
+     * @return date contained in the values of either record, or {@code null} if not found or
+     *         it cannot be parsed.
+     */
+    @Nullable
+    public static Instant getDate(@Nullable GenericRecord keyField,
+            @Nullable GenericRecord valueField) {
+        Schema.Field timeField;
+
+        if (valueField != null) {
+            timeField = valueField.getSchema().getField("time");
+            if (timeField != null && timeField.schema().getType() == Type.DOUBLE) {
+                double time = (Double) valueField.get(timeField.pos());
+                // Convert from millis to date and apply dateFormat
+                return Instant.ofEpochMilli((long) (time * 1000d));
+            }
+        }
+
+        if (keyField != null) {
+            timeField = keyField.getSchema().getField("timeStart");
+
+            if (timeField != null && timeField.schema().getType() == Type.DOUBLE) {
+                double time = (Double) keyField.get(timeField.pos());
+                // Convert from millis to date and apply dateFormat
+                return Instant.ofEpochMilli((long) (time * 1000d));
+            }
+
+            // WindowedKey
+            timeField = keyField.getSchema().getField("start");
+            if (timeField != null && timeField.schema().getType() == Type.LONG) {
+                return Instant.ofEpochMilli((Long) keyField.get("start"));
+            }
+        }
+
+        if (valueField != null) {
+            Instant result = parseDateTime(valueField);
+            if (result != null) {
+                return result;
+            }
+            result = parseDate(valueField);
+            if (result != null) {
+                return result;
+            }
         }
 
-        // WindowedKey
-        timeField = keyField.getSchema().getField("start");
-        if (timeField == null) {
-            return null;
+        return null;
+    }
+
+    /**
+     * Parse the dateTime field of a record, if present.
+     *
+     * @param record record that may contain a dateTime field
+     * @return {@code Instant} representing the dateTime or {@code null} if the field cannot be
+     * found or parsed.
+     */
+    @Nullable
+    public static Instant parseDateTime(@Nonnull GenericRecord record) {
+        // dateTime
+        Field timeField = record.getSchema().getField("dateTime");
+        if (timeField != null && timeField.schema().getType() == Type.STRING) {
+            String dateTime = record.get(timeField.pos()).toString();
+            try {
+                if (dateTime.charAt(dateTime.length() - 1) == 'Z') {
+                    return Instant.parse(dateTime);
+                } else {
+                    return LocalDateTime.parse(dateTime).toInstant(UTC);
+                }
+            } catch (DateTimeParseException ex) {
+                // try next data type
+            }
         }
-        return Instant.ofEpochMilli((Long) keyField.get("start"));
+
+        return null;
+    }
+
+    /**
+     * Parse the date field of a record, if present.
+     *
+     * @param record record that may contain a date field
+     * @return {@code Instant} representing the start of given date or {@code null} if the field
+     * cannot be found or parsed.
+     */
+    @Nullable
+    public static Instant parseDate(@Nonnull GenericRecord record) {
+        // dateTime
+        Field timeField = record.getSchema().getField("date");
+        if (timeField != null && timeField.schema().getType() == Type.STRING) {
+            String date = record.get(timeField.pos()).toString();
+            try {
+                return LocalDate.parse(date).atStartOfDay(UTC).toInstant();
+            } catch (DateTimeParseException ex) {
+                // no other options
+            }
+        }
+
+        return null;
     }
 
     public static String sanitizeId(Object id, String defaultValue) {

src/main/java/org/radarcns/hdfs/RecordPathFactory.java

@@ -17,7 +17,7 @@
 package org.radarcns.hdfs.accounting;
 
 import java.util.Objects;
-import org.jetbrains.annotations.NotNull;
+import javax.annotation.Nonnull;
 
 public final class TopicPartition implements Comparable<TopicPartition> {
     public final String topic;
@@ -45,7 +45,7 @@ public int hashCode() {
     }
 
     @Override
-    public int compareTo(@NotNull TopicPartition o) {
+    public int compareTo(@Nonnull TopicPartition o) {
         int result = topic.compareTo(o.topic);
         if (result != 0) {
             return result;