Fix NullPointerException and 'no more capacity' issues in blob store

ambry-api/src/test/java/com.github.ambry/store/MockWrite.java

@@ -16,6 +16,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
+import java.util.Objects;
 
 
 /**
@@ -53,8 +54,8 @@ public void appendFrom(ReadableByteChannel channel, long size) throws StoreExcep
     } catch (IOException e) {
       // The IOException message may vary in different java versions. As code evolves, we may need to update IO_ERROR_STR
       // in StoreException (based on java version that is being employed) to correctly capture disk I/O related errors.
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while writing into store", e, errorCode);
     }
     buf.limit(savedLimit);

ambry-replication/src/test/java/com.github.ambry.replication/InMemoryStore.java

@@ -35,6 +35,7 @@
 import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Objects;
 import java.util.Set;
 
 import static com.github.ambry.replication.ReplicationTest.*;
@@ -123,8 +124,9 @@ public void appendFrom(ReadableByteChannel channel, long size) throws StoreExcep
           sizeRead += channel.read(buf);
         }
       } catch (IOException e) {
-        StoreErrorCodes errorCode = e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
-            : StoreErrorCodes.Unknown_Error;
+        StoreErrorCodes errorCode =
+            Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+                : StoreErrorCodes.Unknown_Error;
         throw new StoreException(errorCode.toString() + " while writing into dummy log", e, errorCode);
       }
       buf.flip();

ambry-replication/src/test/java/com.github.ambry.replication/ReplicationTest.java

@@ -1128,8 +1128,8 @@ public void replicaTokenTest() throws InterruptedException {
     Time time = new MockTime();
     MockFindToken token1 = new MockFindToken(0, 0);
     RemoteReplicaInfo remoteReplicaInfo = new RemoteReplicaInfo(new MockReplicaId(), new MockReplicaId(),
-        new InMemoryStore(null, Collections.emptyList(), Collections.emptyList(), null), token1, tokenPersistInterval, time,
-        new Port(5000, PortType.PLAINTEXT));
+        new InMemoryStore(null, Collections.emptyList(), Collections.emptyList(), null), token1, tokenPersistInterval,
+        time, new Port(5000, PortType.PLAINTEXT));
 
     // The equality check is for the reference, which is fine.
     // Initially, the current token and the token to persist are the same.

ambry-store/src/main/java/com.github.ambry.store/HardDeleter.java

@@ -30,6 +30,7 @@
 import java.util.EnumSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -234,8 +235,8 @@ void performRecovery() throws StoreException {
       }
     } catch (IOException e) {
       metrics.hardDeleteExceptionsCount.inc();
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " occurred while performing hard delete ", e, errorCode);
     }
       /* Now that all the blobs in the range were successfully hard deleted, the next time hard deletes can be resumed
@@ -540,8 +541,8 @@ private void persistCleanupToken() throws IOException, StoreException {
       fileStream.getChannel().force(true);
       tempFile.renameTo(actual);
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(
           errorCode.toString() + " while persisting cleanup tokens to disk " + tempFile.getAbsoluteFile(), errorCode);
     } finally {
@@ -621,8 +622,8 @@ private void performHardDeletes(List<MessageInfo> messageInfoList) throws StoreE
         diskIOScheduler.getSlice(HARD_DELETE_CLEANUP_JOB_NAME, HARD_DELETE_CLEANUP_JOB_NAME, logWriteInfo.size);
       }
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while performing hard delete ", e, errorCode);
     }
     logger.trace("Performed hard deletes from {} to {} for {}", startToken, endToken, dataDir);

ambry-store/src/main/java/com.github.ambry.store/IndexSegment.java

@@ -41,6 +41,7 @@
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.NavigableSet;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.ConcurrentSkipListMap;
@@ -400,8 +401,8 @@ private void persistBloomFilter() throws StoreException {
       stream.writeLong(crcValue);
       stream.close();
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while trying to persist bloom filter", e, errorCode);
     }
   }
@@ -448,12 +449,12 @@ private StoreKey getKeyAt(ByteBuffer mmap, int index) throws StoreException {
       mmap.position(firstKeyRelativeOffset + index * persistedEntrySize);
       storeKey = factory.getStoreKey(new DataInputStream(new ByteBufferInputStream(mmap)));
     } catch (InternalError e) {
-      throw e.getMessage().equals(StoreException.INTERNAL_ERROR_STR) ? new StoreException(
+      throw Objects.equals(e.getMessage(), StoreException.INTERNAL_ERROR_STR) ? new StoreException(
           "Internal error occurred due to unsafe memory access", e, StoreErrorCodes.IOError)
           : new StoreException("Unknown internal error while trying to get store key", e,
               StoreErrorCodes.Unknown_Error);
     } catch (IOException e) {
-      throw e.getMessage().equals(StoreException.IO_ERROR_STR) ? new StoreException(
+      throw Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? new StoreException(
           "IO error while trying to get store key", e, StoreErrorCodes.IOError)
           : new StoreException("Unknown IO error while trying to get store key", e, StoreErrorCodes.Unknown_Error);
     } catch (Throwable t) {
@@ -700,8 +701,9 @@ void writeIndexSegmentToFile(Offset safeEndPoint) throws FileNotFoundException,
         // swap temp file with the original file
         temp.renameTo(getFile());
       } catch (IOException e) {
-        StoreErrorCodes errorCode = e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
-            : StoreErrorCodes.Unknown_Error;
+        StoreErrorCodes errorCode =
+            Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+                : StoreErrorCodes.Unknown_Error;
         throw new StoreException(
             "IndexSegment : " + indexFile.getAbsolutePath() + " encountered " + errorCode.toString()
                 + " while persisting index to disk", e, errorCode);
@@ -794,8 +796,8 @@ private void map() throws StoreException {
     } catch (FileNotFoundException e) {
       throw new StoreException("File not found while mapping the segment of index", e, StoreErrorCodes.File_Not_Found);
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while mapping the segment of index", e, errorCode);
     } finally {
       rwLock.writeLock().unlock();
@@ -906,8 +908,8 @@ private void readFromFile(File fileToRead, Journal journal) throws StoreExceptio
             StoreErrorCodes.Index_Creation_Failure);
       }
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException("IndexSegment : " + indexFile.getAbsolutePath() + " encountered " + errorCode.toString()
           + " while reading from file ", e, errorCode);
     }

ambry-store/src/main/java/com.github.ambry.store/Log.java

@@ -23,6 +23,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentNavigableMap;
 import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.atomic.AtomicLong;
@@ -326,7 +327,7 @@ private LogSegment checkArgsAndGetFirstSegment(long segmentCapacity) throws Stor
    * Creates {@link LogSegment} instances from {@code segmentFiles}.
    * @param segmentFiles the files that form the segments of the log.
    * @return {@code List} of {@link LogSegment} instances corresponding to {@code segmentFiles}.
-   * @throws IOException if there is an I/O error loading the segment files or creating {@link LogSegment} instances.
+   * @throws StoreException if there is an I/O error loading the segment files or creating {@link LogSegment} instances.
    */
   private List<LogSegment> getSegmentsToLoad(File[] segmentFiles) throws StoreException {
     List<LogSegment> segments = new ArrayList<>(segmentFiles.length);
@@ -391,8 +392,9 @@ private File allocate(String filename, long size) throws StoreException {
       try {
         diskSpaceAllocator.allocate(segmentFile, size);
       } catch (IOException e) {
-        StoreErrorCodes errorCode = e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
-            : StoreErrorCodes.Unknown_Error;
+        StoreErrorCodes errorCode =
+            Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+                : StoreErrorCodes.Unknown_Error;
         throw new StoreException(errorCode.toString() + " while allocating the file", e, errorCode);
       }
     }
@@ -410,8 +412,8 @@ private void free(LogSegment logSegment) throws StoreException {
       logSegment.close();
       diskSpaceAllocator.free(segmentFile, logSegment.getCapacityInBytes());
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while freeing log segment", e, errorCode);
     }
   }
@@ -463,9 +465,26 @@ private void ensureCapacity(long writeSize) throws StoreException {
     Pair<String, String> segmentNameAndFilename = getNextSegmentNameAndFilename();
     logger.info("Allocating new segment with name: " + segmentNameAndFilename.getFirst());
     File newSegmentFile = allocate(segmentNameAndFilename.getSecond(), segmentCapacity);
-    LogSegment newSegment =
-        new LogSegment(segmentNameAndFilename.getFirst(), newSegmentFile, segmentCapacity, metrics, true);
-    segmentsByName.put(segmentNameAndFilename.getFirst(), newSegment);
+    try {
+      LogSegment newSegment =
+          new LogSegment(segmentNameAndFilename.getFirst(), newSegmentFile, segmentCapacity, metrics, true);
+      segmentsByName.put(segmentNameAndFilename.getFirst(), newSegment);
+    } catch (StoreException e) {
+      logger.error("Failed to create new log segment {} with store exception: ", segmentNameAndFilename.getFirst(), e);
+      try {
+        diskSpaceAllocator.free(newSegmentFile, segmentCapacity);
+        remainingUnallocatedSegments.incrementAndGet();
+      } catch (IOException exception) {
+        StoreErrorCodes errorCode =
+            Objects.equals(exception.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+                : StoreErrorCodes.Unknown_Error;
+        throw new StoreException(
+            errorCode.toString() + " while freeing log segment " + segmentNameAndFilename.getFirst(), exception,
+            errorCode);
+      } finally {
+        metrics.overflowWriteError.inc();
+      }
+    }
   }
 
   /**

ambry-store/src/main/java/com.github.ambry.store/LogSegment.java

@@ -26,6 +26,7 @@
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 import java.nio.channels.ReadableByteChannel;
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -135,8 +136,8 @@ class LogSegment implements Read, Write {
     } catch (FileNotFoundException e) {
       throw new StoreException("File not found while creating log segment", e, StoreErrorCodes.File_Not_Found);
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while creating log segment", e, errorCode);
     }
   }
@@ -164,8 +165,8 @@ public int appendFrom(ByteBuffer buffer) throws StoreException {
     } catch (ClosedChannelException e) {
       throw new StoreException("Channel closed while writing into the log segment", e, StoreErrorCodes.Channel_Closed);
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while writing into the log segment", e, errorCode);
     }
     endOffset.addAndGet(bytesWritten);
@@ -209,8 +210,9 @@ public void appendFrom(ReadableByteChannel channel, long size) throws StoreExcep
           bytesWritten += fileChannel.write(byteBufferForAppend, endOffset.get() + bytesWritten);
         }
       } catch (IOException e) {
-        StoreErrorCodes errorCode = e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
-            : StoreErrorCodes.Unknown_Error;
+        StoreErrorCodes errorCode =
+            Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+                : StoreErrorCodes.Unknown_Error;
         throw new StoreException(errorCode.toString() + " while writing into the log segment", e, errorCode);
       }
     }
@@ -239,8 +241,8 @@ int appendFromDirectly(byte[] byteArray, int offset, int length) throws StoreExc
       directFile.write(byteArray, offset, length);
       endOffset.addAndGet(length);
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while writing into segment via direct IO", e, errorCode);
     }
     return length;
@@ -422,8 +424,8 @@ void setEndOffset(long endOffset) throws StoreException {
       fileChannel.position(endOffset);
       this.endOffset.set(endOffset);
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " while setting end offset of segment", e, errorCode);
     }
   }

ambry-store/src/main/java/com.github.ambry.store/PersistentIndex.java

@@ -32,6 +32,7 @@
 import java.util.ListIterator;
 import java.util.Map;
 import java.util.NavigableSet;
+import java.util.Objects;
 import java.util.Random;
 import java.util.Set;
 import java.util.TreeMap;
@@ -828,8 +829,8 @@ private BlobReadOptions getDeletedBlobReadOptions(IndexValue value, StoreKey key
             + "]", StoreErrorCodes.ID_Deleted);
       }
     } catch (IOException e) {
-      StoreErrorCodes errorCode =
-          e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError : StoreErrorCodes.Unknown_Error;
+      StoreErrorCodes errorCode = Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+          : StoreErrorCodes.Unknown_Error;
       throw new StoreException(errorCode.toString() + " when reading delete blob info from the log " + dataDir, e,
           errorCode);
     }
@@ -1661,8 +1662,9 @@ public synchronized void write() throws StoreException {
       } catch (FileNotFoundException e) {
         throw new StoreException("File not found while writing index to file", e, StoreErrorCodes.File_Not_Found);
       } catch (IOException e) {
-        StoreErrorCodes errorCode = e.getMessage().equals(StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
-            : StoreErrorCodes.Unknown_Error;
+        StoreErrorCodes errorCode =
+            Objects.equals(e.getMessage(), StoreException.IO_ERROR_STR) ? StoreErrorCodes.IOError
+                : StoreErrorCodes.Unknown_Error;
         throw new StoreException(errorCode.toString() + " while persisting index to disk", e, errorCode);
       } finally {
         context.stop();