Pull 4.2-rc4 changes

LICENSE-binary

@@ -242,7 +242,7 @@ io.dropwizard.metrics:metrics-json
 io.dropwizard.metrics:metrics-jvm
 io.fabric8:kubernetes-client
 io.fabric8:kubernetes-client-api
-io.fabric8:kubernetes-httpclient-okhttp
+io.fabric8:kubernetes-httpclient-vertx
 io.fabric8:kubernetes-model-admissionregistration
 io.fabric8:kubernetes-model-apiextensions
 io.fabric8:kubernetes-model-apps
@@ -305,6 +305,7 @@ joda-time:joda-time
 net.sf.opencsv:opencsv
 net.sf.supercsv:super-csv
 net.sf.jpam:jpam
+org.apache.arrow:arrow-compression
 org.apache.arrow:arrow-format
 org.apache.arrow:arrow-memory-core
 org.apache.arrow:arrow-memory-netty
@@ -349,7 +350,7 @@ org.apache.logging.log4j:log4j-1.2-api
 org.apache.logging.log4j:log4j-api
 org.apache.logging.log4j:log4j-core
 org.apache.logging.log4j:log4j-layout-template-json
-org.apache.logging.log4j:log4j-slf4j-impl
+org.apache.logging.log4j:log4j-slf4j2-impl
 org.apache.orc:orc-core
 org.apache.orc:orc-format
 org.apache.orc:orc-mapreduce
@@ -510,6 +511,7 @@ javax.transaction:transaction-api
 Common Development and Distribution License (CDDL) 1.1
 ------------------------------------------------------
 javax.transaction:jta http://www.oracle.com/technetwork/java/index.html
+javax.servlet:javax.servlet-api https://oss.oracle.com/licenses/CDDL+GPL-1.1
 javax.xml.bind:jaxb-api    https://github.com/javaee/jaxb-v2
 
 

R/pkg/DESCRIPTION

@@ -1,6 +1,6 @@
 Package: SparkR
 Type: Package
-Version: 4.2.0.1-4.3.0-1
+Version: 4.2.0.1-4.3.0-2
 Title: R Front End for 'Apache Spark'
 Description: Provides an R Front end for 'Apache Spark' <https://spark.apache.org>.
 Authors@R:

assembly/pom.xml

@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.2.0.1-4.3.0-1</version>
+    <version>4.2.0.1-4.3.0-2</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

common/kvstore/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.2.0.1-4.3.0-1</version>
+    <version>4.2.0.1-4.3.0-2</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/kvstore/src/main/java/org/apache/spark/util/kvstore/LevelDBIterator.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.lang.ref.Cleaner;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
@@ -289,17 +290,10 @@ private boolean isEndMarker(byte[] key) {
         key[key.length - 1] == LevelDBTypeInfo.END_MARKER[0]);
   }
 
+  @VisibleForTesting
   static int compare(byte[] a, byte[] b) {
-    int diff = 0;
-    int minLen = Math.min(a.length, b.length);
-    for (int i = 0; i < minLen; i++) {
-      diff += (a[i] - b[i]);
-      if (diff != 0) {
-        return diff;
-      }
-    }
-
-    return a.length - b.length;
+    // Unsigned bytewise comparison, matching LevelDB's key ordering.
+    return Arrays.compareUnsigned(a, b);
   }
 
   static class ResourceCleaner implements Runnable {

common/kvstore/src/main/java/org/apache/spark/util/kvstore/RocksDBIterator.java

@@ -278,17 +278,10 @@ private boolean isEndMarker(byte[] key) {
         key[key.length - 1] == RocksDBTypeInfo.END_MARKER[0]);
   }
 
+  @VisibleForTesting
   static int compare(byte[] a, byte[] b) {
-    int diff = 0;
-    int minLen = Math.min(a.length, b.length);
-    for (int i = 0; i < minLen; i++) {
-      diff += (a[i] - b[i]);
-      if (diff != 0) {
-        return diff;
-      }
-    }
-
-    return a.length - b.length;
+    // Unsigned bytewise comparison, matching RocksDB's key ordering.
+    return Arrays.compareUnsigned(a, b);
   }
 
   static class ResourceCleaner implements Runnable {

common/kvstore/src/test/java/org/apache/spark/util/kvstore/DBIteratorCompareSuite.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.util.kvstore;
+
+import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Unit tests for the static byte-array comparator that LevelDBIterator and RocksDBIterator
+ * each define (the two backends are kept independent, so the method is duplicated). The
+ * comparator is pure and needs no database, so these tests run on every platform.
+ */
+public class DBIteratorCompareSuite {
+
+  @FunctionalInterface
+  private interface ByteArrayComparator {
+    int compare(byte[] a, byte[] b);
+  }
+
+  @Test
+  public void testLevelDBIteratorCompare() {
+    checkUnsignedByteOrdering(LevelDBIterator::compare);
+  }
+
+  @Test
+  public void testRocksDBIteratorCompare() {
+    checkUnsignedByteOrdering(RocksDBIterator::compare);
+  }
+
+  private static void checkUnsignedByteOrdering(ByteArrayComparator cmp) {
+    // Equal arrays compare equal.
+    assertEquals(0, cmp.compare(new byte[] { 1, 2, 3 }, new byte[] { 1, 2, 3 }));
+
+    // Empty arrays compare equal, and an empty array sorts before any non-empty one.
+    assertEquals(0, cmp.compare(new byte[] {}, new byte[] {}));
+    assertTrue(cmp.compare(new byte[] {}, new byte[] { 1 }) < 0);
+    assertTrue(cmp.compare(new byte[] { 1 }, new byte[] {}) > 0);
+
+    // A prefix sorts before the longer array that extends it.
+    assertTrue(cmp.compare(new byte[] { 1, 2 }, new byte[] { 1, 2, 3 }) < 0);
+    assertTrue(cmp.compare(new byte[] { 1, 2, 3 }, new byte[] { 1, 2 }) > 0);
+
+    // Bytes must be ordered as unsigned, matching the underlying key ordering: 0x80 (128) is
+    // greater than 0x7f (127). A signed comparison would wrongly treat 0x80 as -128.
+    byte[] highBit = new byte[] { (byte) 0x80 };
+    byte[] lowBit = new byte[] { 0x7f };
+    assertTrue(cmp.compare(highBit, lowBit) > 0);
+    assertTrue(cmp.compare(lowBit, highBit) < 0);
+
+    // 0xff (255) is the largest byte value when unsigned, so it sorts after 0x00.
+    assertTrue(cmp.compare(new byte[] { (byte) 0xff }, new byte[] { 0x00 }) > 0);
+
+    // The first differing byte decides the order, regardless of later bytes.
+    assertTrue(cmp.compare(new byte[] { 0x01, (byte) 0xff }, new byte[] { 0x02, 0x00 }) < 0);
+  }
+
+}

common/network-common/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.2.0.1-4.3.0-1</version>
+    <version>4.2.0.1-4.3.0-2</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

common/network-common/src/main/java/org/apache/spark/network/protocol/Encoders.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.Objects;
 
 import io.netty.buffer.ByteBuf;
 import org.roaringbitmap.RoaringBitmap;
@@ -41,6 +42,7 @@ public static void encode(ByteBuf buf, String s) {
 
     public static String decode(ByteBuf buf) {
       int length = buf.readInt();
+      Objects.checkFromIndexSize(0, length, buf.readableBytes());
       byte[] bytes = new byte[length];
       buf.readBytes(bytes);
       return new String(bytes, StandardCharsets.UTF_8);
@@ -105,6 +107,7 @@ public static void encode(ByteBuf buf, byte[] arr) {
 
     public static byte[] decode(ByteBuf buf) {
       int length = buf.readInt();
+      Objects.checkFromIndexSize(0, length, buf.readableBytes());
       byte[] bytes = new byte[length];
       buf.readBytes(bytes);
       return bytes;
@@ -130,6 +133,7 @@ public static void encode(ByteBuf buf, String[] strings) {
 
     public static String[] decode(ByteBuf buf) {
       int numStrings = buf.readInt();
+      Objects.checkFromIndexSize(0, numStrings, buf.readableBytes() / 4);
       String[] strings = new String[numStrings];
       for (int i = 0; i < strings.length; i ++) {
         strings[i] = Strings.decode(buf);
@@ -153,6 +157,7 @@ public static void encode(ByteBuf buf, int[] ints) {
 
     public static int[] decode(ByteBuf buf) {
       int numInts = buf.readInt();
+      Objects.checkFromIndexSize(0, numInts, buf.readableBytes() / 4);
       int[] ints = new int[numInts];
       for (int i = 0; i < ints.length; i ++) {
         ints[i] = buf.readInt();
@@ -176,6 +181,7 @@ public static void encode(ByteBuf buf, long[] longs) {
 
     public static long[] decode(ByteBuf buf) {
       int numLongs = buf.readInt();
+      Objects.checkFromIndexSize(0, numLongs, buf.readableBytes() / 8);
       long[] longs = new long[numLongs];
       for (int i = 0; i < longs.length; i ++) {
         longs[i] = buf.readLong();
@@ -207,6 +213,9 @@ public static void encode(ByteBuf buf, RoaringBitmap[] bitmaps) {
 
     public static RoaringBitmap[] decode(ByteBuf buf) {
       int numBitmaps = buf.readInt();
+      // The divisor 8 is the minimum on-wire size of one element, since an empty RoaringBitmap
+      // serializes to 8 bytes (a 4-byte cookie followed by a 4-byte size).
+      Objects.checkFromIndexSize(0, numBitmaps, buf.readableBytes() / 8);
       RoaringBitmap[] bitmaps = new RoaringBitmap[numBitmaps];
       for (int i = 0; i < bitmaps.length; i ++) {
         bitmaps[i] = Bitmaps.decode(buf);

common/network-common/src/test/java/org/apache/spark/network/protocol/EncodersSuite.java

@@ -48,6 +48,28 @@ public void testRoaringBitmapEncodeShouldFailWhenBufferIsSmall() {
       () -> Encoders.Bitmaps.encode(buf, bitmap));
   }
 
+  @Test
+  public void testStringsEncodeDecode() {
+    String s = "spark";
+    ByteBuf buf = Unpooled.buffer(Encoders.Strings.encodedLength(s));
+    Encoders.Strings.encode(buf, s);
+    assertEquals(s, Encoders.Strings.decode(buf));
+  }
+
+  @Test
+  public void testStringsDecodeShouldFailWhenLengthIsNegative() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(-1);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.Strings.decode(buf));
+  }
+
+  @Test
+  public void testStringsDecodeShouldFailWhenLengthExceedsReadableBytes() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(Integer.MAX_VALUE);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.Strings.decode(buf));
+  }
+
   @Test
   public void testBitmapArraysEncodeDecode() {
     RoaringBitmap[] bitmaps = new RoaringBitmap[] {
@@ -66,4 +88,106 @@ public void testBitmapArraysEncodeDecode() {
     RoaringBitmap[] decodedBitmaps = Encoders.BitmapArrays.decode(buf);
     assertArrayEquals(bitmaps, decodedBitmaps);
   }
+
+  @Test
+  public void testBitmapArraysDecodeShouldFailWhenLengthIsNegative() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(-1);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.BitmapArrays.decode(buf));
+  }
+
+  @Test
+  public void testBitmapArraysDecodeShouldFailWhenLengthExceedsReadableBytes() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(Integer.MAX_VALUE);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.BitmapArrays.decode(buf));
+  }
+
+  @Test
+  public void testByteArraysEncodeDecode() {
+    byte[] arr = new byte[] { 1, 2, 3, 4, 5 };
+    ByteBuf buf = Unpooled.buffer(Encoders.ByteArrays.encodedLength(arr));
+    Encoders.ByteArrays.encode(buf, arr);
+    assertArrayEquals(arr, Encoders.ByteArrays.decode(buf));
+  }
+
+  @Test
+  public void testByteArraysDecodeShouldFailWhenLengthIsNegative() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(-1);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.ByteArrays.decode(buf));
+  }
+
+  @Test
+  public void testByteArraysDecodeShouldFailWhenLengthExceedsReadableBytes() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(Integer.MAX_VALUE);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.ByteArrays.decode(buf));
+  }
+
+  @Test
+  public void testIntArraysEncodeDecode() {
+    int[] arr = new int[] { 1, 2, 3, 4, 5 };
+    ByteBuf buf = Unpooled.buffer(Encoders.IntArrays.encodedLength(arr));
+    Encoders.IntArrays.encode(buf, arr);
+    assertArrayEquals(arr, Encoders.IntArrays.decode(buf));
+  }
+
+  @Test
+  public void testIntArraysDecodeShouldFailWhenLengthIsNegative() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(-1);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.IntArrays.decode(buf));
+  }
+
+  @Test
+  public void testIntArraysDecodeShouldFailWhenLengthExceedsReadableBytes() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(Integer.MAX_VALUE);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.IntArrays.decode(buf));
+  }
+
+  @Test
+  public void testLongArraysEncodeDecode() {
+    long[] arr = new long[] { 1L, 2L, 3L, 4L, 5L };
+    ByteBuf buf = Unpooled.buffer(Encoders.LongArrays.encodedLength(arr));
+    Encoders.LongArrays.encode(buf, arr);
+    assertArrayEquals(arr, Encoders.LongArrays.decode(buf));
+  }
+
+  @Test
+  public void testLongArraysDecodeShouldFailWhenLengthIsNegative() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(-1);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.LongArrays.decode(buf));
+  }
+
+  @Test
+  public void testLongArraysDecodeShouldFailWhenLengthExceedsReadableBytes() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(Integer.MAX_VALUE);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.LongArrays.decode(buf));
+  }
+
+  @Test
+  public void testStringArraysEncodeDecode() {
+    String[] arr = new String[] { "spark", "", "rocks" };
+    ByteBuf buf = Unpooled.buffer(Encoders.StringArrays.encodedLength(arr));
+    Encoders.StringArrays.encode(buf, arr);
+    assertArrayEquals(arr, Encoders.StringArrays.decode(buf));
+  }
+
+  @Test
+  public void testStringArraysDecodeShouldFailWhenLengthIsNegative() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(-1);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.StringArrays.decode(buf));
+  }
+
+  @Test
+  public void testStringArraysDecodeShouldFailWhenLengthExceedsReadableBytes() {
+    ByteBuf buf = Unpooled.buffer();
+    buf.writeInt(Integer.MAX_VALUE);
+    assertThrows(IndexOutOfBoundsException.class, () -> Encoders.StringArrays.decode(buf));
+  }
 }

common/network-shuffle/pom.xml

@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.2.0.1-4.3.0-1</version>
+    <version>4.2.0.1-4.3.0-2</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>