CM-1630: Fixed by properly registering onComplete handlers.

Author: yaricom

comet-java-client/src/main/java/ml/comet/experiment/impl/BaseExperimentAsync.java

@@ -313,7 +313,11 @@ void logAssetFolder(@NonNull File folder, boolean logFilePath, boolean recursive
             Observable<LogDataResponse> observable =
                     Observable.fromStream(assets)
                             .flatMap(asset -> Observable.fromSingle(
-                                    sendAssetAsync(getRestApiClient()::logAsset, asset, onComplete)), true);
+                                    sendAssetAsync(getRestApiClient()::logAsset, asset)), true);
+
+            if (onComplete.isPresent()) {
+                observable = observable.doFinally(onComplete.get());
+            }
 
             // subscribe for processing results
             observable
@@ -450,7 +454,11 @@ void logAsset(@NonNull final Asset asset, Optional<Action> onComplete) {
     @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
     private <T extends Asset> void logAsset(final BiFunction<T, String, Single<LogDataResponse>> func,
                                             @NonNull final T asset, Optional<Action> onComplete) {
-        Single<LogDataResponse> single = this.sendAssetAsync(func, asset, onComplete);
+        Single<LogDataResponse> single = this.sendAssetAsync(func, asset);
+
+        if (onComplete.isPresent()) {
+            single = single.doFinally(onComplete.get());
+        }
 
         // subscribe to get operation completed
         single.subscribe(
@@ -467,30 +475,21 @@ private <T extends Asset> void logAsset(final BiFunction<T, String, Single<LogDa
      * Attempts to send given {@link Asset} or its subclass asynchronously.
      * This method will wrap send operation into {@link Single} and transparently log any errors that may happen.
      *
-     * @param func       the function to be invoked to send asset to the backend.
-     * @param asset      the {@link Asset} or subclass to be sent.
-     * @param onComplete The optional action to be invoked when this operation
-     *                   asynchronously completes. Can be {@code null} if not interested in completion signal.
-     * @param <T>        the {@link Asset} or its subclass.
+     * @param func  the function to be invoked to send asset to the backend.
+     * @param asset the {@link Asset} or subclass to be sent.
+     * @param <T>   the {@link Asset} or its subclass.
      * @return the {@link Single} which can be used to subscribe for operation results.
      */
-    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
     private <T extends Asset> Single<LogDataResponse> sendAssetAsync(
-            final BiFunction<T, String, Single<LogDataResponse>> func,
-            @NonNull final T asset, Optional<Action> onComplete) {
+            final BiFunction<T, String, Single<LogDataResponse>> func, @NonNull final T asset) {
 
-        Single<LogDataResponse> single = validateAndGetExperimentKey()
+        return validateAndGetExperimentKey()
                 .subscribeOn(Schedulers.io())
-                .concatMap(experimentKey -> func.apply(asset, experimentKey))
+                .concatMap(experimentKey1 -> func.apply(asset, experimentKey1))
                 .doOnSuccess(logDataResponse ->
                         AsyncDataResponseLogger.checkAndLog(logDataResponse, getLogger(), asset))
                 .doOnError(throwable ->
                         getLogger().error(getString(FAILED_TO_SEND_LOG_ASSET_REQUEST, asset), throwable));
-
-        if (onComplete.isPresent()) {
-            return single.doFinally(onComplete.get());
-        }
-        return single;
     }
 
     /**