@Override
        public List<SegmentSearchResult> findAll(final long fromTime, final long toTime) {
            long currNextTimestamp = nextTimestamp;
            final List<SegmentSearchResult> segmentSearchResults = new ArrayList<>();
            long currTimestamp = -1L; // choose an invalid timestamp. if this is valid, this needs to be re-worked
            int currValueSize;
            int currIndex = 0;
            int cumValueSize = 0;
            while (currTimestamp != minTimestamp) {
                final int timestampSegmentIndex = 2 * TIMESTAMP_SIZE + currIndex * (TIMESTAMP_SIZE + VALUE_SIZE);
                currTimestamp = ByteBuffer.wrap(segmentValue).getLong(timestampSegmentIndex);
                currValueSize = ByteBuffer.wrap(segmentValue).getInt(timestampSegmentIndex + TIMESTAMP_SIZE);
                cumValueSize += Math.max(currValueSize, 0);
                if (currValueSize >= 0) {
                    final byte[] value = new byte[currValueSize];
                    final int valueSegmentIndex = segmentValue.length - cumValueSize;
                    System.arraycopy(segmentValue, valueSegmentIndex, value, 0, currValueSize);
                    if (currTimestamp <= toTime && currNextTimestamp > fromTime) {
                        segmentSearchResults.add(new SegmentSearchResult(currIndex, currTimestamp, currNextTimestamp, value));
                    }
                }

                // prep for next iteration
                currNextTimestamp = currTimestamp;
                currIndex++;
            }
            return segmentSearchResults;
        }
This value will be null if the caller did not specify that the value should be
included with the return result from {@link SegmentValue#find(long, boolean)}.
In this case, it is up to the caller to not call this method, as the "value"
returned will not be meaningful.
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreSegmentValueFormatter.java