Avoid swallowing unexpected exceptions. - Fixes
@@ -1752,7 +1752,9 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
topicBytes = U.marshal(ctx, topic);
}
catch (IgniteCheckedException e) {
- U.error(log, "Failed to marshal (request will not be sent).", e);
+ U.error(log, "Failed to marshal.", e);
+
+ curFut.onDone(new IgniteException("Failed to marshal.", e));
return;
}
@@ -1770,8 +1772,9 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
cache.context().deploy().onEnter();
}
catch (IgniteCheckedException e) {
- U.error(log, "Failed to deploy class (request will not be sent): " +
- jobPda0.deployClass(), e);
+ U.error(log, "Failed to deploy class: " + jobPda0.deployClass(), e);
+
+ curFut.onDone(new IgniteException("Failed to deploy class: " + jobPda0.deployClass(), e));
return;
}
@@ -2131,7 +2134,11 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
// No-op.
}
catch (IgniteCheckedException ex) {
- throw new IgniteException("Failed to set initial value for cache entry", ex);
+ IgniteLogger log = cache.unwrap(Ignite.class).log();
+
+ U.error(log, "Failed to set initial value for cache entry: " + e, ex);
+
+ throw new IgniteException("Failed to set initial value for cache entry.", ex);
}
finally {
cctx.shared().database().checkpointReadUnlock();
@@ -2152,7 +2159,9 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed
cctx.shared().wal().flush(null, false);
}
catch (IgniteCheckedException e) {
- U.error(log, "Failed to write preloaded entries into write-ahead log: " + e, e);
+ U.error(log, "Failed to write preloaded entries into write-ahead log.", e);
+
+ throw new IgniteException("Failed to write preloaded entries into write-ahead log.", e);
}
}
}
