diff --git a/java/vortex-spark/common/src/main/java/dev/vortex/spark/io/VortexIo.java b/java/vortex-spark/common/src/main/java/dev/vortex/spark/io/VortexIo.java index 48d4ac920c7..25e494673dd 100644 --- a/java/vortex-spark/common/src/main/java/dev/vortex/spark/io/VortexIo.java +++ b/java/vortex-spark/common/src/main/java/dev/vortex/spark/io/VortexIo.java @@ -9,6 +9,7 @@ import java.util.Locale; import java.util.Objects; import org.apache.hadoop.conf.Configuration; +import org.apache.spark.SerializableWritable; import org.apache.spark.SparkContext; import org.apache.spark.broadcast.Broadcast; import scala.reflect.ClassTag; @@ -48,8 +49,14 @@ public final class VortexIo implements Serializable { /** The configuration this instance carries itself, or null when it reads {@link #broadcastConf}. */ private final SerializableHadoopConf conf; - /** The configuration broadcast to every executor once, or null when this instance carries {@link #conf}. */ - private final Broadcast broadcastConf; + /** + * The configuration broadcast to every executor once, or null when this instance carries {@link #conf}. + * + *

A broadcast value goes through {@code spark.serializer}, not Java serialization. Kryo skips + * {@link SerializableHadoopConf}'s transient field and would hand executors a null configuration; Spark's Kryo + * setup serializes {@link SerializableWritable} with Java serialization, so it arrives intact under either. + */ + private final Broadcast> broadcastConf; private final int readConcurrency; private final int splitConcurrency; @@ -72,10 +79,9 @@ public static VortexIo create(VortexOptions options, Configuration hadoopConf) { public static VortexIo broadcast(VortexOptions options, Configuration hadoopConf, SparkContext context) { Objects.requireNonNull(options, "options"); Objects.requireNonNull(hadoopConf, "hadoopConf"); - ClassTag tag = ClassTag$.MODULE$.apply(SerializableHadoopConf.class); return new VortexIo( null, - context.broadcast(new SerializableHadoopConf(hadoopConf), tag), + context.broadcast(broadcastValue(hadoopConf), BROADCAST_TAG), nonNegative(options, READ_CONCURRENCY_OPTION, DEFAULT_READ_CONCURRENCY), nonNegative(options, SPLIT_CONCURRENCY_OPTION, DEFAULT_SPLIT_CONCURRENCY)); } @@ -86,9 +92,18 @@ public static VortexIo defaults() { new SerializableHadoopConf(new Configuration()), null, DEFAULT_READ_CONCURRENCY, DEFAULT_SPLIT_CONCURRENCY); } + @SuppressWarnings("unchecked") + private static final ClassTag> BROADCAST_TAG = + (ClassTag>) (ClassTag) ClassTag$.MODULE$.apply(SerializableWritable.class); + + /** What {@link #broadcast} sends to executors for {@code hadoopConf}. */ + static SerializableWritable broadcastValue(Configuration hadoopConf) { + return new SerializableWritable<>(hadoopConf); + } + private VortexIo( SerializableHadoopConf conf, - Broadcast broadcastConf, + Broadcast> broadcastConf, int readConcurrency, int splitConcurrency) { this.conf = conf; diff --git a/java/vortex-spark/common/src/test/java/dev/vortex/spark/io/VortexIoTest.java b/java/vortex-spark/common/src/test/java/dev/vortex/spark/io/VortexIoTest.java index 019f811fca0..df7d8090d47 100644 --- a/java/vortex-spark/common/src/test/java/dev/vortex/spark/io/VortexIoTest.java +++ b/java/vortex-spark/common/src/test/java/dev/vortex/spark/io/VortexIoTest.java @@ -14,7 +14,16 @@ import java.io.ObjectOutputStream; import java.util.Map; import org.apache.hadoop.conf.Configuration; +import org.apache.spark.SerializableWritable; +import org.apache.spark.SparkConf; +import org.apache.spark.serializer.JavaSerializer; +import org.apache.spark.serializer.KryoSerializer; +import org.apache.spark.serializer.SerializerInstance; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import scala.reflect.ClassTag; +import scala.reflect.ClassTag$; final class VortexIoTest { @Test @@ -89,6 +98,25 @@ private static VortexOptions options(String key, String value) { return VortexOptions.of(Map.of(key, value)); } + /** + * A broadcast reaches remote executors through {@code spark.serializer}. Local mode hands tasks the driver's object + * without serializing it, so only a round trip through each serializer shows what an executor gets. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void broadcastConfigurationSurvivesEachSparkSerializer(boolean kryo) { + Configuration conf = new Configuration(false); + conf.set("fs.s3a.endpoint", "https://storage.example"); + SerializerInstance serializer = (kryo ? new KryoSerializer(new SparkConf()) : new JavaSerializer(new SparkConf())) + .newInstance(); + ClassTag tag = ClassTag$.MODULE$.apply(Object.class); + + SerializableWritable shipped = (SerializableWritable) + serializer.deserialize(serializer.serialize(VortexIo.broadcastValue(conf), tag), tag); + + assertEquals("https://storage.example", ((Configuration) shipped.value()).get("fs.s3a.endpoint")); + } + private static VortexIo roundTrip(VortexIo io) throws IOException, ClassNotFoundException { ByteArrayOutputStream bytes = new ByteArrayOutputStream(); try (ObjectOutputStream out = new ObjectOutputStream(bytes)) {