[multistage] [bug] Limited capacity for mailbox queue causes mailbox to drop

DEFAULT_CHANNEL_CAPACITY for each InMemory/Grpc mailbox is set to 5.

5 is a pretty small number.

Whenever we have a MailboxSendOperator that produces more than 5 block really fast before MailboxReceiveOperator has a chance to run, we will run into error since MailboxSendOperator doesn't retry.

Propose solution
1) Increase the DEFAULT_CHANNEL_CAPACITY. This is easy to do. but this will add memory pressure
2) Return an non-op block after mailbox send reaches 5. this is ideal. however, it needs some re-design to make it happen: one idea is to replace ArrayBlockingQueue with AsynchronousChannel to be consistent with grpc exchange channel.

The problematic code is InMemorySendingMailbox.send()

@Override
public void send(TransferableBlock data)
throws UnsupportedOperationException {
try {
if (!_queue.offer(
data, InMemoryMailboxService.DEFAULT_CHANNEL_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
throw new RuntimeException(String.format("Timed out when sending block in mailbox=%s", _mailboxId));
}
_gotMailCallback.accept(new StringMailboxIdentifier(_mailboxId));
} catch (InterruptedException e) {
throw new RuntimeException("Interrupted trying to send data through the channel", e);
}
}
_queue.offer will throw an exception when q is full and it will always fail.

@walterddr @ankitsultana FYI

GrpcMailbox has the same issue.

For grpcMailbox, the mailbox will silently get dropped if the queue is full, which is worse.

The code is in MailboxContentStreamObserver onNext().

@Override
public void onNext(Mailbox.MailboxContent mailboxContent) {
_mailboxId = new StringMailboxIdentifier(mailboxContent.getMailboxId());

GrpcReceivingMailbox receivingMailbox = (GrpcReceivingMailbox) _mailboxService.getReceivingMailbox(_mailboxId);
_gotMailCallback = receivingMailbox.init(this);

if (!mailboxContent.getMetadataMap().containsKey(ChannelUtils.MAILBOX_METADATA_BEGIN_OF_STREAM_KEY)) {
// when the receiving end receives a message put it in the mailbox queue.
_receivingBuffer.offer(mailboxContent);
_gotMailCallback.accept(_mailboxId);

if (_isEnabledFeedback) {
// TODO: this has race conditions with onCompleted() because sender blindly closes connection channels once
// it has finished sending all the data packets.
int remainingCapacity = _receivingBuffer.remainingCapacity() - 1;
Mailbox.MailboxStatus.Builder builder =
Mailbox.MailboxStatus.newBuilder().setMailboxId(mailboxContent.getMailboxId())
.putMetadata(ChannelUtils.MAILBOX_METADATA_BUFFER_SIZE_KEY, String.valueOf(remainingCapacity));
if (mailboxContent.getMetadataMap().get(ChannelUtils.MAILBOX_METADATA_END_OF_STREAM_KEY) != null) {
builder.putMetadata(ChannelUtils.MAILBOX_METADATA_END_OF_STREAM_KEY, "true");
}
Mailbox.MailboxStatus status = builder.build();
// returns the buffer available size to sender for rate controller / throttling.
_responseObserver.onNext(status);
}
}
}

_receivingBuffer.offer(mailboxContent); will silent drop the mailbox if the queue is full. I checked in the cluster testing and this should be the reason for https://github.com/apache/pinot/issues/9963

All of this means we need to redesign on how mailbox communication happens or we temperately set the queue size to a very large number just for now.