[flink] branch master updated: [FLINK-21517][tests] Test harnesses should not bypass serialization stack for events

[pn...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 01352aa  [FLINK-21517][tests] Test harnesses should not bypass serialization stack for events
01352aa is described below

commit 01352aa7ddd5e9abf15984ff77fa929b1eabaff4
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Feb 26 13:55:43 2021 +0100

    [FLINK-21517][tests] Test harnesses should not bypass serialization stack for events
    
    Since FLINK-19297 we accidentally removed a test coverage for event (de)serialization from a lot of the unit tests,
    that were/are using test harnesses. For example because of that I almost broke 1.12.2 release,
    since `stop-with-savepoint --drain` was never only tested using test harnesses (didn't have an ITCases
    and/or end to end test).
---
 .../RecordOrEventCollectingResultPartitionWriter.java   | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordOrEventCollectingResultPartitionWriter.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordOrEventCollectingResultPartitionWriter.java
index cc85daa..2653ce0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordOrEventCollectingResultPartitionWriter.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/api/writer/RecordOrEventCollectingResultPartitionWriter.java
@@ -20,9 +20,11 @@ package org.apache.flink.runtime.io.network.api.writer;
 
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.runtime.event.AbstractEvent;
+import org.apache.flink.runtime.io.network.api.serialization.EventSerializer;
 import org.apache.flink.runtime.io.network.api.serialization.RecordDeserializer;
 import org.apache.flink.runtime.io.network.api.serialization.SpillingAdaptiveSpanningRecordDeserializer;
 import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferConsumer;
 import org.apache.flink.runtime.plugable.DeserializationDelegate;
 import org.apache.flink.runtime.plugable.NonReusingDeserializationDelegate;
 
@@ -48,7 +50,20 @@ public class RecordOrEventCollectingResultPartitionWriter<T>
 
     @Override
     public void broadcastEvent(AbstractEvent event, boolean isPriorityEvent) throws IOException {
-        output.add(event);
+        // Go through the serialization/deserialization to provide better test coverage in the
+        // unit tests using test harnesses. Otherwise bugs in event serialisation would be only
+        // visible in ITCases or end to end tests.
+        try (BufferConsumer eventBufferConsumer =
+                EventSerializer.toBufferConsumer(event, isPriorityEvent)) {
+            Buffer buffer = eventBufferConsumer.build();
+            try {
+                AbstractEvent deserializedEvent =
+                        EventSerializer.fromBuffer(buffer, getClass().getClassLoader());
+                output.add(deserializedEvent);
+            } finally {
+                buffer.recycleBuffer();
+            }
+        }
     }
 
     @Override