Merge pull request #47 from RADAR-base/release-0.5.7

Release 0.5.7

Author: blootsvoets

.travis.yml

@@ -1,4 +1,6 @@
 language: java
+# Force use of dist where oraclejdk8 is supported
+dist: trusty
 jdk:
   - oraclejdk8
 sudo: false

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.6 -n hdfs-namenode -o /output /myTopic
+docker run --rm -t --network hadoop -v "$PWD/output:/output" radarbase/radar-hdfs-restructure:0.5.7 -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.6.tar.gz -C /usr/local --strip-components=1
+sudo tar -xzf build/distributions/radar-hdfs-restructure-0.5.7.tar.gz -C /usr/local --strip-components=1
 ```
 
 Now the `radar-hdfs-restructure` command should be available.
@@ -60,6 +60,8 @@ By default, files records are not deduplicated after writing. To enable this beh
 
 To set the output user ID and group ID, specify the `-p local-uid=123` and `-p local-gid=12` properties.
 
+To run the output generator as a service that will regularly poll the HDFS directory, add the `--service` flag and optionally the `--interval` flag to adjust the polling interval.
+
 ## Extending the connector
 
 To implement alternative storage paths, storage drivers or storage formats, put your custom JAR in

build.gradle

@@ -7,7 +7,7 @@ plugins {
 }
 
 group 'org.radarcns'
-version '0.5.6'
+version '0.5.7'
 mainClassName = 'org.radarcns.hdfs.Application'
 
 sourceCompatibility = '1.8'
@@ -190,5 +190,5 @@ bintray {
 }
 
 wrapper {
-    gradleVersion '5.2.1'
+    gradleVersion '5.4.1'
 }

gradle/wrapper/gradle-wrapper.jar

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

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,21 @@
 #!/usr/bin/env sh
 
+#
+# Copyright 2015 the original author or authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
 ##############################################################################
 ##
 ##  Gradle start up script for UN*X
@@ -28,7 +44,7 @@ APP_NAME="Gradle"
 APP_BASE_NAME=`basename "$0"`
 
 # Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-DEFAULT_JVM_OPTS='"-Xmx64m"'
+DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"

gradlew

@@ -1,3 +1,19 @@
+@rem
+@rem Copyright 2015 the original author or authors.
+@rem
+@rem Licensed under the Apache License, Version 2.0 (the "License");
+@rem you may not use this file except in compliance with the License.
+@rem You may obtain a copy of the License at
+@rem
+@rem      http://www.apache.org/licenses/LICENSE-2.0
+@rem
+@rem Unless required by applicable law or agreed to in writing, software
+@rem distributed under the License is distributed on an "AS IS" BASIS,
+@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+@rem See the License for the specific language governing permissions and
+@rem limitations under the License.
+@rem
+
 @if "%DEBUG%" == "" @echo off
 @rem ##########################################################################
 @rem
@@ -14,7 +30,7 @@ set APP_BASE_NAME=%~n0
 set APP_HOME=%DIRNAME%
 
 @rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script.
-set DEFAULT_JVM_OPTS="-Xmx64m"
+set DEFAULT_JVM_OPTS="-Xmx64m" "-Xms64m"
 
 @rem Find java.exe
 if defined JAVA_HOME goto findJavaFromJavaHome

gradlew.bat

@@ -31,6 +31,9 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import org.radarcns.hdfs.accounting.Accountant;
 import org.radarcns.hdfs.config.HdfsSettings;
 import org.radarcns.hdfs.config.RestructureSettings;
@@ -58,10 +61,15 @@ public class Application implements FileStoreFactory {
     private final RecordPathFactory pathFactory;
     private final List<String> inputPaths;
     private final RestructureSettings settings;
+    private final int pollInterval;
+    private final boolean isService;
+    private RadarHdfsRestructure hdfsReader;
 
     private Application(Builder builder) {
         this.storageDriver = builder.storageDriver;
         this.settings = builder.settings;
+        this.isService = builder.asService;
+        this.pollInterval = builder.pollInterval;
 
         converterFactory = builder.formatFactory.get(settings.getFormat());
         compression = builder.compressionFactory.get(settings.getCompression());
@@ -131,6 +139,8 @@ public static void main(String [] args) {
                     .storageDriver(commandLineArgs.storageDriver)
                     .properties(commandLineArgs.properties)
                     .inputPaths(commandLineArgs.inputPaths)
+                    .asService(commandLineArgs.asService)
+                    .pollInterval(commandLineArgs.pollInterval)
                     .build();
         } catch (IllegalArgumentException ex) {
             logger.error("HDFS High availability name node configuration is incomplete."
@@ -193,10 +203,41 @@ public void start() {
         System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism",
                 String.valueOf(settings.getNumThreads() - 1));
 
-        Instant timeStart = Instant.now();
-        RadarHdfsRestructure hdfsReader = new RadarHdfsRestructure(this);
         try {
             Files.createDirectories(settings.getTempDir());
+        } catch (IOException ex) {
+            logger.error("Failed to create temporary directory");
+            return;
+        }
+
+        ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+        executorService.execute(() -> hdfsReader = new RadarHdfsRestructure(this));
+
+        if (isService) {
+            logger.info("Press Ctrl+C to exit...");
+            executorService.scheduleAtFixedRate(this::runRestructure,
+                    pollInterval / 4, pollInterval, TimeUnit.SECONDS);
+        } else {
+            executorService.execute(this::runRestructure);
+        }
+
+        try {
+            Thread.sleep(Long.MAX_VALUE);
+        } catch (InterruptedException e) {
+            logger.info("Interrupted, shutting down...");
+            executorService.shutdownNow();
+            try {
+                executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
+                Thread.currentThread().interrupt();
+            } catch (InterruptedException ex) {
+                logger.info("Interrupted again...");
+            }
+        }
+    }
+
+    private void runRestructure() {
+        Instant timeStart = Instant.now();
+        try {
             for (String input : inputPaths) {
                 logger.info("In:  {}", input);
                 logger.info("Out: {}", pathFactory.getRoot());
@@ -220,6 +261,8 @@ public static class Builder {
         private FormatFactory formatFactory;
         private Map<String, String> properties = new HashMap<>();
         private List<String> inputPaths;
+        private boolean asService;
+        private int pollInterval;
 
         public Builder(RestructureSettings settings) {
             this.settings = settings;
@@ -256,6 +299,16 @@ public Builder properties(Map<String, String> props) {
             return this;
         }
 
+        public Builder asService(boolean asService) {
+            this.asService = asService;
+            return this;
+        }
+
+        public Builder pollInterval(int pollInterval) {
+            this.pollInterval = pollInterval;
+            return this;
+        }
+
         public Application build() throws IOException {
             pathFactory = nonNullOrDefault(pathFactory, ObservationKeyPathFactory::new);
             pathFactory.init(properties);

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

@@ -96,6 +96,12 @@ public class CommandLineArgs {
     @Parameter(names = {"--exclude-topic"}, description = "Topic to exclude when processing the records. Can be supplied more than once to exclude multiple topics.")
     public List<String> excludeTopics = new ArrayList<>();
 
+    @Parameter(names = {"-S", "--service"}, description = "Run the output generation as a service")
+    public boolean asService = false;
+
+    @Parameter(names = {"-i", "--interval"}, description = "Polling interval when running as a service (seconds)")
+    public int pollInterval = 3600;
+
     public static <T> T nonNullOrDefault(T value, Supplier<T> defaultValue) {
         return value != null ? value : defaultValue.get();
     }