HDDS-5674.[Ozone-Streaming] Handle client retries on exception

[sadanand48]

What changes were proposed in this pull request?

Handle client write retries on exception

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-5674

How was this patch tested?

Unit tests.

[szetszwo]

@sadanand48 , thanks a lot for working on this.

In Streaming, we should not have a buffer pool since we won't reuse buffers. We probably need a new class for the bufferList. Then, we can put all the related logic inside the new class.

BTW, please add some tests. Thanks.

[sadanand48]

Thanks @szetszwo and @bshashikant for the review. Yes I will be adding more tests. Also , we are not using BufferPool
here instead maintaining a list that contains a reference to the buffers that are passed by the client. Currently we maintain a commitIndexMap which contains list of buffers for a particular commitIndex and the buffers are discarded from the list during stream watchForCommit.

Currently on retry , we discard the previous buffers in the bufferList and start from 0 regardless of whether all the nodes have upto certain data. In a follow up task , I will maintain stream ack like you suggested and only retry writing the data that is not present in all nodes.

[szetszwo]

A major problem is the following line in StreamCommitWatcher.releaseBuffers(..)

 bufferPool.remove(byteBuffer);

since it is super expensive. We need some way to avoid it. Thanks.

[szetszwo]

int j = 0;
for(; j < buffers.size(); j++) {
  if (byteBuffer == buffers.get(i)) {
    break;
  }
}
buffers.remove(j);

[szetszwo]

Actually, using == to compare ByteBuffer may not work since there are duplicated buffers. We need to create a new class. Let me think about it.

[szetszwo]

public class StreamBuffer {
  private final ByteBuffer buffer;

  public StreamBuffer(ByteBuffer buffer) {
    this.buffer = buffer.asReadOnlyBuffer();
  }

  public StreamBuffer(ByteBuffer buffer, int offset, int length) {
    this((ByteBuffer) buffer.asReadOnlyBuffer().position(offset).limit(offset + length));
  }


  public ByteBuffer duplicate() {
    return buffer.duplicate();
  }

  public int length() {
    return buffer.limit() - buffer.position();
  }
}

We may create a new StreamBuffer as shown above. Then, we can use ==. Tried it with the current change. It seems working; please test it: https://issues.apache.org/jira/secure/attachment/13035081/StreamBuffer.patch

[sadanand48]

Thanks @szetszwo for the review comments. I have updated the patch. Will add more tests in a separate jira

[sadanand48]

We may create a new StreamBuffer as shown above. Then, we can use ==. Tried it with the current change. It seems working; please test it: https://issues.apache.org/jira/secure/attachment/13035081/StreamBuffer.patch

We are not comparing duplicated buffers here, since the same StreamBuffer object is present in bufferList and the commitInfoMap so we can use equals() and since its on the new StreamBuffer Class it will not compare every byte of the ByteBuffer if I am not wrong,

[szetszwo]

@sadanand48 , thanks for the update. Just a few minor comments inlined.

[szetszwo]

+1 the latest change look good.