Updated version and added Javadoc

Author: blootsvoets

build.gradle

@@ -7,7 +7,7 @@ plugins {
 }
 
 group 'org.radarcns'
-version '0.5.3-SNAPSHOT'
+version '0.5.3'
 mainClassName = 'org.radarcns.hdfs.Application'
 
 sourceCompatibility = '1.8'

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

@@ -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/RecordPathFactory.java

@@ -25,7 +25,10 @@
 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;
@@ -41,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");
 
@@ -58,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() {
@@ -86,80 +118,147 @@ 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 && 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));
+    /**
+     * 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));
+            }
         }
-        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));
+        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"));
+            }
         }
 
-        // 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;
+            }
         }
 
+        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
-        timeField = valueField.getSchema().getField("dateTime");
+        Field timeField = record.getSchema().getField("dateTime");
         if (timeField != null && timeField.schema().getType() == Type.STRING) {
-            String dateTime = valueField.get(timeField.pos()).toString();
+            String dateTime = record.get(timeField.pos()).toString();
             try {
-                return Instant.parse(dateTime);
+                if (dateTime.charAt(dateTime.length() - 1) == 'Z') {
+                    return Instant.parse(dateTime);
+                } else {
+                    return LocalDateTime.parse(dateTime).toInstant(UTC);
+                }
             } catch (DateTimeParseException ex) {
-                // try local date
-            }
-            try {
-                return LocalDateTime.parse(dateTime).toInstant(UTC);
-            } catch (DateTimeParseException ex) {
-                // no other options
+                // try next data type
             }
         }
 
-        timeField = valueField.getSchema().getField("date");
+        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 = valueField.get(timeField.pos()).toString();
+            String date = record.get(timeField.pos()).toString();
             try {
                 return LocalDate.parse(date).atStartOfDay(UTC).toInstant();
             } catch (DateTimeParseException ex) {
                 // no other options
             }
         }
+
         return null;
     }
 

src/main/java/org/radarcns/hdfs/accounting/TopicPartition.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;