KAFKA-13264: fix inMemoryWindowStore backward fetch not in reversed o…

…rder (#11292)

When introducing backward iterator for WindowStroe in #9138, we forgot to make "each segment" in reverse order (i.e. in descendingMap) in InMemoryWindowStore. Fix it and add integration tests for it.

Currently, in Window store, we store records in [segments -> [records] ].

For example:
window size = 500,
input records:

key: "a", value: "aa", timestamp: 0 ==> will be in [0, 500] window
key: "b", value: "bb", timestamp: 10 ==> will be in [0, 500] window
key: "c", value: "cc", timestamp: 510 ==> will be in [500, 1000] window

So, internally, the "a" and "b" will be in the same segment, and "c" in another segments.
segments: [0 /* window start */, records], [500, records].
And the records for window start 0 will be "a" and "b".
the records for window start 500 will be "c".

Before this change, we did have a reverse iterator for segments, but not in "records". So, when doing backwardFetchAll, we'll have the records returned in order: "c", "a", "b", which should be "c", "b", "a" obviously.

Reviewers: Jorge Esteban Quilcate Otoya <quilcate.jorge@gmail.com>, Anna Sophie Blee-Goldman <ableegoldman@apache.org>, Guozhang Wang <wangguoz@gmail.com>

streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java

@@ -499,10 +499,14 @@ Iterator<Map.Entry<Bytes, byte[]>> setRecordIterator() {
             final Map.Entry<Long, ConcurrentNavigableMap<Bytes, byte[]>> currentSegment = segmentIterator.next();
             currentTime = currentSegment.getKey();
 
-            if (allKeys) {
-                return currentSegment.getValue().entrySet().iterator();
+            final ConcurrentNavigableMap<Bytes, byte[]> subMap = allKeys ?
+                currentSegment.getValue() :
+                currentSegment.getValue().subMap(keyFrom, true, keyTo, true);
+
+            if (forward) {
+                return subMap.entrySet().iterator();
             } else {
-                return currentSegment.getValue().subMap(keyFrom, true, keyTo, true).entrySet().iterator();
+                return subMap.descendingMap().entrySet().iterator();
             }
         }