Auto-calculate reasonable values for WriteBatchPoolSize and return batches to the pool even if commitlog queue is full

src/dbnode/persist/fs/commitlog/commit_log.go

@@ -638,6 +638,13 @@ func (l *commitLog) writeWait(
 	if numEnqueued > l.maxQueueSize {
 		atomic.AddInt64(&l.numWritesInQueue, int64(-numToEnqueue))
 		l.closedState.RUnlock()
+
+		if write.writeBatch != nil {
+			// Make sure to finalize the write batch even though we didn't accept the writes
+			// so it can be returned to the pool.
+			write.writeBatch.Finalize()
+		}
+
 		return ErrCommitLogQueueFull
 	}
 
@@ -676,6 +683,13 @@ func (l *commitLog) writeBehind(
 	if numEnqueued > l.maxQueueSize {
 		atomic.AddInt64(&l.numWritesInQueue, int64(-numToEnqueue))
 		l.closedState.RUnlock()
+
+		if write.writeBatch != nil {
+			// Make sure to finalize the write batch even though we didn't accept the writes
+			// so it can be returned to the pool.
+			write.writeBatch.Finalize()
+		}
+
 		return ErrCommitLogQueueFull
 	}
 

src/dbnode/server/server.go

@@ -984,14 +984,32 @@ func withEncodingAndPoolingOptions(
 
 	var writeBatchPoolInitialBatchSize *int
 	if policy.WriteBatchPool.InitialBatchSize != nil {
+		// Use config value if available.
 		writeBatchPoolInitialBatchSize = policy.WriteBatchPool.InitialBatchSize
+	} else {
+		// Otherwise use the default batch size that the client will use.
+		clientDefaultSize := client.DefaultWriteBatchSize
+		writeBatchPoolInitialBatchSize = &clientDefaultSize
 	}
+
 	var writeBatchPoolMaxBatchSize *int
 	if policy.WriteBatchPool.MaxBatchSize != nil {
 		writeBatchPoolMaxBatchSize = policy.WriteBatchPool.MaxBatchSize
 	}
+
+	writeBatchPoolPolicy := policy.WriteBatchPool.Pool
+	if writeBatchPoolPolicy.Size == 0 {
+		// If no value set, calculate a reasonabble value based on the commit log
+		// queue size. We base it off the commitlog queue size because we will
+		// want to be able to buffer at least one full commitlog queues worth of
+		// writes without allocating because these objects are very expensive to
+		// allocate.
+		commitlogQueueSize := opts.CommitLogOptions().BacklogQueueSize()
+		expectedBatchSize := *writeBatchPoolInitialBatchSize
+		writeBatchPoolPolicy.Size = commitlogQueueSize / expectedBatchSize
+	}
 	writeBatchPool := ts.NewWriteBatchPool(
-		poolOptions(policy.WriteBatchPool.Pool, scope.SubScope("write-batch-pool")),
+		poolOptions(writeBatchPoolPolicy, scope.SubScope("write-batch-pool")),
 		writeBatchPoolInitialBatchSize,
 		writeBatchPoolMaxBatchSize)
 

src/dbnode/storage/database.go

@@ -505,9 +505,6 @@ func (d *db) Write(
 	}
 
 	series, err := n.Write(ctx, id, timestamp, value, unit, annotation)
-	if err == commitlog.ErrCommitLogQueueFull {
-		d.errors.Record(1)
-	}
 	if err != nil {
 		return err
 	}
@@ -517,7 +514,15 @@ func (d *db) Write(
 	}
 
 	dp := ts.Datapoint{Timestamp: timestamp, Value: value}
-	return d.commitLog.Write(ctx, series, dp, unit, annotation)
+	err = d.commitLog.Write(ctx, series, dp, unit, annotation)
+	if err == commitlog.ErrCommitLogQueueFull {
+		d.errors.Record(1)
+	}
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (d *db) WriteTagged(
@@ -537,9 +542,6 @@ func (d *db) WriteTagged(
 	}
 
 	series, err := n.WriteTagged(ctx, id, tags, timestamp, value, unit, annotation)
-	if err == commitlog.ErrCommitLogQueueFull {
-		d.errors.Record(1)
-	}
 	if err != nil {
 		return err
 	}
@@ -549,7 +551,15 @@ func (d *db) WriteTagged(
 	}
 
 	dp := ts.Datapoint{Timestamp: timestamp, Value: value}
-	return d.commitLog.Write(ctx, series, dp, unit, annotation)
+	err = d.commitLog.Write(ctx, series, dp, unit, annotation)
+	if err == commitlog.ErrCommitLogQueueFull {
+		d.errors.Record(1)
+	}
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (d *db) BatchWriter(namespace ident.ID, batchSize int) (ts.BatchWriter, error) {
@@ -634,10 +644,6 @@ func (d *db) writeBatch(
 				write.Write.Annotation,
 			)
 		}
-
-		if err == commitlog.ErrCommitLogQueueFull {
-			d.errors.Record(1)
-		}
 		if err != nil {
 			// Return errors with the original index provided by the caller so they
 			// can associate the error with the write that caused it.
@@ -653,10 +659,22 @@ func (d *db) writeBatch(
 	}
 
 	if !n.Options().WritesToCommitLog() {
+		// Finalize here because we can't rely on the commitlog to do it since we're not
+		// using it.
+		writes.Finalize()
 		return nil
 	}
 
-	return d.commitLog.WriteBatch(ctx, writes)
+	err = d.commitLog.WriteBatch(ctx, writes)
+	if err == commitlog.ErrCommitLogQueueFull {
+		numFailedWrites := int64(len(writes.Iter()))
+		d.errors.Record(numFailedWrites)
+	}
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (d *db) QueryIDs(