diff --git a/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTable.java b/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTable.java
index eb54de2..7252584 100644
--- a/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTable.java
+++ b/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTable.java
@@ -196,7 +196,7 @@
   public CompletableFuture<Map<K, V>> getAllAsync(List<K> keys, Object ... args) {
     Preconditions.checkNotNull(keys, "null keys");
     if (keys.isEmpty()) {
-      return CompletableFuture.completedFuture(Collections.EMPTY_MAP);
+      return CompletableFuture.completedFuture(Collections.emptyMap());
     }
     return instrument(() -> asyncTable.getAllAsync(keys, args), metrics.numGetAlls, metrics.getAllNs)
         .handle((result, e) -> {
@@ -219,7 +219,7 @@
 
   @Override
   public <T> CompletableFuture<T> readAsync(int opId, Object ... args) {
-    return (CompletableFuture<T>) instrument(() -> asyncTable.readAsync(opId, args), metrics.numReads, metrics.readNs)
+    return instrument(() -> asyncTable.<T>readAsync(opId, args), metrics.numReads, metrics.readNs)
         .exceptionally(e -> {
           throw new SamzaException(String.format("Failed to read, opId=%d", opId), e);
         });
@@ -244,7 +244,7 @@
 
     return instrument(() -> asyncTable.putAsync(key, value, args), metrics.numPuts, metrics.putNs)
         .exceptionally(e -> {
-          throw new SamzaException("Failed to put a record with key=" + key, (Throwable) e);
+          throw new SamzaException("Failed to put a record with key=" + key, e);
         });
   }
 
@@ -282,7 +282,7 @@
         instrument(() -> asyncTable.putAllAsync(putRecords, args), metrics.numPutAlls, metrics.putAllNs))
         .exceptionally(e -> {
           String strKeys = records.stream().map(r -> r.getKey().toString()).collect(Collectors.joining(","));
-          throw new SamzaException(String.format("Failed to put records with keys=" + strKeys), e);
+          throw new SamzaException("Failed to put records with keys=" + strKeys, e);
         });
   }
 
@@ -301,7 +301,7 @@
     Preconditions.checkNotNull(key, "null key");
     return instrument(() -> asyncTable.deleteAsync(key, args), metrics.numDeletes, metrics.deleteNs)
         .exceptionally(e -> {
-          throw new SamzaException(String.format("Failed to delete the record for " + key), (Throwable) e);
+          throw new SamzaException("Failed to delete the record for " + key, e);
         });
   }
 
@@ -324,7 +324,7 @@
 
     return instrument(() -> asyncTable.deleteAllAsync(keys, args), metrics.numDeleteAlls, metrics.deleteAllNs)
         .exceptionally(e -> {
-          throw new SamzaException(String.format("Failed to delete records for " + keys), e);
+          throw new SamzaException("Failed to delete records for " + keys, e);
         });
   }
 
@@ -339,7 +339,7 @@
 
   @Override
   public <T> CompletableFuture<T> writeAsync(int opId, Object... args) {
-    return (CompletableFuture<T>) instrument(() -> asyncTable.writeAsync(opId, args), metrics.numWrites, metrics.writeNs)
+    return instrument(() -> asyncTable.<T>writeAsync(opId, args), metrics.numWrites, metrics.writeNs)
         .exceptionally(e -> {
           throw new SamzaException(String.format("Failed to write, opId=%d", opId), e);
         });