Prevent int overflow for large amounts of records

Author: blootsvoets

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

@@ -58,7 +58,7 @@ public class RadarHdfsRestructure {
     private static final Logger logger = LoggerFactory.getLogger(RadarHdfsRestructure.class);
 
     /** Number of offsets to process in a single task. */
-    private static final int BATCH_SIZE = 500_000;
+    private static final long BATCH_SIZE = 500_000;
 
     private final int numThreads;
     private final Configuration conf;
@@ -174,8 +174,8 @@ private void processPaths(List<TopicFileList> topicPaths, Accountant accountant)
                     String topic = paths.files.get(0).topic;
                     logger.info("Processing {} records for topic {}", size, topic);
                     executor.execute(() -> {
-                        int batchSize = (int)(BATCH_SIZE * ThreadLocalRandom.current().nextDouble(0.75, 1.25));
-                        int currentSize = 0;
+                        long batchSize = Math.round(BATCH_SIZE * ThreadLocalRandom.current().nextDouble(0.75, 1.25));
+                        long currentSize = 0;
                         try (FileCacheStore cache = fileStoreFactory.newFileCacheStore(accountant)) {
                             for (TopicFile file : paths.files) {
                                 try {
@@ -280,7 +280,7 @@ private static class TopicFileList {
         public TopicFileList(Stream<TopicFile> files) {
             this.files = files.collect(Collectors.toList());
             this.size = this.files.stream()
-                    .mapToInt(TopicFile::size)
+                    .mapToLong(TopicFile::size)
                     .sum();
         }
 
@@ -308,8 +308,8 @@ public String getTopic() {
             return topic;
         }
 
-        public int size() {
-            return 1 + (int) (range.getOffsetTo() - range.getOffsetFrom());
+        public long size() {
+            return 1 + range.getOffsetTo() - range.getOffsetFrom();
         }
     }
 }