docs: @SuppressWarnings("unchecked")
    private void consume(LogRecord record) {
      // Decode the raw log operation from the record.
      LogOperation operation = decodeInternal(record.value());
  
      // If this operation is not destined for this primitive, ignore it.
      // TODO: If multiple primitives of different types are created and destroyed on the same distributed log,
      // we need to be able to differentiate between different instances of a service by the service ID.
      if (!operation.primitive().equals(name())) {
        return;
      }
  
      // Create a session from the log record.
      Session session = getOrCreateSession(operation.sessionId());
  
      // Update the local context for the service.
      currentIndex = record.index();
      currentSession = session;
      currentOperation = operation.operationId().type();
      currentTimestamp = record.timestamp();
  
      // Apply the operation to the service.
      byte[] output = service.apply(new DefaultCommit<>(
          currentIndex,
          operation.operationId(),
          operation.operation(),
          currentSession,
          currentTimestamp));
  
      // If the operation session matches the local session, complete the write future.
      if (operation.sessionId().equals(this.session.sessionId())) {
        CompletableFuture future = writeFutures.remove(operation.operationIndex());
        if (future != null) {
          future.complete(decode(output));
        }
      }
  
      // Iterate through pending reads and complete any reads at indexes less than or equal to the applied index.
      PendingRead pendingRead = pendingReads.peek();
      while (pendingRead != null && pendingRead.index <= record.index()) {
        session = getOrCreateSession(this.session.sessionId());
        currentSession = session;
        currentOperation = OperationType.QUERY;
        try {
          output = service.apply(new DefaultCommit<>(
              currentIndex,
              pendingRead.operationId,
              pendingRead.bytes,
              session,
              currentTimestamp));
          pendingRead.future.complete(output);
        } catch (Exception e) {
          pendingRead.future.completeExceptionally(new PrimitiveException.ServiceException());
        }
        pendingReads.remove();
        pendingRead = pendingReads.peek();
      }
    }