[CELEBORN-1079] Fix use of GuardedBy in client-flink/common

### What changes were proposed in this pull request?

* Fix use of `GuardedBy` on nonexistant lock.
* Annotate methods, which are expected to be called with lock held, with `GuardedBy` so that error prone can analyze all invocations

### Why are the changes needed?
There is no functional change, but it helps errorprone analysis.

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Unit tests

Closes #2029 from mridulm/fix-flink-guarded-by-annotation.

Authored-by: Mridul Muralidharan <mridulatgmail.com>
Signed-off-by: zky.zhoukeyong <[email protected]>

client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/PartitionSortedBuffer.java

@@ -28,7 +28,6 @@
 import java.util.Arrays;
 
 import javax.annotation.Nullable;
-import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.NotThreadSafe;
 
 import org.apache.flink.core.memory.MemorySegment;
@@ -63,7 +62,6 @@ public class PartitionSortedBuffer implements SortBuffer {
   private final BufferPool bufferPool;
 
   /** A segment list as a joint buffer which stores all records and index entries. */
-  @GuardedBy("lock")
   private final ArrayList<MemorySegment> buffers = new ArrayList<>();
 
   /** Addresses of the first record's index entry for each subpartition. */
@@ -92,7 +90,6 @@ public class PartitionSortedBuffer implements SortBuffer {
   // For writing
   // ---------------------------------------------------------------------------------------------
   /** Whether this sort buffer is released. A released sort buffer can not be used. */
-  @GuardedBy("lock")
   private boolean isReleased;
   /** Array index in the segment list of the current available buffer for writing. */
   private int writeSegmentIndex;

client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/buffer/TransferBufferPool.java

@@ -155,6 +155,7 @@ public void recycle(ByteBuffer buffer) {
     }
   }
 
+  @GuardedBy("lock")
   private int assignCredits(CreditListener creditListener) {
     assert Thread.holdsLock(lock);
 
@@ -176,6 +177,7 @@ private int assignCredits(CreditListener creditListener) {
     return numCredits;
   }
 
+  @GuardedBy("lock")
   private List<CreditAssignment> dispatchReservedCredits() {
     assert Thread.holdsLock(lock);