diff --git a/codec-java/src/main/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoder.java b/codec-java/src/main/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoder.java index 9adb061a..346d950d 100644 --- a/codec-java/src/main/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoder.java +++ b/codec-java/src/main/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoder.java @@ -418,11 +418,8 @@ public byte[] useAsByteArray(Consumer block) { // --- private private EdgeBuffer borrow() { - if (pool != null) { - return pool.poll(); - } else { - return new EdgeBuffer(); - } + EdgeBuffer reusable = pool == null ? null : pool.poll(); + return reusable != null ? reusable : new EdgeBuffer(); } private void release(EdgeBuffer reusable) { diff --git a/codec-java/src/test/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoderTest.java b/codec-java/src/test/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoderTest.java index 5cbc1d86..62531b73 100644 --- a/codec-java/src/test/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoderTest.java +++ b/codec-java/src/test/java/com/kakao/actionbase/v2/core/code/AbstractEdgeEncoderTest.java @@ -2,13 +2,71 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; class AbstractEdgeEncoderTest { + @Test + void shouldEncodeWhenPoolIsExhaustedByConcurrentBorrow() throws Exception { + EdgeEncoderFactory factory = new EdgeEncoderFactory(1); + BytesKeyValueEdgeEncoder encoder = factory.getBytesKeyValueEncoder(); + CountDownLatch firstBorrowed = new CountDownLatch(1); + Semaphore releaseFirst = new Semaphore(0); + AtomicReference firstBuffer = new AtomicReference<>(); + AtomicReference secondBuffer = new AtomicReference<>(); + ExecutorService executor = Executors.newFixedThreadPool(2); + + try { + Future first = + executor.submit( + () -> + encoder.useAsByteArray( + buffer -> { + firstBuffer.set(buffer); + buffer.encodeInt8((byte) 1); + firstBorrowed.countDown(); + releaseFirst.acquireUninterruptibly(); + })); + + assertTrue(firstBorrowed.await(10, TimeUnit.SECONDS)); + Future second = + executor.submit( + () -> + encoder.useAsByteArray( + buffer -> { + secondBuffer.set(buffer); + buffer.encodeInt8((byte) 2); + })); + + byte[] secondBytes = second.get(10, TimeUnit.SECONDS); + assertNotSame(firstBuffer.get(), secondBuffer.get()); + releaseFirst.release(); + byte[] firstBytes = first.get(10, TimeUnit.SECONDS); + + BytesKeyValueEdgeEncoder unpooled = new EdgeEncoderFactory().getBytesKeyValueEncoder(); + assertArrayEquals(unpooled.useAsByteArray(buffer -> buffer.encodeInt8((byte) 1)), firstBytes); + assertArrayEquals( + unpooled.useAsByteArray(buffer -> buffer.encodeInt8((byte) 2)), secondBytes); + } finally { + releaseFirst.release(); + executor.shutdownNow(); + executor.awaitTermination(10, TimeUnit.SECONDS); + } + } + @Test void shouldReturnBorrowedBufferAfterEncodingFails() { EdgeEncoderFactory factory = new EdgeEncoderFactory(1);