Fix throwing deserialization exceptions for SnapshotIndexBlobs

[ajothomas]

Bug:
BlobStoreUtil's getSnapshotIndex method currently gets the SnapshotIndex blob using BlobStoreManager#get and deserializes it using SnapshotIndexSerde
We chain the deserialization logic to the get future and in turn chain handle to the deserialization future which tries to capture deserialization exceptions. This is incorrect as any exceptions that occurred during snapshot index get would also be processed by the chained handle.

This presents two issues:

• The current exception message says that the error is due to deserialization which is incorrect. It could have been due to some arbitrary exception during get
• It also wraps any exception into SamzaException class. get implementation of BlobStoreManager classifies some exceptions into retriable exceptions which will become useless if we wrap it into SamzaException.

Changes:

• Catch deserialization exception and throw appropriate message within thenApplyAsync where we deserialize the SnapshotIndex bytes.

Tests:

• ./gradlew build
• Added unit tests to assert if appropriate exception is thrown

API Changes:
None

Upgrade Instructions:
None

Usage Instructions:
None