[hotfix] Parameterlize RecordWriterTest and migrate it to junit5

jonchase · Sep 14, 2023 · e198c09 · e198c09
1 parent 2d74260
commit e198c09
Show file tree

Hide file tree

Showing 2 changed files with 96 additions and 98 deletions.
diff --git a/...c/test/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriterTest.java b/...c/test/java/org/apache/flink/runtime/io/network/api/writer/BroadcastRecordWriterTest.java
@@ -31,31 +31,31 @@
 import org.apache.flink.testutils.serialization.types.SerializationTestTypeFactory;
 import org.apache.flink.testutils.serialization.types.Util;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 
 import java.io.IOException;
+import java.nio.file.Path;
 import java.util.ArrayDeque;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import static org.junit.Assert.assertEquals;
+import static org.apache.flink.runtime.io.network.api.writer.RecordWriterTest.createResultPartition;
+import static org.apache.flink.runtime.io.network.api.writer.RecordWriterTest.verifyDeserializationResults;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for the {@link BroadcastRecordWriter}. */
-public class BroadcastRecordWriterTest extends RecordWriterTest {
-
-    public BroadcastRecordWriterTest() {
-        super(true);
-    }
+class BroadcastRecordWriterTest {
 
     /**
      * Tests the number of requested buffers and results are correct in the case of switching modes
      * between {@link BroadcastRecordWriter#broadcastEmit(IOReadableWritable)} and {@link
      * BroadcastRecordWriter#randomEmit(IOReadableWritable)}.
      */
     @Test
-    public void testBroadcastMixedRandomEmitRecord() throws Exception {
+    void testBroadcastMixedRandomEmitRecord(@TempDir Path tempPath) throws Exception {
         final int numberOfChannels = 8;
         final int numberOfRecords = 8;
         final int bufferSize = 32;
@@ -65,7 +65,7 @@ public void testBroadcastMixedRandomEmitRecord() throws Exception {
                 new BroadcastRecordWriter<>(partition, -1, "test");
         final RecordDeserializer<SerializationTestType> deserializer =
                 new SpillingAdaptiveSpanningRecordDeserializer<>(
-                        new String[] {tempFolder.getRoot().getAbsolutePath()});
+                        new String[] {tempPath.toString()});
 
         // generate the configured number of int values as global record set
         final Iterable<SerializationTestType> records =
@@ -94,11 +94,11 @@ public void testBroadcastMixedRandomEmitRecord() throws Exception {
                 partition.getBufferPool().bestEffortGetNumOfUsedBuffers();
         // verify the expected number of requested buffers, and it would always request a new buffer
         // while random emitting
-        assertEquals(2 * numberOfRecords, numberOfCreatedBuffers);
+        assertThat(2 * numberOfRecords).isEqualTo(numberOfCreatedBuffers);
 
         for (int i = 0; i < numberOfChannels; i++) {
             // every channel would queue the number of above crated buffers
-            assertEquals(numberOfRecords + 1, partition.getNumberOfQueuedBuffers(i));
+            assertThat(partition.getNumberOfQueuedBuffers(i)).isEqualTo(numberOfRecords + 1);
 
             final int excessRandomRecords = i < numberOfRecords % numberOfChannels ? 1 : 0;
             final int numberOfRandomRecords =
@@ -119,7 +119,7 @@ public void testBroadcastMixedRandomEmitRecord() throws Exception {
      * when all consumers are closed.
      */
     @Test
-    public void testRandomEmitAndBufferRecycling() throws Exception {
+    void testRandomEmitAndBufferRecycling() throws Exception {
         int recordSize = 8;
         int numberOfChannels = 2;
 
@@ -132,35 +132,34 @@ public void testRandomEmitAndBufferRecycling() throws Exception {
         List<Buffer> buffers =
                 Arrays.asList(bufferPool.requestBuffer(), bufferPool.requestBuffer());
         buffers.forEach(Buffer::recycleBuffer);
-        assertEquals(3, bufferPool.getNumberOfAvailableMemorySegments());
+        assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isEqualTo(3);
 
         // fill first buffer
         writer.broadcastEmit(new IntType(1));
         writer.broadcastEmit(new IntType(2));
-        assertEquals(2, bufferPool.getNumberOfAvailableMemorySegments());
+        assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isEqualTo(2);
 
         // simulate consumption of first buffer consumer; this should not free buffers
-        assertEquals(1, partition.getNumberOfQueuedBuffers(0));
+        assertThat(partition.getNumberOfQueuedBuffers(0)).isOne();
         ResultSubpartitionView view0 =
                 partition.createSubpartitionView(0, new NoOpBufferAvailablityListener());
         closeConsumer(view0, 2 * recordSize);
-        assertEquals(2, bufferPool.getNumberOfAvailableMemorySegments());
+        assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isEqualTo(2);
 
         // use second buffer
         writer.emit(new IntType(3), 0);
-        assertEquals(1, bufferPool.getNumberOfAvailableMemorySegments());
-
+        assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isOne();
         // fully free first buffer
-        assertEquals(1, partition.getNumberOfQueuedBuffers(1));
+        assertThat(partition.getNumberOfQueuedBuffers(1)).isOne();
         ResultSubpartitionView view1 =
                 partition.createSubpartitionView(1, new NoOpBufferAvailablityListener());
         closeConsumer(view1, 2 * recordSize);
-        assertEquals(2, bufferPool.getNumberOfAvailableMemorySegments());
+        assertThat(bufferPool.getNumberOfAvailableMemorySegments()).isEqualTo(2);
     }
 
     public void closeConsumer(ResultSubpartitionView view, int expectedSize) throws IOException {
         Buffer buffer = view.getNextBuffer().buffer();
-        assertEquals(expectedSize, buffer.getSize());
+        assertThat(buffer.getSize()).isEqualTo(expectedSize);
         buffer.recycleBuffer();
     }
 }