This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 560c050aae [CELEBORN-2493] Complete integrity validation for empty AQE
chunk splits
560c050aae is described below
commit 560c050aaee5b47927f6ed32ebe1b95c56ce558c
Author: zhengtao <[email protected]>
AuthorDate: Fri Oct 30 22:52:47 2026 +0800
[CELEBORN-2493] Complete integrity validation for empty AQE chunk splits
### What changes were proposed in this pull request?
Report zero checksum and zero bytes for an optimized AQE child when the
parent has committed locations but chunk splitting selects no locations for
that child. Preserve the original subpartition count/index in
`ReadReducerPartitionEnd`.
The report runs inside the reader's existing stream-creation error
handling. The shared empty stream and existing failure propagation remain in
use. Ordinary empty reducer partitions retain their current filtering path.
Close the static `ShuffleClient` mock in the existing reader test so it
does not leak into the new regression suite.
JIRA: https://issues.apache.org/jira/browse/CELEBORN-2493
### Why are the changes needed?
AQE's requested split count comes from shuffle size estimates, while
Celeborn assigns complete physical chunks to the child tasks. A nonempty parent
with one physical chunk can be split into two children: child 0 selects no
chunks and child 1 reads the whole chunk. The current empty path returns
without reporting child 0. The driver waits for both reports and never compares
the aggregated checksum, allowing a corrupted sibling to escape validation.
#### Revised scope after #3885
This is a replacement for the closed
https://github.com/apache/celeborn/pull/3885, using the same JIRA.
The earlier approach retained every requested reducer partition whenever
integrity checking was enabled and validated every empty return in the shared
client/factory. Although this detects injected missing-location metadata, it
also turns ordinary zero-write/zero-read partitions into stream-creation tasks
and one blocking validation RPC each. Sparse shuffles can contain many such
partitions. That cost is broader than the reproduced AQE defect, and a
successful driver response silently [...]
| Path | Earlier #3885 | This PR |
| --- | --- | --- |
| Ordinary empty reducer omitted from file groups | Keep it, create an
empty-read task and send an integrity RPC | Keep the existing filter; no stream
task or integrity RPC |
| Empty AQE chunk selection from a parent with locations | Validate through
the generic empty-stream path | Send a zero report at the Spark reader, where
the nonempty parent and empty selection are both known |
| Empty/missing locations in generic client/factory calls | Validate all
such returns | Preserve existing behavior |
| Driver inventory / configuration / wire format | No inventory scan | No
inventory scan, configuration or wire-format changes |
Only an empty AQE child of a parent with locations adds a completion RPC.
No completion RPC is added for nonempty reads, and there is no per-record work.
#### Relationship to #3905
https://github.com/apache/celeborn/pull/3905 handles safe failure after a
detected partial-read integrity error. This PR supplies the missing completion
report so that the parent checksum comparison can run. The changes are separate
and no #3905 code is included here. Please land #3905 first so newly detected
AQE errors use its safe failure handling when stage rerun is enabled.
### Does this PR resolve a correctness bug?
- [x] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
With integrity checking and optimized skewed reads enabled, empty AQE
children participate in parent integrity validation. Corrupt aggregate metadata
is rejected instead of leaving the parent validation incomplete. Both option
defaults, the RPC schema and the ordinary empty-partition path are unchanged.
### How was this patch tested?
Local validation used Java 17, Maven 3.9.16, Scala 2.12.18 and Spark 3.5.9.
- Identical regression sources on unmodified production: **6 failed, 3
passed**. Failures cover missing child reports/parent completion and a corrupt
sibling being accepted.
- With the fix: **all 9 new regressions passed**. Two/four-way splits are
tested with the empty reports arriving before and after the nonempty sibling;
corruption tests cover the reader's existing failure propagation with stage
rerun on/off.
- **100 unique tests passed**: 54 client commit/validator tests, 19 reader
tests (including the 9 new tests), 17 `ShuffleClientSuiteJ`, 5 crypto
round-trip tests, and 5 `CelebornPartitionUtilSuiteJ` tests.
- A request for **10,000 ordinary empty reducers** creates **0 integrity
RPCs, 0 stream tasks and 0 worker connections**. A present-but-empty location
entry and disabled integrity checking also add no integrity RPC.
- After formatting, **all 19 reader tests passed in one JVM with the
existing suite first**, checking that its static client mock no longer leaks.
- Full `dev/reformat`: **all 16 profiles succeeded**; `git diff --check`
passed.
Commands:
```bash
dev/reformat
build/mvn -Pspark-3.5 -pl client-spark/spark-3 -am test \
-Dtest=CelebornPartitionUtilSuiteJ,ShuffleClientSuiteJ,CelebornInputStreamCryptoRoundTripSuiteJ
\
-DwildcardSuites=org.apache.spark.shuffle.celeborn.EmptySkewPartitionIntegritySuite,org.apache.spark.shuffle.celeborn.CelebornShuffleReaderSuite,org.apache.celeborn.client.commit
\
-Dmaven.gitcommitid.skip=true
build/mvn -Pspark-3.5 -pl client-spark/spark-3 -am install \
-DskipTests -Dmaven.gitcommitid.skip=true
build/mvn -Pspark-3.5 -pl client-spark/spark-3 test -Dtest=none \
-Dsuites=org.apache.spark.shuffle.celeborn.CelebornShuffleReaderSuite,org.apache.spark.shuffle.celeborn.EmptySkewPartitionIntegritySuite
\
-Dmaven.gitcommitid.skip=true
```
The unit fixture mocks RPC transport and the unused worker connection
factory. Chunk selection, client request creation, mapper metadata registration
and the parent validator are production implementations; sibling metadata is
supplied by the fixture. The 10,000-partition test checks operation counts, not
workload latency.
Actual Spark validation used the community server/client built from this
commit with JDK 8, EMR Spark 3.5.3 `local[4,2]` on an ECS master, and **two
real remote community Celeborn workers**. It used the production
`SparkShuffleManager`, optimized skewed reads, a 128 KiB AQE target,
compression `NONE`, no replication, and stage rerun disabled (with
failed-shuffle cleanup also disabled to satisfy current main's configuration
constraint).
| Workload | Result |
| --- | --- |
| Healthy 16,000-row skewed shuffle | 16,000 correct rows; **4 empty AQE
children reported**, **0 incomplete parent validations** |
| Same workload, one committed worker payload byte flipped | Parent CRC
mismatch detected; Spark query fails after its two task attempts; no successful
result returned |
| Same injected corruption with integrity checking disabled | Query returns
16,000 rows with exactly **1 changed payload**, confirming the injection |
The injection plugin changes only a payload byte in its own local test
application's committed worker file before result tasks read it. It does not
replace the manager or fabricate file groups, read reports or expected
checksums. The harness exits successfully after verifying the expected query
failure in the corruption case. This is a real Spark/Celeborn workload in local
Spark mode with remote workers, not a distributed YARN fault-recovery or
performance benchmark.
Closes #3914 from zaynt4606/codex/aqe-empty-report-community-20261029.
Authored-by: zhengtao <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
---
.../shuffle/celeborn/CelebornShuffleReader.scala | 13 +
.../celeborn/CelebornShuffleReaderSuite.scala | 23 +-
.../EmptySkewPartitionIntegritySuite.scala | 334 +++++++++++++++++++++
3 files changed, 365 insertions(+), 5 deletions(-)
diff --git
a/client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala
b/client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala
index c0c2b20ec8..69202f7039 100644
---
a/client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala
+++
b/client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReader.scala
@@ -422,6 +422,19 @@ class CelebornShuffleReader[K, C](
} else null
if (exceptionRef.get() == null) {
try {
+ // Chunk boundaries can leave an AQE child empty even though its
parent has files.
+ // Its zero report is still required to complete the parent's
integrity validation.
+ if (splitSkewPartitionWithoutMapRange && locationList.isEmpty &&
+ conf.clientShuffleIntegrityCheckEnabled &&
+ !fileGroups.partitionGroups.get(partitionId).isEmpty) {
+ shuffleClient.readReducerPartitionEnd(
+ shuffleId,
+ partitionId,
+ startMapIndex,
+ endMapIndex,
+ 0,
+ 0L)
+ }
val inputStream = shuffleClient.readPartition(
shuffleId,
handle.shuffleId,
diff --git
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
index c8ac70c59b..ea64b0534d 100644
---
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
+++
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
@@ -61,12 +61,25 @@ class CelebornShuffleReaderSuite extends AnyFunSuite {
val conf = new CelebornConf()
val tmpFile = Files.createTempFile("test", ".tmp").toFile
- mockStatic(classOf[ShuffleClient]).when(() =>
- ShuffleClient.get(any(), any(), any(), any(), any(), any(),
any())).thenReturn(
- new DummyShuffleClient(conf, tmpFile))
-
+ val clientFactory = mockStatic(classOf[ShuffleClient])
val shuffleReader =
- new CelebornShuffleReader[Int, Int](handler, 0, 0, 0, 0, context, conf,
metricReporter, null)
+ try {
+ clientFactory.when(() =>
+ ShuffleClient.get(any(), any(), any(), any(), any(), any(),
any())).thenReturn(
+ new DummyShuffleClient(conf, tmpFile))
+ new CelebornShuffleReader[Int, Int](
+ handler,
+ 0,
+ 0,
+ 0,
+ 0,
+ context,
+ conf,
+ metricReporter,
+ null)
+ } finally {
+ clientFactory.close()
+ }
val exception1: Throwable = new CelebornIOException("test1", new
InterruptedException("test1"))
val exception2: Throwable = new CelebornIOException("test2", new
TimeoutException("test2"))
diff --git
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/EmptySkewPartitionIntegritySuite.scala
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/EmptySkewPartitionIntegritySuite.scala
new file mode 100644
index 0000000000..40dce39663
--- /dev/null
+++
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/EmptySkewPartitionIntegritySuite.scala
@@ -0,0 +1,334 @@
+/*
+ * 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.shuffle.celeborn
+
+import java.util
+import java.util.concurrent.CopyOnWriteArrayList
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.{ShuffleDependency, SparkConf, TaskContext}
+import org.apache.spark.executor.TaskMetrics
+import org.apache.spark.serializer.JavaSerializer
+import org.apache.spark.shuffle.{FetchFailedException,
ShuffleReadMetricsReporter}
+import org.mockito.Answers
+import org.mockito.ArgumentMatchers.any
+import org.mockito.Mockito._
+import org.mockito.invocation.InvocationOnMock
+import org.mockito.stubbing.Answer
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.celeborn.client.{ShuffleClient, ShuffleClientImpl,
WorkerStatusTracker}
+import org.apache.celeborn.client.CommitManager.CommittedPartitionInfo
+import org.apache.celeborn.client.LifecycleManager.ShuffleAllocatedWorkers
+import org.apache.celeborn.client.commit.ReducePartitionCommitHandler
+import org.apache.celeborn.common.{CelebornConf, CommitMetadata}
+import org.apache.celeborn.common.exception.CelebornIOException
+import org.apache.celeborn.common.identity.UserIdentifier
+import org.apache.celeborn.common.network.client.TransportClientFactory
+import org.apache.celeborn.common.protocol._
+import
org.apache.celeborn.common.protocol.message.ControlMessages.{GetReducerFileGroup,
GetReducerFileGroupResponse}
+import org.apache.celeborn.common.protocol.message.StatusCode
+import org.apache.celeborn.common.rpc.RpcEndpointRef
+import org.apache.celeborn.common.util.ThreadUtils
+
+class EmptySkewPartitionIntegritySuite extends AnyFunSuite with
BeforeAndAfterAll {
+ private val shuffleId = 7
+ private val partitionId = 1
+ private var previousStreamCreatorPool:
java.util.concurrent.ThreadPoolExecutor = _
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ previousStreamCreatorPool = CelebornShuffleReader.streamCreatorPool
+ }
+
+ override def afterAll(): Unit = {
+ try {
+ if (previousStreamCreatorPool == null &&
CelebornShuffleReader.streamCreatorPool != null) {
+ ThreadUtils.shutdown(CelebornShuffleReader.streamCreatorPool)
+ CelebornShuffleReader.streamCreatorPool = null
+ }
+ } finally {
+ super.afterAll()
+ }
+ }
+
+ // The reader, chunk splitter, client RPC serialization and driver validator
are production
+ // implementations. Only RPC transport and the unused worker connection
factory are mocked.
+ private class Fixture(
+ integrityEnabled: Boolean = true,
+ numPartitions: Int = 3,
+ nonemptyParent: Boolean = true)
+ extends AutoCloseable {
+ val conf = new CelebornConf()
+ .set("celeborn.client.shuffle.integrityCheck.enabled",
integrityEnabled.toString)
+ .set("celeborn.client.adaptive.optimizeSkewedPartitionRead.enabled",
"true")
+ private val rpcPool =
ThreadUtils.newDaemonCachedThreadPool("empty-skew-test-rpc")
+ private val scheduler =
+
ThreadUtils.newDaemonSingleThreadScheduledExecutor("empty-skew-test-scheduler")
+ val handler = new ReducePartitionCommitHandler(
+ "empty-skew-test",
+ conf,
+ new ShuffleAllocatedWorkers(),
+ new CommittedPartitionInfo(),
+ new WorkerStatusTracker(conf, null),
+ rpcPool,
+ scheduler,
+ null)
+ val expected = new CommitMetadata()
+ private val payload = "nonempty parent with one chunk".getBytes("UTF-8")
+ if (nonemptyParent) expected.addDataWithOffsetAndLength(payload, 0,
payload.length)
+ handler.registerShuffle(shuffleId, 2, isSegmentGranularityVisible = false,
numPartitions)
+ (0 until 2).foreach { mapId =>
+ val metadata = if (mapId == 1) expected else new CommitMetadata()
+ val checksums = new Array[Int](numPartitions)
+ val bytes = new Array[Long](numPartitions)
+ checksums(partitionId) = metadata.getChecksum
+ bytes(partitionId) = metadata.getBytes
+ assert(handler.finishMapperAttempt(
+ shuffleId,
+ mapId,
+ 0,
+ 2,
+ partitionId,
+ null,
+ _ => (),
+ numPartitions,
+ checksums,
+ bytes) ==
+ ((true, mapId == 1)))
+ }
+
+ val location = new PartitionLocation(
+ partitionId,
+ 0,
+ "localhost",
+ 19001,
+ 19002,
+ 19003,
+ 19004,
+ PartitionLocation.Mode.PRIMARY)
+ location.setStorageInfo(new StorageInfo(
+ StorageInfo.Type.HDD,
+ "mount",
+ true,
+ "file",
+ StorageInfo.LOCAL_DISK_MASK,
+ payload.length.toLong,
+ util.Arrays.asList(0L, payload.length.toLong)))
+ val fileGroups = new util.HashMap[Integer, util.Set[PartitionLocation]]()
+ fileGroups.put(partitionId, util.Collections.singleton(location))
+ val reports = new CopyOnWriteArrayList[PbReadReducerPartitionEnd]()
+ val validations = new CopyOnWriteArrayList[(Boolean, String)]()
+ val workerFactory = mock(classOf[TransportClientFactory])
+ private val endpoint = mock(
+ classOf[RpcEndpointRef],
+ new Answer[AnyRef] {
+ override def answer(invocation: InvocationOnMock): AnyRef = {
+ if (invocation.getMethod.getName != "askSync") {
+ Answers.RETURNS_DEFAULTS.answer(invocation)
+ } else {
+ invocation.getArgument[AnyRef](0) match {
+ case _: GetReducerFileGroup =>
+ GetReducerFileGroupResponse(
+ StatusCode.SUCCESS,
+ fileGroups,
+ handler.getMapperAttempts(shuffleId))
+ case request: PbReadReducerPartitionEnd =>
+ reports.add(request)
+ val result = handler.finishPartition(
+ request.getShuffleId,
+ request.getPartitionId,
+ request.getStartMaxIndex,
+ request.getEndMapIndex,
+ new CommitMetadata(request.getCrc32,
request.getBytesWritten))
+ validations.add(result)
+ PbReadReducerPartitionEndResponse.newBuilder()
+ .setStatus(if (result._1) StatusCode.SUCCESS.getValue
+ else StatusCode.READ_REDUCER_PARTITION_END_FAILED.getValue)
+ .setErrorMsg(result._2).build()
+ case _: PbGetStageEnd =>
+ PbGetStageEndResponse.newBuilder().setStageEnd(true).build()
+ case _: PbGetShuffleId =>
+
PbGetShuffleIdResponse.newBuilder().setShuffleId(shuffleId).setSuccess(true).build()
+ case _: PbReportShuffleFetchFailure =>
+
PbReportShuffleFetchFailureResponse.newBuilder().setSuccess(true).build()
+ case other => throw new AssertionError(s"Unexpected RPC: $other")
+ }
+ }
+ }
+ })
+ val client = new ShuffleClientImpl("empty-skew-test", conf,
UserIdentifier("test", "test")) {
+ override def getDataClientFactory(): TransportClientFactory =
workerFactory
+ }
+ client.setupLifecycleManagerRef(endpoint)
+
+ def report(metadata: CommitMetadata, count: Int, index: Int): Unit = {
+ client.readReducerPartitionEnd(
+ shuffleId,
+ partitionId,
+ count,
+ index,
+ metadata.getChecksum,
+ metadata.getBytes)
+ }
+
+ override def close(): Unit = {
+ client.shutdown()
+ rpcPool.shutdownNow()
+ scheduler.shutdownNow()
+ }
+ }
+
+ private def withFixture(
+ integrityEnabled: Boolean = true,
+ numPartitions: Int = 3,
+ nonemptyParent: Boolean = true)(
+ body: Fixture => Unit): Unit = {
+ val fixture = new Fixture(integrityEnabled, numPartitions, nonemptyParent)
+ try body(fixture)
+ finally fixture.close()
+ }
+
+ private def reader(
+ fixture: Fixture,
+ count: Int = 2,
+ index: Int = 0,
+ firstPartition: Int = partitionId,
+ lastPartition: Int = partitionId + 1,
+ stageRerun: Boolean = false): CelebornShuffleReader[Int, Int] = {
+ val dependency = mock(classOf[ShuffleDependency[Int, Int, Int]])
+ when(dependency.serializer).thenReturn(new JavaSerializer(new
SparkConf(false)))
+ when(dependency.keyOrdering).thenReturn(None)
+ when(dependency.aggregator).thenReturn(None)
+ val handle = new CelebornShuffleHandle[Int, Int, Int](
+ "empty-skew-test",
+ "localhost",
+ 0,
+ UserIdentifier("test", "test"),
+ shuffleId,
+ stageRerun,
+ 2,
+ dependency)
+ val context = mock(classOf[TaskContext])
+ when(context.taskMetrics()).thenReturn(TaskMetrics.empty)
+ val factory = mockStatic(classOf[ShuffleClient])
+ try {
+ factory.when(() => ShuffleClient.get(any(), any(), any(), any(), any(),
any(), any()))
+ .thenReturn(fixture.client)
+ new CelebornShuffleReader[Int, Int](
+ handle,
+ firstPartition,
+ lastPartition,
+ count,
+ index,
+ context,
+ fixture.conf,
+ mock(classOf[ShuffleReadMetricsReporter]),
+ new ExecutorShuffleIdTracker())
+ } finally {
+ factory.close()
+ }
+ }
+
+ for {
+ count <- Seq(2, 4)
+ emptyFirst <- Seq(false, true)
+ } {
+ test(
+ s"empty chunk selections complete parent validation: count=$count,
emptyFirst=$emptyFirst") {
+ withFixture() { fixture =>
+ if (!emptyFirst) fixture.report(fixture.expected, count, count - 1)
+ (0 until count - 1).foreach { index =>
+ assert(CelebornPartitionUtil.splitSkewedPartitionLocations(
+ new util.ArrayList(fixture.fileGroups.get(partitionId)),
+ count,
+ index).isEmpty)
+ val iter = reader(fixture, count, index).read()
+ assert(!iter.hasNext)
+ assert(!iter.hasNext)
+ }
+ if (emptyFirst) fixture.report(fixture.expected, count, count - 1)
+ assert(fixture.reports.size() == count)
+ val emptyReports = fixture.reports.asScala.filter(_.getEndMapIndex <
count - 1)
+ assert(emptyReports.map(_.getEndMapIndex).toSet == (0 until count -
1).toSet)
+ assert(emptyReports.forall(r =>
+ r.getStartMaxIndex == count &&
+ r.getBytesWritten == 0 && r.getCrc32 == 0))
+ assert(fixture.validations.asScala.last == ((true, "Partition is
complete")))
+ verifyNoInteractions(fixture.workerFactory)
+ }
+ }
+ }
+
+ Seq(false, true).foreach { stageRerun =>
+ test(s"an empty child's final report exposes a corrupt sibling:
stageRerun=$stageRerun") {
+ withFixture() { fixture =>
+ fixture.report(
+ new CommitMetadata(fixture.expected.getChecksum ^ 1,
fixture.expected.getBytes),
+ 2,
+ 1)
+ assert(fixture.validations.asScala.last._1)
+ val error = intercept[Exception] {
+ reader(fixture, stageRerun = stageRerun).read().toList
+ }
+ val integrityError =
+ if (stageRerun) {
+ assert(error.isInstanceOf[FetchFailedException])
+ error.getCause
+ } else error
+ assert(integrityError.isInstanceOf[CelebornIOException])
+ assert(integrityError.getMessage.contains("failed validation check"))
+ assert(fixture.reports.size() == 2)
+ assert(!fixture.validations.asScala.last._1)
+ verifyNoInteractions(fixture.workerFactory)
+ }
+ }
+ }
+
+ test("ordinary empty partitions keep the zero-RPC and zero-stream-task fast
path") {
+ withFixture(numPartitions = 10002) { fixture =>
+ val emptyReader =
+ reader(fixture, count = 0, index = Int.MaxValue, firstPartition = 2,
lastPartition = 10002)
+ val tasksBefore = Option(CelebornShuffleReader.streamCreatorPool)
+ .map(_.getTaskCount).getOrElse(0L)
+ assert(emptyReader.read().toList.isEmpty)
+ assert(fixture.reports.isEmpty)
+ assert(CelebornShuffleReader.streamCreatorPool.getTaskCount ==
tasksBefore)
+ verifyNoInteractions(fixture.workerFactory)
+ }
+ }
+
+ test("an empty parent location entry does not gain an integrity RPC") {
+ withFixture(nonemptyParent = false) { fixture =>
+ fixture.fileGroups.put(partitionId,
util.Collections.emptySet[PartitionLocation]())
+ assert(reader(fixture).read().toList.isEmpty)
+ assert(fixture.reports.isEmpty)
+ verifyNoInteractions(fixture.workerFactory)
+ }
+ }
+
+ test("disabled integrity checking does not report empty AQE children") {
+ withFixture(integrityEnabled = false) { fixture =>
+ assert(reader(fixture).read().toList.isEmpty)
+ assert(fixture.reports.isEmpty)
+ verifyNoInteractions(fixture.workerFactory)
+ }
+ }
+}
