You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/08/10 06:48:32 UTC
(spark) branch master updated: [SPARK-53217][CORE][DSTREAM] Use Java `Set.of` instead of `Sets.newHashSet`
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e6e79b021ece [SPARK-53217][CORE][DSTREAM] Use Java `Set.of` instead of `Sets.newHashSet`
e6e79b021ece is described below
commit e6e79b021ece3532e9ff1fa7deb58b730d8d9088
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Sat Aug 9 23:48:23 2025 -0700
[SPARK-53217][CORE][DSTREAM] Use Java `Set.of` instead of `Sets.newHashSet`
### What changes were proposed in this pull request?
This PR aims to use Java 9+ `Set.of` API instead of `Sets.newHashSet`. In addition, we had better distinguish immutable and mutable sets clearly like Scala `val` and `var` usage.
- For immutable sets, `Sets.newHashSet(...)` -> `Set.of(...)`.
- For mutable sets, `Sets.newHashSet(...)` -> `new HashSet(Set.of(...))`
### Why are the changes needed?
`Set.of` is neat and simpler like the following code-change example.
```java
- Sets.newHashSet(1),
- Sets.newHashSet(2, 1),
- Sets.newHashSet(3, 2, 1),
- Sets.newHashSet(4, 3),
- Sets.newHashSet(5),
+ Set.of(1),
+ Set.of(2, 1),
+ Set.of(3, 2, 1),
+ Set.of(4, 3),
+ Set.of(5),
```
Note that Guava community also don't recommend `Sets.newHashSet` and plans to deprecate in the future.
- https://guava.dev/releases/33.4.6-jre/api/docs/com/google/common/collect/Sets.html#newHashSet(E...)
> This method is not actually very useful and will likely be deprecated in the future.
### Does this PR introduce _any_ user-facing change?
No behavior change.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #51942 from dongjoon-hyun/SPARK-53217.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../spark/network/ChunkFetchIntegrationSuite.java | 13 +++++----
.../apache/spark/network/RpcIntegrationSuite.java | 31 +++++++++++-----------
.../network/shuffle/ExternalBlockHandler.java | 4 +--
.../network/shuffle/RetryingBlockTransferor.java | 3 +--
.../shuffle/ExternalShuffleIntegrationSuite.java | 23 ++++++++--------
.../shuffle/RetryingBlockTransferorSuite.java | 3 +--
.../spark/streaming/JavaMapWithStateSuite.java | 28 +++++++++----------
.../org/apache/spark/streaming/Java8APISuite.java | 7 +++--
.../org/apache/spark/streaming/JavaAPISuite.java | 20 +++++++-------
9 files changed, 62 insertions(+), 70 deletions(-)
diff --git a/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java b/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java
index 576a106934fd..75ccd8d5789d 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/ChunkFetchIntegrationSuite.java
@@ -30,7 +30,6 @@ import java.util.Set;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import com.google.common.collect.Sets;
import com.google.common.io.Closeables;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -188,7 +187,7 @@ public class ChunkFetchIntegrationSuite {
@Test
public void fetchBufferChunk() throws Exception {
FetchResult res = fetchChunks(Arrays.asList(BUFFER_CHUNK_INDEX));
- assertEquals(Sets.newHashSet(BUFFER_CHUNK_INDEX), res.successChunks);
+ assertEquals(Set.of(BUFFER_CHUNK_INDEX), res.successChunks);
assertTrue(res.failedChunks.isEmpty());
assertBufferListsEqual(Arrays.asList(bufferChunk), res.buffers);
res.releaseBuffers();
@@ -197,7 +196,7 @@ public class ChunkFetchIntegrationSuite {
@Test
public void fetchFileChunk() throws Exception {
FetchResult res = fetchChunks(Arrays.asList(FILE_CHUNK_INDEX));
- assertEquals(Sets.newHashSet(FILE_CHUNK_INDEX), res.successChunks);
+ assertEquals(Set.of(FILE_CHUNK_INDEX), res.successChunks);
assertTrue(res.failedChunks.isEmpty());
assertBufferListsEqual(Arrays.asList(fileChunk), res.buffers);
res.releaseBuffers();
@@ -207,14 +206,14 @@ public class ChunkFetchIntegrationSuite {
public void fetchNonExistentChunk() throws Exception {
FetchResult res = fetchChunks(Arrays.asList(12345));
assertTrue(res.successChunks.isEmpty());
- assertEquals(Sets.newHashSet(12345), res.failedChunks);
+ assertEquals(Set.of(12345), res.failedChunks);
assertTrue(res.buffers.isEmpty());
}
@Test
public void fetchBothChunks() throws Exception {
FetchResult res = fetchChunks(Arrays.asList(BUFFER_CHUNK_INDEX, FILE_CHUNK_INDEX));
- assertEquals(Sets.newHashSet(BUFFER_CHUNK_INDEX, FILE_CHUNK_INDEX), res.successChunks);
+ assertEquals(Set.of(BUFFER_CHUNK_INDEX, FILE_CHUNK_INDEX), res.successChunks);
assertTrue(res.failedChunks.isEmpty());
assertBufferListsEqual(Arrays.asList(bufferChunk, fileChunk), res.buffers);
res.releaseBuffers();
@@ -223,8 +222,8 @@ public class ChunkFetchIntegrationSuite {
@Test
public void fetchChunkAndNonExistent() throws Exception {
FetchResult res = fetchChunks(Arrays.asList(BUFFER_CHUNK_INDEX, 12345));
- assertEquals(Sets.newHashSet(BUFFER_CHUNK_INDEX), res.successChunks);
- assertEquals(Sets.newHashSet(12345), res.failedChunks);
+ assertEquals(Set.of(BUFFER_CHUNK_INDEX), res.successChunks);
+ assertEquals(Set.of(12345), res.failedChunks);
assertBufferListsEqual(Arrays.asList(bufferChunk), res.buffers);
res.releaseBuffers();
}
diff --git a/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java b/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
index ae7788128b35..7fbb27f4ac1f 100644
--- a/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
+++ b/common/network-common/src/test/java/org/apache/spark/network/RpcIntegrationSuite.java
@@ -24,7 +24,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import com.google.common.collect.Sets;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -247,14 +246,14 @@ public class RpcIntegrationSuite {
@Test
public void singleRPC() throws Exception {
RpcResult res = sendRPC("hello/Aaron");
- assertEquals(Sets.newHashSet("Hello, Aaron!"), res.successMessages);
+ assertEquals(Set.of("Hello, Aaron!"), res.successMessages);
assertTrue(res.errorMessages.isEmpty());
}
@Test
public void doubleRPC() throws Exception {
RpcResult res = sendRPC("hello/Aaron", "hello/Reynold");
- assertEquals(Sets.newHashSet("Hello, Aaron!", "Hello, Reynold!"), res.successMessages);
+ assertEquals(Set.of("Hello, Aaron!", "Hello, Reynold!"), res.successMessages);
assertTrue(res.errorMessages.isEmpty());
}
@@ -262,28 +261,28 @@ public class RpcIntegrationSuite {
public void returnErrorRPC() throws Exception {
RpcResult res = sendRPC("return error/OK");
assertTrue(res.successMessages.isEmpty());
- assertErrorsContain(res.errorMessages, Sets.newHashSet("Returned: OK"));
+ assertErrorsContain(res.errorMessages, Set.of("Returned: OK"));
}
@Test
public void throwErrorRPC() throws Exception {
RpcResult res = sendRPC("throw error/uh-oh");
assertTrue(res.successMessages.isEmpty());
- assertErrorsContain(res.errorMessages, Sets.newHashSet("Thrown: uh-oh"));
+ assertErrorsContain(res.errorMessages, Set.of("Thrown: uh-oh"));
}
@Test
public void doubleTrouble() throws Exception {
RpcResult res = sendRPC("return error/OK", "throw error/uh-oh");
assertTrue(res.successMessages.isEmpty());
- assertErrorsContain(res.errorMessages, Sets.newHashSet("Returned: OK", "Thrown: uh-oh"));
+ assertErrorsContain(res.errorMessages, Set.of("Returned: OK", "Thrown: uh-oh"));
}
@Test
public void sendSuccessAndFailure() throws Exception {
RpcResult res = sendRPC("hello/Bob", "throw error/the", "hello/Builder", "return error/!");
- assertEquals(Sets.newHashSet("Hello, Bob!", "Hello, Builder!"), res.successMessages);
- assertErrorsContain(res.errorMessages, Sets.newHashSet("Thrown: the", "Returned: !"));
+ assertEquals(Set.of("Hello, Bob!", "Hello, Builder!"), res.successMessages);
+ assertErrorsContain(res.errorMessages, Set.of("Thrown: the", "Returned: !"));
}
@Test
@@ -310,7 +309,7 @@ public class RpcIntegrationSuite {
for (String stream : StreamTestHelper.STREAMS) {
RpcResult res = sendRpcWithStream(stream);
assertTrue(res.errorMessages.isEmpty(), "there were error messages!" + res.errorMessages);
- assertEquals(Sets.newHashSet(stream), res.successMessages);
+ assertEquals(Set.of(stream), res.successMessages);
}
}
@@ -321,7 +320,7 @@ public class RpcIntegrationSuite {
streams[i] = StreamTestHelper.STREAMS[i % StreamTestHelper.STREAMS.length];
}
RpcResult res = sendRpcWithStream(streams);
- assertEquals(Sets.newHashSet(StreamTestHelper.STREAMS), res.successMessages);
+ assertEquals(Set.of(StreamTestHelper.STREAMS), res.successMessages);
assertTrue(res.errorMessages.isEmpty());
}
@@ -341,8 +340,8 @@ public class RpcIntegrationSuite {
RpcResult exceptionInOnComplete =
sendRpcWithStream("fail/exception-oncomplete/smallBuffer", "smallBuffer");
assertErrorsContain(exceptionInOnComplete.errorMessages,
- Sets.newHashSet("Failure post-processing"));
- assertEquals(Sets.newHashSet("smallBuffer"), exceptionInOnComplete.successMessages);
+ Set.of("Failure post-processing"));
+ assertEquals(Set.of("smallBuffer"), exceptionInOnComplete.successMessages);
}
private void assertErrorsContain(Set<String> errors, Set<String> contains) {
@@ -364,14 +363,14 @@ public class RpcIntegrationSuite {
// We expect 1 additional error due to closed connection and here are possible keywords in the
// error message.
- Set<String> possibleClosedErrors = Sets.newHashSet(
+ Set<String> possibleClosedErrors = Set.of(
"closed",
"Connection reset",
"java.nio.channels.ClosedChannelException",
"io.netty.channel.StacklessClosedChannelException",
"java.io.IOException: Broken pipe"
);
- Set<String> containsAndClosed = Sets.newHashSet(expectedError);
+ Set<String> containsAndClosed = new HashSet<>(Set.of(expectedError));
containsAndClosed.addAll(possibleClosedErrors);
Pair<Set<String>, Set<String>> r = checkErrorsContain(errors, containsAndClosed);
@@ -391,8 +390,8 @@ public class RpcIntegrationSuite {
private Pair<Set<String>, Set<String>> checkErrorsContain(
Set<String> errors,
Set<String> contains) {
- Set<String> remainingErrors = Sets.newHashSet(errors);
- Set<String> notFound = Sets.newHashSet();
+ Set<String> remainingErrors = new HashSet<>(errors);
+ Set<String> notFound = new HashSet<>();
for (String contain : contains) {
Iterator<String> it = remainingErrors.iterator();
boolean foundMatch = false;
diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
index 64283fb6b2c2..45d0ff69de90 100644
--- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
+++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java
@@ -21,6 +21,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Objects;
@@ -36,7 +37,6 @@ import com.codahale.metrics.RatioGauge;
import com.codahale.metrics.Timer;
import com.codahale.metrics.Counter;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.collect.Sets;
import org.apache.spark.internal.SparkLogger;
import org.apache.spark.internal.SparkLoggerFactory;
@@ -199,7 +199,7 @@ public class ExternalBlockHandler extends RpcHandler
} else if (msgObj instanceof GetLocalDirsForExecutors msg) {
checkAuth(client, msg.appId);
- Set<String> execIdsForBlockResolver = Sets.newHashSet(msg.execIds);
+ Set<String> execIdsForBlockResolver = new HashSet<>(Set.of(msg.execIds));
boolean fetchMergedBlockDirs = execIdsForBlockResolver.remove(SHUFFLE_MERGER_IDENTIFIER);
Map<String, String[]> localDirs = blockManager.getLocalDirs(msg.appId,
execIdsForBlockResolver);
diff --git a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java
index 3d7b5bb18f9e..7f1629560071 100644
--- a/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java
+++ b/common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java
@@ -26,7 +26,6 @@ import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
-import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.spark.internal.SparkLogger;
@@ -131,7 +130,7 @@ public class RetryingBlockTransferor {
this.listener = listener;
this.maxRetries = conf.maxIORetries();
this.retryWaitTime = conf.ioRetryWaitTimeMs();
- this.outstandingBlocksIds = Sets.newLinkedHashSet();
+ this.outstandingBlocksIds = new LinkedHashSet<>();
Collections.addAll(outstandingBlocksIds, blockIds);
this.currentListener = new RetryingBlockTransferListener();
this.errorHandler = errorHandler;
diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java
index ec71f83ba743..59381cabe063 100644
--- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java
+++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/ExternalShuffleIntegrationSuite.java
@@ -32,7 +32,6 @@ import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
-import com.google.common.collect.Sets;
import org.apache.spark.network.buffer.FileSegmentManagedBuffer;
import org.apache.spark.network.server.OneForOneStreamManager;
import org.junit.jupiter.api.AfterAll;
@@ -222,7 +221,7 @@ public class ExternalShuffleIntegrationSuite {
try (ExternalBlockStoreClient client = createExternalBlockStoreClient()) {
registerExecutor(client, "exec-0", dataContext0.createExecutorInfo(SORT_MANAGER));
FetchResult exec0Fetch = fetchBlocks("exec-0", new String[] { "shuffle_0_0_0" });
- assertEquals(Sets.newHashSet("shuffle_0_0_0"), exec0Fetch.successBlocks);
+ assertEquals(Set.of("shuffle_0_0_0"), exec0Fetch.successBlocks);
assertTrue(exec0Fetch.failedBlocks.isEmpty());
assertBufferListsEqual(exec0Fetch.buffers, Arrays.asList(exec0Blocks[0]));
exec0Fetch.releaseBuffers();
@@ -235,7 +234,7 @@ public class ExternalShuffleIntegrationSuite {
registerExecutor(client,"exec-0", dataContext0.createExecutorInfo(SORT_MANAGER));
FetchResult exec0Fetch = fetchBlocks("exec-0",
new String[]{"shuffle_0_0_0", "shuffle_0_0_1", "shuffle_0_0_2"});
- assertEquals(Sets.newHashSet("shuffle_0_0_0", "shuffle_0_0_1", "shuffle_0_0_2"),
+ assertEquals(Set.of("shuffle_0_0_0", "shuffle_0_0_1", "shuffle_0_0_2"),
exec0Fetch.successBlocks);
assertTrue(exec0Fetch.failedBlocks.isEmpty());
assertBufferListsEqual(exec0Fetch.buffers, Arrays.asList(exec0Blocks));
@@ -256,7 +255,7 @@ public class ExternalShuffleIntegrationSuite {
registerExecutor(client, "exec-1", dataContext0.createExecutorInfo(SORT_MANAGER));
FetchResult execFetch = fetchBlocks("exec-1", new String[]{"broadcast_1"});
assertTrue(execFetch.successBlocks.isEmpty());
- assertEquals(Sets.newHashSet("broadcast_1"), execFetch.failedBlocks);
+ assertEquals(Set.of("broadcast_1"), execFetch.failedBlocks);
}
}
@@ -267,7 +266,7 @@ public class ExternalShuffleIntegrationSuite {
String validBlockId = "rdd_" + RDD_ID + "_" + SPLIT_INDEX_VALID_BLOCK;
FetchResult execFetch = fetchBlocks("exec-1", new String[]{validBlockId});
assertTrue(execFetch.failedBlocks.isEmpty());
- assertEquals(Sets.newHashSet(validBlockId), execFetch.successBlocks);
+ assertEquals(Set.of(validBlockId), execFetch.successBlocks);
assertBuffersEqual(new NioManagedBuffer(ByteBuffer.wrap(exec0RddBlockValid)),
execFetch.buffers.get(0));
}
@@ -280,7 +279,7 @@ public class ExternalShuffleIntegrationSuite {
String missingBlockId = "rdd_" + RDD_ID + "_" + SPLIT_INDEX_MISSING_FILE;
FetchResult execFetch = fetchBlocks("exec-1", new String[]{missingBlockId});
assertTrue(execFetch.successBlocks.isEmpty());
- assertEquals(Sets.newHashSet(missingBlockId), execFetch.failedBlocks);
+ assertEquals(Set.of(missingBlockId), execFetch.failedBlocks);
}
}
@@ -310,7 +309,7 @@ public class ExternalShuffleIntegrationSuite {
String corruptBlockId = "rdd_" + RDD_ID + "_" + SPLIT_INDEX_CORRUPT_LENGTH;
FetchResult execFetch = fetchBlocks("exec-1", new String[]{corruptBlockId});
assertTrue(execFetch.successBlocks.isEmpty());
- assertEquals(Sets.newHashSet(corruptBlockId), execFetch.failedBlocks);
+ assertEquals(Set.of(corruptBlockId), execFetch.failedBlocks);
}
}
@@ -321,7 +320,7 @@ public class ExternalShuffleIntegrationSuite {
FetchResult execFetch = fetchBlocks("exec-0",
new String[]{"shuffle_2_0_0"});
assertTrue(execFetch.successBlocks.isEmpty());
- assertEquals(Sets.newHashSet("shuffle_2_0_0"), execFetch.failedBlocks);
+ assertEquals(Set.of("shuffle_2_0_0"), execFetch.failedBlocks);
}
}
@@ -331,8 +330,8 @@ public class ExternalShuffleIntegrationSuite {
registerExecutor(client,"exec-0", dataContext0.createExecutorInfo(SORT_MANAGER));
FetchResult execFetch0 = fetchBlocks("exec-0", new String[]{"shuffle_0_0_0" /* right */});
FetchResult execFetch1 = fetchBlocks("exec-0", new String[]{"shuffle_1_0_0" /* wrong */});
- assertEquals(Sets.newHashSet("shuffle_0_0_0"), execFetch0.successBlocks);
- assertEquals(Sets.newHashSet("shuffle_1_0_0"), execFetch1.failedBlocks);
+ assertEquals(Set.of("shuffle_0_0_0"), execFetch0.successBlocks);
+ assertEquals(Set.of("shuffle_1_0_0"), execFetch1.failedBlocks);
}
}
@@ -343,7 +342,7 @@ public class ExternalShuffleIntegrationSuite {
FetchResult execFetch = fetchBlocks("exec-2",
new String[]{"shuffle_0_0_0", "shuffle_1_0_0"});
assertTrue(execFetch.successBlocks.isEmpty());
- assertEquals(Sets.newHashSet("shuffle_0_0_0", "shuffle_1_0_0"), execFetch.failedBlocks);
+ assertEquals(Set.of("shuffle_0_0_0", "shuffle_1_0_0"), execFetch.failedBlocks);
}
}
@@ -355,7 +354,7 @@ public class ExternalShuffleIntegrationSuite {
FetchResult execFetch = fetchBlocks("exec-0",
new String[]{"shuffle_1_0_0", "shuffle_1_0_1"}, clientConf, 1 /* port */);
assertTrue(execFetch.successBlocks.isEmpty());
- assertEquals(Sets.newHashSet("shuffle_1_0_0", "shuffle_1_0_1"), execFetch.failedBlocks);
+ assertEquals(Set.of("shuffle_1_0_0", "shuffle_1_0_1"), execFetch.failedBlocks);
}
}
diff --git a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockTransferorSuite.java b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockTransferorSuite.java
index 84c8b1b3353f..d8fecc063989 100644
--- a/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockTransferorSuite.java
+++ b/common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockTransferorSuite.java
@@ -29,7 +29,6 @@ import java.util.Map;
import java.util.concurrent.TimeoutException;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Sets;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.stubbing.Answer;
@@ -425,7 +424,7 @@ public class RetryingBlockTransferorSuite {
Stubber stub = null;
// Contains all blockIds that are referenced across all interactions.
- LinkedHashSet<String> blockIds = Sets.newLinkedHashSet();
+ LinkedHashSet<String> blockIds = new LinkedHashSet<>();
for (Map<String, Object> interaction : interactions) {
blockIds.addAll(interaction.keySet());
diff --git a/streaming/src/test/java/org/apache/spark/streaming/JavaMapWithStateSuite.java b/streaming/src/test/java/org/apache/spark/streaming/JavaMapWithStateSuite.java
index 8aaafc3d3d5e..9874385761e0 100644
--- a/streaming/src/test/java/org/apache/spark/streaming/JavaMapWithStateSuite.java
+++ b/streaming/src/test/java/org/apache/spark/streaming/JavaMapWithStateSuite.java
@@ -21,12 +21,12 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
import scala.Tuple2;
-import com.google.common.collect.Sets;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.util.ManualClock;
import org.junit.jupiter.api.Assertions;
@@ -106,23 +106,23 @@ public class JavaMapWithStateSuite extends LocalJavaStreamingContext implements
List<Set<Integer>> outputData = Arrays.asList(
Collections.emptySet(),
- Sets.newHashSet(1),
- Sets.newHashSet(2, 1),
- Sets.newHashSet(3, 2, 1),
- Sets.newHashSet(4, 3),
- Sets.newHashSet(5),
+ Set.of(1),
+ Set.of(2, 1),
+ Set.of(3, 2, 1),
+ Set.of(4, 3),
+ Set.of(5),
Collections.emptySet()
);
@SuppressWarnings("unchecked")
List<Set<Tuple2<String, Integer>>> stateData = Arrays.asList(
Collections.emptySet(),
- Sets.newHashSet(new Tuple2<>("a", 1)),
- Sets.newHashSet(new Tuple2<>("a", 2), new Tuple2<>("b", 1)),
- Sets.newHashSet(new Tuple2<>("a", 3), new Tuple2<>("b", 2), new Tuple2<>("c", 1)),
- Sets.newHashSet(new Tuple2<>("a", 4), new Tuple2<>("b", 3), new Tuple2<>("c", 1)),
- Sets.newHashSet(new Tuple2<>("a", 5), new Tuple2<>("b", 3), new Tuple2<>("c", 1)),
- Sets.newHashSet(new Tuple2<>("a", 5), new Tuple2<>("b", 3), new Tuple2<>("c", 1))
+ Set.of(new Tuple2<>("a", 1)),
+ Set.of(new Tuple2<>("a", 2), new Tuple2<>("b", 1)),
+ Set.of(new Tuple2<>("a", 3), new Tuple2<>("b", 2), new Tuple2<>("c", 1)),
+ Set.of(new Tuple2<>("a", 4), new Tuple2<>("b", 3), new Tuple2<>("c", 1)),
+ Set.of(new Tuple2<>("a", 5), new Tuple2<>("b", 3), new Tuple2<>("c", 1)),
+ Set.of(new Tuple2<>("a", 5), new Tuple2<>("b", 3), new Tuple2<>("c", 1))
);
Function3<String, Optional<Integer>, State<Integer>, Integer> mappingFunc =
@@ -150,11 +150,11 @@ public class JavaMapWithStateSuite extends LocalJavaStreamingContext implements
List<Set<T>> collectedOutputs =
Collections.synchronizedList(new ArrayList<>());
- mapWithStateDStream.foreachRDD(rdd -> collectedOutputs.add(Sets.newHashSet(rdd.collect())));
+ mapWithStateDStream.foreachRDD(rdd -> collectedOutputs.add(new HashSet<>(rdd.collect())));
List<Set<Tuple2<K, S>>> collectedStateSnapshots =
Collections.synchronizedList(new ArrayList<>());
mapWithStateDStream.stateSnapshots().foreachRDD(rdd ->
- collectedStateSnapshots.add(Sets.newHashSet(rdd.collect())));
+ collectedStateSnapshots.add(new HashSet<>(rdd.collect())));
BatchCounter batchCounter = new BatchCounter(ssc.ssc());
ssc.start();
((ManualClock) ssc.ssc().scheduler().clock())
diff --git a/streaming/src/test/java/test/org/apache/spark/streaming/Java8APISuite.java b/streaming/src/test/java/test/org/apache/spark/streaming/Java8APISuite.java
index 00b1fee76453..82ad517fefc2 100644
--- a/streaming/src/test/java/test/org/apache/spark/streaming/Java8APISuite.java
+++ b/streaming/src/test/java/test/org/apache/spark/streaming/Java8APISuite.java
@@ -31,7 +31,6 @@ import org.apache.spark.streaming.StateSpec;
import org.apache.spark.streaming.Time;
import scala.Tuple2;
-import com.google.common.collect.Sets;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -218,12 +217,12 @@ public class Java8APISuite extends LocalJavaStreamingContext implements Serializ
List<Set<Tuple2<String, Tuple2<String, String>>>> expected = Arrays.asList(
- Sets.newHashSet(
+ Set.of(
new Tuple2<>("california",
new Tuple2<>("dodgers", "giants")),
new Tuple2<>("new york",
new Tuple2<>("yankees", "mets"))),
- Sets.newHashSet(
+ Set.of(
new Tuple2<>("california",
new Tuple2<>("sharks", "ducks")),
new Tuple2<>("new york",
@@ -244,7 +243,7 @@ public class Java8APISuite extends LocalJavaStreamingContext implements Serializ
List<List<Tuple2<String, Tuple2<String, String>>>> result = JavaTestUtils.runStreams(ssc, 2, 2);
List<Set<Tuple2<String, Tuple2<String, String>>>> unorderedResult = new ArrayList<>();
for (List<Tuple2<String, Tuple2<String, String>>> res : result) {
- unorderedResult.add(Sets.newHashSet(res));
+ unorderedResult.add(new HashSet<>(res));
}
Assertions.assertEquals(expected, unorderedResult);
diff --git a/streaming/src/test/java/test/org/apache/spark/streaming/JavaAPISuite.java b/streaming/src/test/java/test/org/apache/spark/streaming/JavaAPISuite.java
index ce37b375ad21..9753efecc0a5 100644
--- a/streaming/src/test/java/test/org/apache/spark/streaming/JavaAPISuite.java
+++ b/streaming/src/test/java/test/org/apache/spark/streaming/JavaAPISuite.java
@@ -40,8 +40,6 @@ import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import com.google.common.collect.Sets;
-
import org.apache.spark.HashPartitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
@@ -442,13 +440,13 @@ public class JavaAPISuite extends LocalJavaStreamingContext implements Serializa
new Tuple2<>("new york", "islanders")));
- List<HashSet<Tuple2<String, Tuple2<String, String>>>> expected = Arrays.asList(
- Sets.newHashSet(
+ List<Set<Tuple2<String, Tuple2<String, String>>>> expected = Arrays.asList(
+ Set.of(
new Tuple2<>("california",
new Tuple2<>("dodgers", "giants")),
new Tuple2<>("new york",
new Tuple2<>("yankees", "mets"))),
- Sets.newHashSet(
+ Set.of(
new Tuple2<>("california",
new Tuple2<>("sharks", "ducks")),
new Tuple2<>("new york",
@@ -471,7 +469,7 @@ public class JavaAPISuite extends LocalJavaStreamingContext implements Serializa
List<List<Tuple2<String, Tuple2<String, String>>>> result = JavaTestUtils.runStreams(ssc, 2, 2);
List<HashSet<Tuple2<String, Tuple2<String, String>>>> unorderedResult = new ArrayList<>();
for (List<Tuple2<String, Tuple2<String, String>>> res: result) {
- unorderedResult.add(Sets.newHashSet(res));
+ unorderedResult.add(new HashSet<>(res));
}
Assertions.assertEquals(expected, unorderedResult);
@@ -1161,15 +1159,15 @@ public class JavaAPISuite extends LocalJavaStreamingContext implements Serializa
Arrays.asList("hello", "moon"),
Arrays.asList("hello"));
- List<HashSet<Tuple2<String, Long>>> expected = Arrays.asList(
- Sets.newHashSet(
+ List<Set<Tuple2<String, Long>>> expected = Arrays.asList(
+ Set.of(
new Tuple2<>("hello", 1L),
new Tuple2<>("world", 1L)),
- Sets.newHashSet(
+ Set.of(
new Tuple2<>("hello", 2L),
new Tuple2<>("world", 1L),
new Tuple2<>("moon", 1L)),
- Sets.newHashSet(
+ Set.of(
new Tuple2<>("hello", 2L),
new Tuple2<>("moon", 1L)));
@@ -1181,7 +1179,7 @@ public class JavaAPISuite extends LocalJavaStreamingContext implements Serializa
List<List<Tuple2<String, Long>>> result = JavaTestUtils.runStreams(ssc, 3, 3);
List<Set<Tuple2<String, Long>>> unorderedResult = new ArrayList<>();
for (List<Tuple2<String, Long>> res: result) {
- unorderedResult.add(Sets.newHashSet(res));
+ unorderedResult.add(new HashSet<>(res));
}
Assertions.assertEquals(expected, unorderedResult);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org