SpanningRecordSerializer implements RecordSerializer<T> { @Override public boolean hasData() { return this.position > 0 || (this.lengthBuffer.hasRemaining() || this.dataBuffer.hasRemaining()); } SpanningRecordSerializer(); @Override SerializationResult addRecord(T record); @Override SerializationResult setNextBuffer(Buffer buffer); @Override Buffer getCurrentBuffer(); @Override void clearCurrentBuffer(); @Override void clear(); @Override boolean hasData(); @Override void instantiateMetrics(TaskIOMetricGroup metrics);  }
@Test public void testHasData() { final int SEGMENT_SIZE = 16; final SpanningRecordSerializer<SerializationTestType> serializer = new SpanningRecordSerializer<SerializationTestType>(); final Buffer buffer = new Buffer(MemorySegmentFactory.allocateUnpooledSegment(SEGMENT_SIZE), mock(BufferRecycler.class)); final SerializationTestType randomIntRecord = Util.randomRecord(SerializationTestTypeFactory.INT); Assert.assertFalse(serializer.hasData()); try { serializer.addRecord(randomIntRecord); Assert.assertTrue(serializer.hasData()); serializer.setNextBuffer(buffer); Assert.assertTrue(serializer.hasData()); serializer.clear(); Assert.assertFalse(serializer.hasData()); serializer.setNextBuffer(buffer); serializer.addRecord(randomIntRecord); Assert.assertTrue(serializer.hasData()); serializer.addRecord(randomIntRecord); Assert.assertTrue(serializer.hasData()); } catch (Exception e) { e.printStackTrace(); Assert.fail(e.getMessage()); } }