You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2022/11/25 12:57:51 UTC
[incubator-celeborn] branch main updated: [CELEBORN-67] Revert "Fix fetch incorrect data chunk" related commits (#1006)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 93dbf3f8 [CELEBORN-67] Revert "Fix fetch incorrect data chunk" related commits (#1006)
93dbf3f8 is described below
commit 93dbf3f8b1885411c5adf8378b4d8aa966e1a10c
Author: Ethan Feng <et...@apache.org>
AuthorDate: Fri Nov 25 20:57:47 2022 +0800
[CELEBORN-67] Revert "Fix fetch incorrect data chunk" related commits (#1006)
* Revert "[CELEBORN-50][FOLLOWUP] Channel inactive may cause new client use old stream id to fetch data (#999)"
This reverts commit 1e8f6dc5e87669d96f84c1acb04a1c0880f3b3fc.
* Revert "[CELEBORN-50] Channel inActive may cause new client use old stream id to fetch data cause IllegalStateException. (#1000)"
This reverts commit f1c4d675d62eed9b9776167f76ffddfec08fe8d7.
* Revert "[CELEBORN-49] Deadlock when kill worker in shuffle read (#998)"
This reverts commit 0be4b3399cffe8c7f809c9d0a47298b64cef07ee.
* Revert "[CELEBORN-47][IMPROVEMENT] Refine logs about tracking fetch chunk (#995)"
This reverts commit 2b05228871981b2fd0a249746797c333c6b168a6.
* Revert "[BUG] Fix fetch incorrect data chunk (#926)"
This reverts commit 6f043f8a
* Revert "[ISSUE-925][FOLLOWUP] Refactor class name of RetryingChunkReceiveCallback (#954)"
This reverts commit 64e8ebf1
---
.../org/apache/celeborn/client/read/Replica.java | 33 +-
.../{ChunkClient.java => RetryingChunkClient.java} | 115 +++---
.../client/read/WorkerPartitionReader.java | 117 ++----
.../client/read/RetryingChunkClientSuiteJ.java | 423 +++++++++++++++++++++
.../network/client/ChunkReceivedCallback.java | 5 +-
.../common/network/client/TransportClient.java | 2 +-
.../network/client/TransportResponseHandler.java | 5 +-
.../network/server/OneForOneStreamManager.java | 29 +-
.../common/network/server/StreamManager.java | 20 +
.../org/apache/celeborn/common/CelebornConf.scala | 10 -
.../network/ChunkFetchIntegrationSuiteJ.java | 5 +-
.../network/RequestTimeoutIntegrationSuiteJ.java | 5 +-
.../network/TransportResponseHandlerSuiteJ.java | 10 +-
.../service/deploy/worker/FetchHandler.scala | 3 -
.../deploy/worker/storage/FileWriterSuiteJ.java | 5 +-
15 files changed, 578 insertions(+), 209 deletions(-)
diff --git a/client/src/main/java/org/apache/celeborn/client/read/Replica.java b/client/src/main/java/org/apache/celeborn/client/read/Replica.java
index 2455c088..1fe7d3b3 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/Replica.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/Replica.java
@@ -21,8 +21,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import com.google.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.celeborn.common.network.client.TransportClient;
import org.apache.celeborn.common.network.client.TransportClientFactory;
@@ -32,7 +30,6 @@ import org.apache.celeborn.common.network.protocol.StreamHandle;
import org.apache.celeborn.common.protocol.PartitionLocation;
class Replica {
- private Logger logger = LoggerFactory.getLogger(Replica.class);
private final long timeoutMs;
private final String shuffleKey;
private final PartitionLocation location;
@@ -60,36 +57,16 @@ class Replica {
public synchronized TransportClient getOrOpenStream() throws IOException, InterruptedException {
if (client == null || !client.isActive()) {
- if (client != null) {
- logger.warn(
- "Current channel from "
- + client.getChannel().localAddress()
- + " to "
- + client.getChannel().remoteAddress()
- + " for "
- + this
- + " is not active.");
- }
client = clientFactory.createClient(location.getHost(), location.getFetchPort());
- // When client is not active, the origin client's corresponding streamId may be removed
- // by channel inactive. Replica should request a new StreamHandle for the new client again.
- // Newly returned numChunks should be the same.
- openStreamInternal();
- }
- // For retried open stream if openStream rpc is failed.
- if (streamHandle == null) {
- openStreamInternal();
+
+ OpenStream openBlocks =
+ new OpenStream(shuffleKey, location.getFileName(), startMapIndex, endMapIndex);
+ ByteBuffer response = client.sendRpcSync(openBlocks.toByteBuffer(), timeoutMs);
+ streamHandle = (StreamHandle) Message.decode(response);
}
return client;
}
- private void openStreamInternal() {
- OpenStream openBlocks =
- new OpenStream(shuffleKey, location.getFileName(), startMapIndex, endMapIndex);
- ByteBuffer response = client.sendRpcSync(openBlocks.toByteBuffer(), timeoutMs);
- streamHandle = (StreamHandle) Message.decode(response);
- }
-
public long getStreamId() {
return streamHandle.streamId;
}
diff --git a/client/src/main/java/org/apache/celeborn/client/read/ChunkClient.java b/client/src/main/java/org/apache/celeborn/client/read/RetryingChunkClient.java
similarity index 73%
rename from client/src/main/java/org/apache/celeborn/client/read/ChunkClient.java
rename to client/src/main/java/org/apache/celeborn/client/read/RetryingChunkClient.java
index 0baa5e66..c044b2b5 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/ChunkClient.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/RetryingChunkClient.java
@@ -50,25 +50,19 @@ import org.apache.celeborn.common.util.Utils;
* file will not be too many. Each retry is actually a switch between Master and Slave. Therefore,
* each retry needs to create a new connection and reopen the file to generate the stream id.
*/
-public class ChunkClient {
- private static final Logger logger = LoggerFactory.getLogger(ChunkClient.class);
+public class RetryingChunkClient {
+ private static final Logger logger = LoggerFactory.getLogger(RetryingChunkClient.class);
private static final ExecutorService executorService =
Executors.newCachedThreadPool(NettyUtils.createThreadFactory("fetch-chuck"));
private final ChunkReceivedCallback callback;
- private final Replica replica;
+ private final Replica[] replicas;
private final long retryWaitMs;
private final int maxTries;
private volatile int numTries = 0;
- private PartitionLocation location;
- private int fetchFailedChunkIndex;
- public PartitionLocation getLocation() {
- return location;
- }
-
- public ChunkClient(
+ public RetryingChunkClient(
CelebornConf conf,
String shuffleKey,
PartitionLocation location,
@@ -77,7 +71,7 @@ public class ChunkClient {
this(conf, shuffleKey, location, callback, clientFactory, 0, Integer.MAX_VALUE);
}
- public ChunkClient(
+ public RetryingChunkClient(
CelebornConf conf,
String shuffleKey,
PartitionLocation location,
@@ -87,52 +81,63 @@ public class ChunkClient {
int endMapIndex) {
TransportConf transportConf =
Utils.fromCelebornConf(conf, TransportModuleConstants.DATA_MODULE, 0);
- this.fetchFailedChunkIndex = conf.testFetchFailedChunkIndex();
+
this.callback = callback;
this.retryWaitMs = transportConf.ioRetryWaitTimeMs();
long fetchTimeoutMs = conf.fetchTimeoutMs();
- this.location = location;
if (location == null) {
throw new IllegalArgumentException("Must contain at least one available PartitionLocation.");
} else {
- replica =
+ Replica main =
new Replica(
fetchTimeoutMs, shuffleKey, location, clientFactory, startMapIndex, endMapIndex);
+ PartitionLocation peerLoc = location.getPeer();
+ if (peerLoc == null) {
+ replicas = new Replica[] {main};
+ } else {
+ Replica peer =
+ new Replica(
+ fetchTimeoutMs, shuffleKey, peerLoc, clientFactory, startMapIndex, endMapIndex);
+ replicas = new Replica[] {main, peer};
+ }
}
- this.maxTries = (transportConf.maxIORetries() + 1);
+ this.maxTries = (transportConf.maxIORetries() + 1) * replicas.length;
}
/**
- * This method should only be called once after PartitionReader is initialized, so it is assumed
- * that there is no concurrency problem when it is called.
+ * This method should only be called once after RetryingChunkReader is initialized, so it is
+ * assumed that there is no concurrency problem when it is called.
*
* @return numChunks.
*/
public synchronized int openChunks() throws IOException {
int numChunks = -1;
+ Replica currentReplica = null;
Exception currentException = null;
while (numChunks == -1 && hasRemainingRetries()) {
// Only not wait for first request to each replicate.
- if (numTries != 0) {
+ currentReplica = getCurrentReplica();
+ if (numTries >= replicas.length) {
logger.info(
"Retrying openChunk ({}/{}) for chunk from {} after {} ms.",
numTries,
maxTries,
- replica,
+ currentReplica,
retryWaitMs);
Uninterruptibles.sleepUninterruptibly(retryWaitMs, TimeUnit.MILLISECONDS);
}
try {
- replica.getOrOpenStream();
- numChunks = replica.getNumChunks();
+ currentReplica.getOrOpenStream();
+ numChunks = currentReplica.getNumChunks();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IOException(e);
} catch (Exception e) {
- logger.error("Exception raised while sending open chunks message to " + replica + ".", e);
+ logger.error(
+ "Exception raised while sending open chunks message to " + currentReplica + ".", e);
currentException = e;
if (shouldRetry(e)) {
numTries += 1;
@@ -143,21 +148,16 @@ public class ChunkClient {
}
if (numChunks == -1) {
if (currentException != null) {
- callback.onFailure(
- 0,
- location,
- new IOException(
- String.format("Could not open chunks from %s after %d tries.", replica, numTries),
- currentException));
+ throw new IOException(
+ String.format(
+ "Could not open chunks from %s after %d tries.", currentReplica, numTries),
+ currentException);
} else {
- callback.onFailure(
- 0,
- location,
- new IOException(
- String.format("Could not open chunks from %s after %d tries.", replica, numTries)));
+ throw new IOException(
+ String.format(
+ "Could not open chunks from %s after %d tries.", currentReplica, numTries));
}
}
- numTries = 0;
return numChunks;
}
@@ -169,35 +169,36 @@ public class ChunkClient {
* @param chunkIndex the index of the chunk to be fetched.
*/
public void fetchChunk(int chunkIndex) {
- FetchChunkCallback callback;
+ Replica replica;
+ RetryingChunkReceiveCallback callback;
synchronized (this) {
- callback = new FetchChunkCallback(numTries);
- if (fetchFailedChunkIndex != 0
- && location.getPeer() != null
- && chunkIndex == fetchFailedChunkIndex
- && location.getMode() == PartitionLocation.Mode.MASTER) {
- RuntimeException manualTriggeredFailure =
- new RuntimeException("Manual triggered fetch failure");
- callback.onFailure(chunkIndex, location, manualTriggeredFailure);
- }
+ replica = getCurrentReplica();
+ callback = new RetryingChunkReceiveCallback(numTries);
}
try {
TransportClient client = replica.getOrOpenStream();
client.fetchChunk(replica.getStreamId(), chunkIndex, callback);
} catch (Exception e) {
logger.error(
- "Exception raised while beginning fetch chunk "
- + chunkIndex
- + (numTries > 0 ? " (after " + numTries + " retries)" : ""),
+ "Exception raised while beginning fetch chunk {}{}.",
+ chunkIndex,
+ numTries > 0 ? " (after " + numTries + " retries)" : "",
e);
+
if (shouldRetry(e)) {
initiateRetry(chunkIndex, callback.currentNumTries);
} else {
- callback.onFailure(chunkIndex, location, e);
+ callback.onFailure(chunkIndex, e);
}
}
}
+ @VisibleForTesting
+ Replica getCurrentReplica() {
+ int currentReplicaIndex = numTries % replicas.length;
+ return replicas[currentReplicaIndex];
+ }
+
@VisibleForTesting
int getNumTries() {
return numTries;
@@ -227,7 +228,7 @@ public class ChunkClient {
currentNumTries,
maxTries,
chunkIndex,
- replica,
+ getCurrentReplica(),
retryWaitMs);
executorService.submit(
@@ -237,27 +238,25 @@ public class ChunkClient {
});
}
- private class FetchChunkCallback implements ChunkReceivedCallback {
+ private class RetryingChunkReceiveCallback implements ChunkReceivedCallback {
final int currentNumTries;
- FetchChunkCallback(int currentNumTries) {
+ RetryingChunkReceiveCallback(int currentNumTries) {
this.currentNumTries = currentNumTries;
}
@Override
- public void onSuccess(int chunkIndex, ManagedBuffer buffer, PartitionLocation location) {
- callback.onSuccess(chunkIndex, buffer, ChunkClient.this.location);
+ public void onSuccess(int chunkIndex, ManagedBuffer buffer) {
+ callback.onSuccess(chunkIndex, buffer);
}
@Override
- public void onFailure(int chunkIndex, PartitionLocation location, Throwable e) {
+ public void onFailure(int chunkIndex, Throwable e) {
if (shouldRetry(e)) {
initiateRetry(chunkIndex, this.currentNumTries);
} else {
- logger.error(
- "Abandon to fetch chunk " + chunkIndex + " after " + this.currentNumTries + " tries.",
- e);
- callback.onFailure(chunkIndex, ChunkClient.this.location, e);
+ logger.error("Abandon to fetch chunk {} after {} tries.", chunkIndex, this.currentNumTries);
+ callback.onFailure(chunkIndex, e);
}
}
}
diff --git a/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java b/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
index 9ea850f2..01e35b4d 100644
--- a/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
+++ b/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
@@ -18,14 +18,12 @@
package org.apache.celeborn.client.read;
import java.io.IOException;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCounted;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,19 +36,17 @@ import org.apache.celeborn.common.protocol.PartitionLocation;
public class WorkerPartitionReader implements PartitionReader {
private final Logger logger = LoggerFactory.getLogger(WorkerPartitionReader.class);
- private ChunkClient client;
- private int numChunks;
+ private final RetryingChunkClient client;
+ private final int numChunks;
private int returnedChunks;
- private int currentChunkIndex;
+ private int chunkIndex;
- private final LinkedBlockingQueue<ChunkData> results;
+ private final LinkedBlockingQueue<ByteBuf> results;
private final AtomicReference<IOException> exception = new AtomicReference<>();
private final int fetchMaxReqsInFlight;
- private AtomicBoolean closed = new AtomicBoolean(false);
- private Set<PartitionLocation> readableLocations = ConcurrentHashMap.newKeySet();
- private Set<PartitionLocation> failedLocations = ConcurrentHashMap.newKeySet();
+ private boolean closed = false;
WorkerPartitionReader(
CelebornConf conf,
@@ -62,89 +58,48 @@ public class WorkerPartitionReader implements PartitionReader {
throws IOException {
fetchMaxReqsInFlight = conf.fetchMaxReqsInFlight();
results = new LinkedBlockingQueue<>();
- readableLocations.add(location);
- if (location.getPeer() != null) {
- readableLocations.add(location.getPeer());
- }
// only add the buffer to results queue if this reader is not closed.
ChunkReceivedCallback callback =
new ChunkReceivedCallback() {
@Override
- public void onSuccess(int chunkIndex, ManagedBuffer buffer, PartitionLocation location) {
+ public void onSuccess(int chunkIndex, ManagedBuffer buffer) {
// only add the buffer to results queue if this reader is not closed.
- ByteBuf buf = ((NettyManagedBuffer) buffer).getBuf();
- if (!closed.get() && !failedLocations.contains(location)) {
- buf.retain();
- results.add(new ChunkData(buf, location));
+ synchronized (this) {
+ ByteBuf buf = ((NettyManagedBuffer) buffer).getBuf();
+ if (!closed) {
+ buf.retain();
+ results.add(buf);
+ }
}
}
@Override
- public void onFailure(int chunkIndex, PartitionLocation location, Throwable e) {
- readableLocations.remove(location);
- if (readableLocations.isEmpty()) {
- String errorMsg = "Fetch chunk " + chunkIndex + " failed.";
- logger.error(errorMsg, e);
- exception.set(new IOException(errorMsg, e));
- } else {
- try {
- synchronized (WorkerPartitionReader.this) {
- if (!failedLocations.contains(location)) {
- failedLocations.add(location);
- client =
- new ChunkClient(
- conf,
- shuffleKey,
- location.getPeer(),
- this,
- clientFactory,
- startMapIndex,
- endMapIndex);
- currentChunkIndex = 0;
- returnedChunks = 0;
- numChunks = client.openChunks();
- fetchChunks();
- }
- }
- } catch (IOException e1) {
- logger.error(e1.getMessage(), e1);
- exception.set(new IOException(e1.getMessage(), e1));
- }
- }
+ public void onFailure(int chunkIndex, Throwable e) {
+ String errorMsg = "Fetch chunk " + chunkIndex + " failed.";
+ logger.error(errorMsg, e);
+ exception.set(new IOException(errorMsg, e));
}
};
client =
- new ChunkClient(
+ new RetryingChunkClient(
conf, shuffleKey, location, callback, clientFactory, startMapIndex, endMapIndex);
numChunks = client.openChunks();
}
- public synchronized boolean hasNext() {
+ public boolean hasNext() {
return returnedChunks < numChunks;
}
public ByteBuf next() throws IOException {
checkException();
- synchronized (this) {
- if (currentChunkIndex < numChunks) {
- fetchChunks();
- }
+ if (chunkIndex < numChunks) {
+ fetchChunks();
}
ByteBuf chunk = null;
try {
while (chunk == null) {
checkException();
- ChunkData chunkData = results.poll(500, TimeUnit.MILLISECONDS);
- if (chunkData != null) {
- synchronized (this) {
- if (failedLocations.contains(chunkData.location)) {
- chunkData.release();
- } else {
- chunk = chunkData.buf;
- returnedChunks++;
- }
- }
- }
+ chunk = results.poll(500, TimeUnit.MILLISECONDS);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
@@ -152,24 +107,26 @@ public class WorkerPartitionReader implements PartitionReader {
exception.set(ioe);
throw ioe;
}
+ returnedChunks++;
return chunk;
}
public void close() {
- closed.set(true);
+ synchronized (this) {
+ closed = true;
+ }
if (results.size() > 0) {
- results.forEach(ChunkData::release);
+ results.forEach(ReferenceCounted::release);
}
results.clear();
}
private void fetchChunks() {
- final int inFlight = currentChunkIndex - returnedChunks;
+ final int inFlight = chunkIndex - returnedChunks;
if (inFlight < fetchMaxReqsInFlight) {
- final int toFetch =
- Math.min(fetchMaxReqsInFlight - inFlight + 1, numChunks - currentChunkIndex);
+ final int toFetch = Math.min(fetchMaxReqsInFlight - inFlight + 1, numChunks - chunkIndex);
for (int i = 0; i < toFetch; i++) {
- client.fetchChunk(currentChunkIndex++);
+ client.fetchChunk(chunkIndex++);
}
}
}
@@ -180,18 +137,4 @@ public class WorkerPartitionReader implements PartitionReader {
throw e;
}
}
-
- private static class ChunkData {
- ByteBuf buf;
- PartitionLocation location;
-
- public ChunkData(ByteBuf buf, PartitionLocation location) {
- this.buf = buf;
- this.location = location;
- }
-
- public void release() {
- buf.release();
- }
- }
}
diff --git a/client/src/test/java/org/apache/celeborn/client/read/RetryingChunkClientSuiteJ.java b/client/src/test/java/org/apache/celeborn/client/read/RetryingChunkClientSuiteJ.java
new file mode 100644
index 00000000..0b85349b
--- /dev/null
+++ b/client/src/test/java/org/apache/celeborn/client/read/RetryingChunkClientSuiteJ.java
@@ -0,0 +1,423 @@
+/*
+ * 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.celeborn.client.read;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyInt;
+import static org.mockito.Mockito.anyObject;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Sets;
+import io.netty.channel.Channel;
+import org.junit.Test;
+import org.mockito.stubbing.Answer;
+
+import org.apache.celeborn.common.CelebornConf;
+import org.apache.celeborn.common.network.buffer.ManagedBuffer;
+import org.apache.celeborn.common.network.buffer.NioManagedBuffer;
+import org.apache.celeborn.common.network.client.ChunkReceivedCallback;
+import org.apache.celeborn.common.network.client.TransportClient;
+import org.apache.celeborn.common.network.client.TransportClientFactory;
+import org.apache.celeborn.common.network.client.TransportResponseHandler;
+import org.apache.celeborn.common.network.protocol.StreamHandle;
+import org.apache.celeborn.common.protocol.PartitionLocation;
+import org.apache.celeborn.common.util.ThreadUtils;
+
+public class RetryingChunkClientSuiteJ {
+
+ private static final int MASTER_RPC_PORT = 1234;
+ private static final int MASTER_PUSH_PORT = 1235;
+ private static final int MASTER_FETCH_PORT = 1236;
+ private static final int MASTER_REPLICATE_PORT = 1237;
+ private static final int SLAVE_RPC_PORT = 4321;
+ private static final int SLAVE_PUSH_PORT = 4322;
+ private static final int SLAVE_FETCH_PORT = 4323;
+ private static final int SLAVE_REPLICATE_PORT = 4324;
+ private static final PartitionLocation masterLocation =
+ new PartitionLocation(
+ 0,
+ 1,
+ "localhost",
+ MASTER_RPC_PORT,
+ MASTER_PUSH_PORT,
+ MASTER_FETCH_PORT,
+ MASTER_REPLICATE_PORT,
+ PartitionLocation.Mode.MASTER);
+ private static final PartitionLocation slaveLocation =
+ new PartitionLocation(
+ 0,
+ 1,
+ "localhost",
+ SLAVE_RPC_PORT,
+ SLAVE_PUSH_PORT,
+ SLAVE_FETCH_PORT,
+ SLAVE_REPLICATE_PORT,
+ PartitionLocation.Mode.SLAVE);
+
+ static {
+ masterLocation.setPeer(slaveLocation);
+ slaveLocation.setPeer(masterLocation);
+ }
+
+ ManagedBuffer chunk0 = new NioManagedBuffer(ByteBuffer.wrap(new byte[13]));
+ ManagedBuffer chunk1 = new NioManagedBuffer(ByteBuffer.wrap(new byte[7]));
+ ManagedBuffer chunk2 = new NioManagedBuffer(ByteBuffer.wrap(new byte[19]));
+
+ @Test
+ public void testNoFailures() throws IOException, InterruptedException {
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(chunk0))
+ .put(1, Arrays.asList(chunk1))
+ .put(2, Arrays.asList(chunk2))
+ .build();
+
+ RetryingChunkClient client = performInteractions(interactions, callback);
+
+ verify(callback, timeout(5000)).onSuccess(eq(0), eq(chunk0));
+ verify(callback, timeout(5000)).onSuccess(eq(1), eq(chunk1));
+ verify(callback, timeout(5000)).onSuccess(eq(2), eq(chunk2));
+ verifyNoMoreInteractions(callback);
+
+ assertEquals(0, client.getNumTries());
+ assertEquals(masterLocation, client.getCurrentReplica().getLocation());
+ }
+
+ @Test
+ public void testUnrecoverableFailure() throws IOException, InterruptedException {
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(new RuntimeException("Ouch!")))
+ .put(1, Arrays.asList(chunk1))
+ .put(2, Arrays.asList(chunk2))
+ .build();
+ RetryingChunkClient client = performInteractions(interactions, callback);
+
+ verify(callback, timeout(5000)).onFailure(eq(0), any());
+ verify(callback, timeout(5000)).onSuccess(eq(1), eq(chunk1));
+ verify(callback, timeout(5000)).onSuccess(eq(2), eq(chunk2));
+ verifyNoMoreInteractions(callback);
+
+ assertEquals(0, client.getNumTries());
+ assertEquals(masterLocation, client.getCurrentReplica().getLocation());
+ }
+
+ @Test
+ public void testDuplicateSuccess() throws IOException, InterruptedException {
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder().put(0, Arrays.asList(chunk0, chunk1)).build();
+ RetryingChunkClient client = performInteractions(interactions, callback);
+ verify(callback, timeout(5000)).onSuccess(eq(0), eq(chunk0));
+ verifyNoMoreInteractions(callback);
+
+ assertEquals(0, client.getNumTries());
+ assertEquals(masterLocation, client.getCurrentReplica().getLocation());
+ }
+
+ @Test
+ public void testSingleIOException() throws IOException, InterruptedException {
+ Map<Integer, Object> result = new ConcurrentHashMap<>();
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Semaphore signal = new Semaphore(3);
+ signal.acquire(3);
+
+ Answer<Void> answer =
+ invocation -> {
+ synchronized (signal) {
+ int chunkIndex = (Integer) invocation.getArguments()[0];
+ assertFalse(result.containsKey(chunkIndex));
+ Object value = invocation.getArguments()[1];
+ result.put(chunkIndex, value);
+ signal.release();
+ }
+ return null;
+ };
+ doAnswer(answer).when(callback).onSuccess(anyInt(), anyObject());
+ doAnswer(answer).when(callback).onFailure(anyInt(), anyObject());
+
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(new IOException(), chunk0))
+ .put(1, Arrays.asList(chunk1))
+ .put(2, Arrays.asList(chunk2))
+ .build();
+ RetryingChunkClient client = performInteractions(interactions, callback);
+
+ while (!signal.tryAcquire(3, 500, TimeUnit.MILLISECONDS)) ;
+ assertEquals(1, client.getNumTries());
+ assertEquals(slaveLocation, client.getCurrentReplica().getLocation());
+ assertEquals(chunk0, result.get(0));
+ assertEquals(chunk1, result.get(1));
+ assertEquals(chunk2, result.get(2));
+ }
+
+ @Test
+ public void testTwoIOExceptions() throws IOException, InterruptedException {
+ Map<Integer, Object> result = new ConcurrentHashMap<>();
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Semaphore signal = new Semaphore(3);
+ signal.acquire(3);
+
+ Answer<Void> answer =
+ invocation -> {
+ synchronized (signal) {
+ int chunkIndex = (Integer) invocation.getArguments()[0];
+ assertFalse(result.containsKey(chunkIndex));
+ Object value = invocation.getArguments()[1];
+ result.put(chunkIndex, value);
+ signal.release();
+ }
+ return null;
+ };
+ doAnswer(answer).when(callback).onSuccess(anyInt(), anyObject());
+ doAnswer(answer).when(callback).onFailure(anyInt(), anyObject());
+
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(new IOException("first ioexception"), chunk0))
+ .put(1, Arrays.asList(new IOException("second ioexception"), chunk1))
+ .put(2, Arrays.asList(chunk2))
+ .build();
+ RetryingChunkClient client = performInteractions(interactions, callback);
+
+ while (!signal.tryAcquire(3, 500, TimeUnit.MILLISECONDS)) ;
+ assertEquals(1, client.getNumTries());
+ assertEquals(slaveLocation, client.getCurrentReplica().getLocation());
+ assertEquals(chunk0, result.get(0));
+ assertEquals(chunk1, result.get(1));
+ assertEquals(chunk2, result.get(2));
+ }
+
+ @Test
+ public void testThreeIOExceptions() throws IOException, InterruptedException {
+ Map<Integer, Object> result = new ConcurrentHashMap<>();
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Semaphore signal = new Semaphore(3);
+ signal.acquire(3);
+
+ Answer<Void> answer =
+ invocation -> {
+ synchronized (signal) {
+ int chunkIndex = (Integer) invocation.getArguments()[0];
+ assertFalse(result.containsKey(chunkIndex));
+ Object value = invocation.getArguments()[1];
+ result.put(chunkIndex, value);
+ signal.release();
+ }
+ return null;
+ };
+ doAnswer(answer).when(callback).onSuccess(anyInt(), anyObject());
+ doAnswer(answer).when(callback).onFailure(anyInt(), anyObject());
+
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(new IOException("first ioexception"), chunk0))
+ .put(1, Arrays.asList(new IOException("second ioexception"), chunk1))
+ .put(2, Arrays.asList(new IOException("third ioexception"), chunk2))
+ .build();
+
+ RetryingChunkClient client = performInteractions(interactions, callback);
+ while (!signal.tryAcquire(3, 500, TimeUnit.MILLISECONDS)) ;
+ assertEquals(1, client.getNumTries());
+ assertEquals(slaveLocation, client.getCurrentReplica().getLocation());
+ assertEquals(chunk0, result.get(0));
+ assertEquals(chunk1, result.get(1));
+ assertEquals(chunk2, result.get(2));
+ }
+
+ @Test
+ public void testFailedWithIOExceptions() throws IOException, InterruptedException {
+ Map<Integer, Object> result = new ConcurrentHashMap<>();
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Semaphore signal = new Semaphore(3);
+ signal.acquire(3);
+
+ Answer<Void> answer =
+ invocation -> {
+ synchronized (signal) {
+ int chunkIndex = (Integer) invocation.getArguments()[0];
+ assertFalse(result.containsKey(chunkIndex));
+ Object value = invocation.getArguments()[1];
+ result.put(chunkIndex, value);
+ signal.release();
+ }
+ return null;
+ };
+ doAnswer(answer).when(callback).onSuccess(anyInt(), anyObject());
+ doAnswer(answer).when(callback).onFailure(anyInt(), anyObject());
+
+ IOException ioe = new IOException("failed exception");
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(ioe, ioe, ioe, ioe, ioe, chunk0))
+ .put(1, Arrays.asList(ioe, ioe, ioe, ioe, ioe, chunk1))
+ .put(2, Arrays.asList(ioe, ioe, ioe, ioe, ioe, chunk2))
+ .build();
+
+ RetryingChunkClient client = performInteractions(interactions, callback);
+ while (!signal.tryAcquire(3, 500, TimeUnit.MILLISECONDS)) ;
+ // Note: this may exceeds the max retries we want, but it doesn't master.
+ assertEquals(4, client.getNumTries());
+ assertEquals(masterLocation, client.getCurrentReplica().getLocation());
+ assertEquals(ioe, result.get(0));
+ assertEquals(ioe, result.get(1));
+ assertEquals(ioe, result.get(2));
+ }
+
+ @Test
+ public void testRetryAndUnrecoverable() throws IOException, InterruptedException {
+ Map<Integer, Object> result = new ConcurrentHashMap<>();
+ ChunkReceivedCallback callback = mock(ChunkReceivedCallback.class);
+ Semaphore signal = new Semaphore(3);
+ signal.acquire(3);
+
+ Answer<Void> answer =
+ invocation -> {
+ synchronized (signal) {
+ int chunkIndex = (Integer) invocation.getArguments()[0];
+ assertFalse(result.containsKey(chunkIndex));
+ Object value = invocation.getArguments()[1];
+ result.put(chunkIndex, value);
+ signal.release();
+ }
+ return null;
+ };
+ doAnswer(answer).when(callback).onSuccess(anyInt(), anyObject());
+ doAnswer(answer).when(callback).onFailure(anyInt(), anyObject());
+
+ Exception re = new RuntimeException("failed exception");
+ Map<Integer, List<Object>> interactions =
+ ImmutableMap.<Integer, List<Object>>builder()
+ .put(0, Arrays.asList(new IOException("first ioexception"), re, chunk0))
+ .put(1, Arrays.asList(chunk1))
+ .put(2, Arrays.asList(new IOException("second ioexception"), chunk2))
+ .build();
+
+ performInteractions(interactions, callback);
+ while (!signal.tryAcquire(3, 500, TimeUnit.MILLISECONDS)) ;
+ assertEquals(re, result.get(0));
+ assertEquals(chunk1, result.get(1));
+ assertEquals(chunk2, result.get(2));
+ }
+
+ private static RetryingChunkClient performInteractions(
+ Map<Integer, List<Object>> interactions, ChunkReceivedCallback callback)
+ throws IOException, InterruptedException {
+ CelebornConf conf = new CelebornConf();
+ conf.set("celeborn.data.io.maxRetries", "1");
+ conf.set("celeborn.data.io.retryWait", "0");
+
+ // Contains all chunk ids that are referenced across all interactions.
+ LinkedHashSet<Integer> chunkIds = Sets.newLinkedHashSet(interactions.keySet());
+
+ final TransportClient client = new DummyTransportClient(chunkIds.size(), interactions);
+ final TransportClientFactory clientFactory = mock(TransportClientFactory.class);
+ doAnswer(invocation -> client).when(clientFactory).createClient(anyString(), anyInt());
+
+ RetryingChunkClient retryingChunkClient =
+ new RetryingChunkClient(conf, "test", masterLocation, callback, clientFactory);
+ chunkIds.stream().sorted().forEach(retryingChunkClient::fetchChunk);
+ return retryingChunkClient;
+ }
+
+ private static class DummyTransportClient extends TransportClient {
+
+ private static final Channel channel = mock(Channel.class);
+ private static final TransportResponseHandler handler = mock(TransportResponseHandler.class);
+
+ private final long streamId = new Random().nextInt(Integer.MAX_VALUE) * 1000L;
+ private final int numChunks;
+ private final Map<Integer, List<Object>> interactions;
+ private final Map<Integer, Integer> chunkIdToInterActionIndex;
+
+ private final ScheduledExecutorService schedule =
+ ThreadUtils.newDaemonThreadPoolScheduledExecutor("test-fetch-chunk", 3);
+
+ DummyTransportClient(int numChunks, Map<Integer, List<Object>> interactions) {
+ super(channel, handler);
+ this.numChunks = numChunks;
+ this.interactions = interactions;
+ this.chunkIdToInterActionIndex = new ConcurrentHashMap<>();
+ interactions.keySet().forEach((chunkId) -> chunkIdToInterActionIndex.putIfAbsent(chunkId, 0));
+ }
+
+ @Override
+ public void fetchChunk(long streamId, int chunkId, ChunkReceivedCallback callback) {
+ schedule.schedule(
+ () -> {
+ Object action;
+ List<Object> interaction = interactions.get(chunkId);
+ synchronized (chunkIdToInterActionIndex) {
+ int index = chunkIdToInterActionIndex.get(chunkId);
+ assertTrue(index < interaction.size());
+ action = interaction.get(index);
+ chunkIdToInterActionIndex.put(chunkId, index + 1);
+ }
+
+ if (action instanceof ManagedBuffer) {
+ callback.onSuccess(chunkId, (ManagedBuffer) action);
+ } else if (action instanceof Exception) {
+ callback.onFailure(chunkId, (Exception) action);
+ } else {
+ fail("Can only handle ManagedBuffers and Exceptions, got " + action);
+ }
+ },
+ 500,
+ TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public ByteBuffer sendRpcSync(ByteBuffer message, long timeoutMs) {
+ StreamHandle handle = new StreamHandle(streamId, numChunks);
+ return handle.toByteBuffer();
+ }
+
+ @Override
+ public void close() {
+ super.close();
+ schedule.shutdownNow();
+ }
+ }
+}
diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/ChunkReceivedCallback.java b/common/src/main/java/org/apache/celeborn/common/network/client/ChunkReceivedCallback.java
index 76017ad8..267a63df 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/client/ChunkReceivedCallback.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/client/ChunkReceivedCallback.java
@@ -18,7 +18,6 @@
package org.apache.celeborn.common.network.client;
import org.apache.celeborn.common.network.buffer.ManagedBuffer;
-import org.apache.celeborn.common.protocol.PartitionLocation;
/**
* Callback for the result of a single chunk result. For a single stream, the callbacks are
@@ -35,7 +34,7 @@ public interface ChunkReceivedCallback {
* this call returns. You must therefore either retain() the buffer or copy its contents before
* returning.
*/
- void onSuccess(int chunkIndex, ManagedBuffer buffer, PartitionLocation location);
+ void onSuccess(int chunkIndex, ManagedBuffer buffer);
/**
* Called upon failure to fetch a particular chunk. Note that this may actually be called due to
@@ -44,5 +43,5 @@ public interface ChunkReceivedCallback {
* <p>After receiving a failure, the stream may or may not be valid. The client should not assume
* that the server's side of the stream has been closed.
*/
- void onFailure(int chunkIndex, PartitionLocation location, Throwable e);
+ void onFailure(int chunkIndex, Throwable e);
}
diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
index e778449a..e5e82db0 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportClient.java
@@ -126,7 +126,7 @@ public class TransportClient implements Closeable {
@Override
protected void handleFailure(String errorMsg, Throwable cause) {
handler.removeFetchRequest(streamChunkSlice);
- callback.onFailure(chunkIndex, null, new IOException(errorMsg, cause));
+ callback.onFailure(chunkIndex, new IOException(errorMsg, cause));
}
};
handler.addFetchRequest(streamChunkSlice, callback);
diff --git a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
index 1a9ab600..c55fbb3b 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/client/TransportResponseHandler.java
@@ -86,7 +86,7 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
private void failOutstandingRequests(Throwable cause) {
for (Map.Entry<StreamChunkSlice, ChunkReceivedCallback> entry : outstandingFetches.entrySet()) {
try {
- entry.getValue().onFailure(entry.getKey().chunkIndex, null, cause);
+ entry.getValue().onFailure(entry.getKey().chunkIndex, cause);
} catch (Exception e) {
logger.warn("ChunkReceivedCallback.onFailure throws exception", e);
}
@@ -144,7 +144,7 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
resp.body().release();
} else {
outstandingFetches.remove(resp.streamChunkSlice);
- listener.onSuccess(resp.streamChunkSlice.chunkIndex, resp.body(), null);
+ listener.onSuccess(resp.streamChunkSlice.chunkIndex, resp.body());
resp.body().release();
}
} else if (message instanceof ChunkFetchFailure) {
@@ -161,7 +161,6 @@ public class TransportResponseHandler extends MessageHandler<ResponseMessage> {
logger.warn("Receive ChunkFetchFailure, errorMsg {}", resp.errorString);
listener.onFailure(
resp.streamChunkSlice.chunkIndex,
- null,
new ChunkFetchFailureException(
"Failure while fetching " + resp.streamChunkSlice + ": " + resp.errorString));
}
diff --git a/common/src/main/java/org/apache/celeborn/common/network/server/OneForOneStreamManager.java b/common/src/main/java/org/apache/celeborn/common/network/server/OneForOneStreamManager.java
index dec24f8e..838f7459 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/server/OneForOneStreamManager.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/server/OneForOneStreamManager.java
@@ -25,6 +25,8 @@ import java.util.concurrent.atomic.AtomicLong;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.netty.channel.Channel;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -91,20 +93,33 @@ public class OneForOneStreamManager extends StreamManager {
// Normally, when all chunks are returned to the client, the stream should be removed here.
// But if there is a switch on the client side, it will not go here at this time, so we need
// to remove the stream when the connection is terminated, and release the unused buffer.
- logger.debug("Removing stream id {}", streamId);
+ logger.trace("Removing stream id {}", streamId);
streams.remove(streamId);
}
return nextChunk;
}
+ public static String genStreamChunkId(long streamId, int chunkId) {
+ return String.format("%d_%d", streamId, chunkId);
+ }
+
+ // Parse streamChunkId to be stream id and chunk id. This is used when fetch remote chunk as a
+ // stream.
+ public static Pair<Long, Integer> parseStreamChunkId(String streamChunkId) {
+ String[] array = streamChunkId.split("_");
+ assert array.length == 2 : "Stream id and chunk index should be specified.";
+ long streamId = Long.parseLong(array[0]);
+ int chunkIndex = Integer.parseInt(array[1]);
+ return ImmutablePair.of(streamId, chunkIndex);
+ }
+
@Override
public void connectionTerminated(Channel channel) {
// Close all streams which have been associated with the channel.
for (Map.Entry<Long, StreamState> entry : streams.entrySet()) {
StreamState state = entry.getValue();
if (state.associatedChannel == channel) {
- logger.debug("Remove stream id {} of channel {}", entry.getKey(), channel.remoteAddress());
streams.remove(entry.getKey());
}
}
@@ -118,6 +133,11 @@ public class OneForOneStreamManager extends StreamManager {
}
}
+ @Override
+ public void streamBeingSent(String streamId) {
+ chunkBeingSent(parseStreamChunkId(streamId).getLeft());
+ }
+
@Override
public void chunkSent(long streamId) {
StreamState streamState = streams.get(streamId);
@@ -126,6 +146,11 @@ public class OneForOneStreamManager extends StreamManager {
}
}
+ @Override
+ public void streamSent(String streamId) {
+ chunkSent(OneForOneStreamManager.parseStreamChunkId(streamId).getLeft());
+ }
+
@Override
public long chunksBeingTransferred() {
long sum = 0L;
diff --git a/common/src/main/java/org/apache/celeborn/common/network/server/StreamManager.java b/common/src/main/java/org/apache/celeborn/common/network/server/StreamManager.java
index f48ebcc9..b73f31c3 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/server/StreamManager.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/server/StreamManager.java
@@ -45,6 +45,20 @@ public abstract class StreamManager {
*/
public abstract ManagedBuffer getChunk(long streamId, int chunkIndex, int offset, int len);
+ /**
+ * Called in response to a stream() request. The returned data is streamed to the client through a
+ * single TCP connection.
+ *
+ * <p>Note the <code>streamId</code> argument is not related to the similarly named argument in
+ * the {@link #getChunk(long, int, int, int)} method.
+ *
+ * @param streamId id of a stream that has been previously registered with the StreamManager.
+ * @return A managed buffer for the stream, or null if the stream was not found.
+ */
+ public ManagedBuffer openStream(String streamId) {
+ throw new UnsupportedOperationException();
+ }
+
/**
* Indicates that the given channel has been terminated. After this occurs, we are guaranteed not
* to read from the associated streams again, so any state can be cleaned up.
@@ -59,6 +73,12 @@ public abstract class StreamManager {
/** Called when start sending a chunk. */
public void chunkBeingSent(long streamId) {}
+ /** Called when start sending a stream. */
+ public void streamBeingSent(String streamId) {}
+
/** Called when a chunk is successfully sent. */
public void chunkSent(long streamId) {}
+
+ /** Called when a stream is successfully sent. */
+ public void streamSent(String streamId) {}
}
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index ffcea627..6767cd3d 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -665,7 +665,6 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def rpcCacheSize: Int = get(RPC_CACHE_SIZE)
def rpcCacheConcurrencyLevel: Int = get(RPC_CACHE_CONCURRENCY_LEVEL)
def rpcCacheExpireTime: Long = get(RPC_CACHE_EXPIRE_TIME)
- def testFetchFailedChunkIndex: Int = get(TEST_FETCH_FAILED_CHUNK_INDEX)
// //////////////////////////////////////////////////////
// Graceful Shutdown & Recover //
@@ -2570,13 +2569,4 @@ object CelebornConf extends Logging {
.doc("The time before a cache item is removed.")
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("15s")
-
- val TEST_FETCH_FAILED_CHUNK_INDEX: ConfigEntry[Int] =
- buildConf("celeborn.test.client.fetchFailedChuckIndex")
- .categories("client")
- .version("0.2.0")
- .internal
- .doc("The chunk index to trigger fetch chunk failure for testing purpose only.")
- .intConf
- .createWithDefault(0)
}
diff --git a/common/src/test/java/org/apache/celeborn/common/network/ChunkFetchIntegrationSuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/ChunkFetchIntegrationSuiteJ.java
index ec9722a8..dd8c90d5 100644
--- a/common/src/test/java/org/apache/celeborn/common/network/ChunkFetchIntegrationSuiteJ.java
+++ b/common/src/test/java/org/apache/celeborn/common/network/ChunkFetchIntegrationSuiteJ.java
@@ -47,7 +47,6 @@ import org.apache.celeborn.common.network.server.BaseMessageHandler;
import org.apache.celeborn.common.network.server.StreamManager;
import org.apache.celeborn.common.network.server.TransportServer;
import org.apache.celeborn.common.network.util.TransportConf;
-import org.apache.celeborn.common.protocol.PartitionLocation;
public class ChunkFetchIntegrationSuiteJ {
static final long STREAM_ID = 1;
@@ -154,7 +153,7 @@ public class ChunkFetchIntegrationSuiteJ {
ChunkReceivedCallback callback =
new ChunkReceivedCallback() {
@Override
- public void onSuccess(int chunkIndex, ManagedBuffer buffer, PartitionLocation location) {
+ public void onSuccess(int chunkIndex, ManagedBuffer buffer) {
buffer.retain();
res.successChunks.add(chunkIndex);
res.buffers.add(buffer);
@@ -162,7 +161,7 @@ public class ChunkFetchIntegrationSuiteJ {
}
@Override
- public void onFailure(int chunkIndex, PartitionLocation location, Throwable e) {
+ public void onFailure(int chunkIndex, Throwable e) {
res.failedChunks.add(chunkIndex);
sem.release();
}
diff --git a/common/src/test/java/org/apache/celeborn/common/network/RequestTimeoutIntegrationSuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/RequestTimeoutIntegrationSuiteJ.java
index 2d66bbab..5933a5af 100644
--- a/common/src/test/java/org/apache/celeborn/common/network/RequestTimeoutIntegrationSuiteJ.java
+++ b/common/src/test/java/org/apache/celeborn/common/network/RequestTimeoutIntegrationSuiteJ.java
@@ -42,7 +42,6 @@ import org.apache.celeborn.common.network.server.BaseMessageHandler;
import org.apache.celeborn.common.network.server.StreamManager;
import org.apache.celeborn.common.network.server.TransportServer;
import org.apache.celeborn.common.network.util.TransportConf;
-import org.apache.celeborn.common.protocol.PartitionLocation;
/**
* Suite which ensures that requests that go without a response for the network timeout period are
@@ -268,7 +267,7 @@ public class RequestTimeoutIntegrationSuiteJ {
}
@Override
- public void onSuccess(int chunkIndex, ManagedBuffer buffer, PartitionLocation location) {
+ public void onSuccess(int chunkIndex, ManagedBuffer buffer) {
try {
successLength = buffer.nioByteBuffer().remaining();
} catch (IOException e) {
@@ -279,7 +278,7 @@ public class RequestTimeoutIntegrationSuiteJ {
}
@Override
- public void onFailure(int chunkIndex, PartitionLocation location, Throwable e) {
+ public void onFailure(int chunkIndex, Throwable e) {
failure = e;
latch.countDown();
}
diff --git a/common/src/test/java/org/apache/celeborn/common/network/TransportResponseHandlerSuiteJ.java b/common/src/test/java/org/apache/celeborn/common/network/TransportResponseHandlerSuiteJ.java
index 661c7e35..16cacbd1 100644
--- a/common/src/test/java/org/apache/celeborn/common/network/TransportResponseHandlerSuiteJ.java
+++ b/common/src/test/java/org/apache/celeborn/common/network/TransportResponseHandlerSuiteJ.java
@@ -42,7 +42,7 @@ public class TransportResponseHandlerSuiteJ {
assertEquals(1, handler.numOutstandingRequests());
handler.handle(new ChunkFetchSuccess(streamChunkSlice, new TestManagedBuffer(123)));
- verify(callback, times(1)).onSuccess(eq(0), any(), any());
+ verify(callback, times(1)).onSuccess(eq(0), any());
assertEquals(0, handler.numOutstandingRequests());
}
@@ -55,7 +55,7 @@ public class TransportResponseHandlerSuiteJ {
assertEquals(1, handler.numOutstandingRequests());
handler.handle(new ChunkFetchFailure(streamChunkSlice, "some error msg"));
- verify(callback, times(1)).onFailure(eq(0), any(), any());
+ verify(callback, times(1)).onFailure(eq(0), any());
assertEquals(0, handler.numOutstandingRequests());
}
@@ -72,9 +72,9 @@ public class TransportResponseHandlerSuiteJ {
handler.exceptionCaught(new Exception("duh duh duhhhh"));
// should fail both b2 and b3
- verify(callback, times(1)).onSuccess(eq(0), any(), any());
- verify(callback, times(1)).onFailure(eq(1), any(), any());
- verify(callback, times(1)).onFailure(eq(2), any(), any());
+ verify(callback, times(1)).onSuccess(eq(0), any());
+ verify(callback, times(1)).onFailure(eq(1), any());
+ verify(callback, times(1)).onFailure(eq(2), any());
assertEquals(0, handler.numOutstandingRequests());
}
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
index f587ae4e..57eaae64 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala
@@ -104,9 +104,6 @@ class FetchHandler(val conf: TransportConf) extends BaseMessageHandler with Logg
if (fileInfo.numChunks() == 0) {
logDebug(s"StreamId $streamId fileName $fileName startMapIndex" +
s" $startMapIndex endMapIndex $endMapIndex is empty.")
- } else {
- logDebug(s"StreamId $streamId fileName $fileName numChunks ${fileInfo.numChunks()} " +
- s"startMapIndex $startMapIndex endMapIndex $endMapIndex")
}
client.getChannel.writeAndFlush(new RpcResponse(
request.requestId,
diff --git a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/FileWriterSuiteJ.java b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/FileWriterSuiteJ.java
index fc97f704..9f7fa994 100644
--- a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/FileWriterSuiteJ.java
+++ b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/FileWriterSuiteJ.java
@@ -71,7 +71,6 @@ import org.apache.celeborn.common.network.server.MemoryTracker;
import org.apache.celeborn.common.network.server.TransportServer;
import org.apache.celeborn.common.network.util.JavaUtils;
import org.apache.celeborn.common.network.util.TransportConf;
-import org.apache.celeborn.common.protocol.PartitionLocation;
import org.apache.celeborn.common.protocol.PartitionSplitMode;
import org.apache.celeborn.common.protocol.PartitionType;
import org.apache.celeborn.common.protocol.StorageInfo;
@@ -210,7 +209,7 @@ public class FileWriterSuiteJ {
ChunkReceivedCallback callback =
new ChunkReceivedCallback() {
@Override
- public void onSuccess(int chunkIndex, ManagedBuffer buffer, PartitionLocation location) {
+ public void onSuccess(int chunkIndex, ManagedBuffer buffer) {
buffer.retain();
res.successChunks.add(chunkIndex);
res.buffers.add(buffer);
@@ -218,7 +217,7 @@ public class FileWriterSuiteJ {
}
@Override
- public void onFailure(int chunkIndex, PartitionLocation location, Throwable e) {
+ public void onFailure(int chunkIndex, Throwable e) {
res.failedChunks.add(chunkIndex);
sem.release();
}