jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix for immediate query fail due to unhandled exception in httpRequest for resumable failure

This commit is contained in:
Ahana 2022-05-18 10:15:15 +05:30
parent 56305dcfc2
commit ff420218b8
3 changed files with 83 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
presto-main/src/main/java/io/prestosql/operator/HttpPageBufferClient.java
View File

@ -431,7 +431,7 @@ public final class HttpPageBufferClient
Throwable throwable = rewriteException(t);
boolean fail = failureRetryPolicy.hasFailed(fromUri(uri));
if (!(throwable instanceof PrestoException) && fail) {
if ((!(throwable instanceof PrestoException) || (throwable instanceof PageTransportServerException)) && fail) {
String message = format("%s (%s - %s failures, failure duration %s, total failed request time %s)",
WORKER_NODE_ERROR,
uri,
@ -447,7 +447,7 @@ public final class HttpPageBufferClient
}
throwable = new PageTransportTimeoutException(fromUri(uri), message, throwable);
}
handleFailure(throwable, resultFuture);
handleFailure(throwable, resultFuture, fail);
}
}, pageBufferClientCallbackExecutor);
}
@ -505,7 +505,7 @@ public final class HttpPageBufferClient
checkState(!Thread.holdsLock(lock), "Cannot execute this method while holding a lock");
}
private void handleFailure(Throwable t, HttpResponseFuture<?> expectedFuture)
private void handleFailure(Throwable t, HttpResponseFuture<?> expectedFuture, boolean failed)
{
// Can not delegate to other callback while holding a lock on this
checkNotHoldsLock(this);
@ -513,7 +513,8 @@ public final class HttpPageBufferClient
requestsFailed.incrementAndGet();
requestsCompleted.incrementAndGet();
if (t instanceof PrestoException) {
if ((t instanceof PageTransportServerException && failed) // resumable failure, failed even after retries
|| (!(t instanceof PageTransportServerException)) && t instanceof PrestoException) {
clientCallback.clientFailed(HttpPageBufferClient.this, t);
}
@ -526,6 +527,11 @@ public final class HttpPageBufferClient
clientCallback.requestComplete(HttpPageBufferClient.this);
}
private void handleFailure(Throwable t, HttpResponseFuture<?> expectedFuture)
{
handleFailure(t, expectedFuture, true);
}
@Override
public boolean equals(Object o)
{
@ -595,7 +601,7 @@ public final class HttpPageBufferClient
{
if (exception instanceof RuntimeException || exception instanceof IOException) {
throw propagate(request,
new PageTransportErrorException(format("%s %d! Error fetching %s: %s",
new PageTransportServerException(format("%s %d! Error fetching %s: %s",
PAGE_TRANSPORT_ERROR_PREFIX, HttpStatus.INTERNAL_SERVER_ERROR.code(),
request.getUri().toASCIIString(), exception.getMessage()),
exception));

33
presto-main/src/main/java/io/prestosql/operator/PageTransportServerException.java Normal file
View File

@ -0,0 +1,33 @@
/*
* Copyright (C) 2018-2021. Huawei Technologies Co., Ltd. All rights reserved.
* Licensed 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 io.prestosql.operator;
import io.prestosql.spi.PrestoException;
import static io.prestosql.spi.StandardErrorCode.PAGE_TRANSPORT_ERROR;
public class PageTransportServerException
extends PrestoException
{
public PageTransportServerException(String message)
{
super(PAGE_TRANSPORT_ERROR, message);
}
public PageTransportServerException(String message, Throwable cause)
{
super(PAGE_TRANSPORT_ERROR, message, cause);
}
}

39
presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java
View File

@ -298,6 +298,45 @@ public class TestHttpPageBufferClient
assertStatus(client, location, "closed", 0, 1, 2, 1, "not scheduled");
}
@Test
public void testResumableServerFailures()
throws Exception
{
TestingTicker ticker = new TestingTicker();
AtomicReference<Duration> tickerIncrement = new AtomicReference<>(new Duration(0, TimeUnit.SECONDS));
TestingHttpClient.Processor processor = (input) -> {
Duration delta = tickerIncrement.get();
ticker.increment(delta.toMillis(), TimeUnit.MILLISECONDS);
throw new PageTransportServerException("Foo");
};
CyclicBarrier requestComplete = new CyclicBarrier(2);
TestingClientCallback callback = new TestingClientCallback(requestComplete);
URI location = URI.create("http://localhost:8080");
String instanceId = "testing instance id";
HttpPageBufferClient client = new HttpPageBufferClient(new TestingHttpClient(processor, scheduler),
new DataSize(10, Unit.MEGABYTE),
true,
new TaskLocation(location, instanceId),
callback,
scheduler,
pageBufferClientCallbackExecutor,
false,
null, failureDetectorManager);
assertStatus(client, location, "queued", 0, 0, 0, 0, "not scheduled");
client.scheduleRequest();
requestComplete.await(10, TimeUnit.SECONDS);
assertEquals(callback.getPages().size(), 0);
assertEquals(callback.getCompletedRequests(), 1);
assertEquals(callback.getFinishedBuffers(), 0);
assertEquals(callback.getFailedBuffers(), 0);
assertStatus(client, location, "queued", 0, 1, 1, 1, "not scheduled");
}
@Test
public void testInvalidResponses()
throws Exception