retryPolicy to dropTempTable and add some comments to the compressBlock

(cherry picked from commit b2b2335)
hawkular · Jul 4, 2017 · 492d462 · 492d462
1 parent d17ca72
commit 492d462
Showing 1 changed file with 9 additions and 7 deletions.
diff --git a/...rics-core-service/src/main/java/org/hawkular/metrics/core/service/MetricsServiceImpl.java b/...rics-core-service/src/main/java/org/hawkular/metrics/core/service/MetricsServiceImpl.java
@@ -750,15 +750,20 @@ public Completable compressBlock(long startTimeSlice, int pageSize, int maxConcu
         return Completable.fromObservable(
                 dataAccess.findAllDataFromBucket(startTimeSlice, pageSize, maxConcurrency)
                         .flatMap(rows -> rows
+                                // Each time the tokenrange changes inside the query, create new window, publish allows
+                                // reuse of the observable in two distinct processing phases
                                 .publish(p -> p.window(
                                         p.map(Row::getPartitionKeyToken)
                                                 .distinctUntilChanged()))
+                                // ConcatMap so we don't mess the order as that's important in the compression job
                                 .concatMap(o -> {
+                                    // Cache the first key from the observable so we can use it to create a key later
                                     Observable<Row> sharedRows = o.share();
                                     Observable<CompressedPointContainer> compressed =
                                             sharedRows.compose(new TempTableCompressTransformer(startTimeSlice));
                                     Observable<Row> keyTake = sharedRows.take(1);
 
+                                    // Merge the first row with the compressed package to be able to write to Cassandra
                                     return compressed.zipWith(keyTake, (cpc, r) -> {
                                         MetricId<?> metricId =
                                                 new MetricId(r.getString(0), MetricType.fromCode(r.getByte(1)),
@@ -769,13 +774,10 @@ public Completable compressBlock(long startTimeSlice, int pageSize, int maxConcu
                                     });
                                 }), maxConcurrency)
                         .flatMap(rs -> rs)
-                        .doOnError(Throwable::printStackTrace)
-                        .doOnCompleted(() -> {
-                            log.infof("Compress part completed");
-                            dataAccess.dropTempTable(startTimeSlice)
-                                    .subscribeOn(Schedulers.io())
-                                    .subscribe();
-                        })
+                        .doOnCompleted(() -> dataAccess.dropTempTable(startTimeSlice)
+                                .compose(applyRetryPolicy())
+                                .subscribeOn(Schedulers.io())
+                                .subscribe())
         );
     }