Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[SPARK-44259][CONNECT][TESTS][FOLLOWUP] No longer initializing Ammonite for Java 21 #41814

Closed
wants to merge 6 commits into from
Closed

[SPARK-44259][CONNECT][TESTS][FOLLOWUP] No longer initializing Ammonite for Java 21 #41814

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
fc44bca
test java 21 connect
LuciferYang Jul 2, 2023
1ed1b8c
streaming
LuciferYang Jul 2, 2023
bae61af
Merge branch 'upmaster' into SPARK-44259-FOLLOWUP
LuciferYang Jul 3, 2023
503e216
revert build_and_test.yml
LuciferYang Jul 3, 2023
5f6abe9
Merge branch 'apache:master' into SPARK-44259-FOLLOWUP
LuciferYang Jul 5, 2023
d0a0a6a
for ga
LuciferYang Jul 5, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/build_and_test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ on:
java:
required: false
type: string
default: 8
default: 21-ea
LuciferYang marked this conversation as resolved.
Show resolved Hide resolved
branch:
description: Branch to run the build against
required: false
Expand Down
40 changes: 22 additions & 18 deletions ...tor/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import java.io.{PipedInputStream, PipedOutputStream}
import java.util.concurrent.{Executors, Semaphore, TimeUnit}

import org.apache.commons.io.output.ByteArrayOutputStream
import org.apache.commons.lang3.{JavaVersion, SystemUtils}
import org.scalatest.BeforeAndAfterEach

import org.apache.spark.sql.connect.client.util.RemoteSparkSession
Expand All @@ -42,26 +43,29 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach {
}

override def beforeAll(): Unit = {
super.beforeAll()
ammoniteOut = new ByteArrayOutputStream()
testSuiteOut = new PipedOutputStream()
// Connect the `testSuiteOut` and `ammoniteIn` pipes
ammoniteIn = new PipedInputStream(testSuiteOut)
errorStream = new ByteArrayOutputStream()

val args = Array("--port", serverPort.toString)
val task = new Runnable {
override def run(): Unit = {
ConnectRepl.doMain(
args = args,
semaphore = Some(semaphore),
inputStream = ammoniteIn,
outputStream = ammoniteOut,
errorStream = errorStream)
// TODO(SPARK-44121) Remove this check condition
if (SystemUtils.isJavaVersionAtMost(JavaVersion.JAVA_17)) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, how does this test work with 21 though? we're skipping the Spark Connect shell here.

Copy link
Contributor Author

@LuciferYang LuciferYang Jul 5, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All RemoteSparkSession-based test using Java 21 already skipped in #41805, so we shouldn't initialize Ammonite here and connect to the remote spark connect server here.

This did not make local test fail, but the test on GA failed (Daily testing of Java 21)

super.beforeAll()
ammoniteOut = new ByteArrayOutputStream()
testSuiteOut = new PipedOutputStream()
// Connect the `testSuiteOut` and `ammoniteIn` pipes
ammoniteIn = new PipedInputStream(testSuiteOut)
errorStream = new ByteArrayOutputStream()

val args = Array("--port", serverPort.toString)
val task = new Runnable {
override def run(): Unit = {
ConnectRepl.doMain(
args = args,
semaphore = Some(semaphore),
inputStream = ammoniteIn,
outputStream = ammoniteOut,
errorStream = errorStream)
}
}
}

executorService.submit(task)
executorService.submit(task)
}
}

override def afterAll(): Unit = {
Expand Down