This is an automated email from the ASF dual-hosted git repository.
hanicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/knox.git
The following commit(s) were added to refs/heads/master by this push:
new 2beb45b33 KNOX-3041: Eliminate load-balancing race condition in HA
dispatch (#1317)
2beb45b33 is described below
commit 2beb45b33c09d5957fc47dd8a1e7a9e951391967
Author: hanicz <[email protected]>
AuthorDate: Thu Jul 23 19:21:11 2026 +0200
KNOX-3041: Eliminate load-balancing race condition in HA dispatch (#1317)
---
.../knox/gateway/ha/dispatch/CommonHaDispatch.java | 145 ++++++--
.../ha/dispatch/ConfigurableHADispatch.java | 5 +-
.../knox/gateway/ha/dispatch/SSEHaDispatch.java | 11 -
.../knox/gateway/ha/provider/HaProvider.java | 11 +
.../knox/gateway/ha/provider/URLManager.java | 2 +
.../ha/provider/impl/BaseZookeeperURLManager.java | 13 +
.../ha/provider/impl/DefaultHaProvider.java | 22 ++
.../ha/provider/impl/DefaultURLManager.java | 10 +
.../ha/dispatch/ConfigurableHADispatchTest.java | 385 +++++++++++++++++++++
.../gateway/ha/dispatch/SSEHaDispatchTest.java | 37 +-
.../org/apache/knox/gateway/sse/SSEDispatch.java | 5 -
11 files changed, 570 insertions(+), 76 deletions(-)
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/CommonHaDispatch.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/CommonHaDispatch.java
index 2918c099e..66e9f5e13 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/CommonHaDispatch.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/CommonHaDispatch.java
@@ -135,7 +135,7 @@ public interface CommonHaDispatch {
// Make sure that the url provided is actually a valid
backend url
if
(getHaConfigurations().getHaProvider().getURLs(getServiceRole()).contains(backendURL))
{
try {
- return
Optional.of(updateHostURL(outboundRequest.getURI(), backendURL));
+ return
Optional.of(updateBackendURL(outboundRequest.getURI(), backendURL));
} catch (URISyntaxException ignore) {
// The cookie was invalid so we just don't set it.
Knox will pick a backend automatically
}
@@ -169,6 +169,79 @@ public interface CommonHaDispatch {
return uriBuilder.build();
}
+ /**
+ * Re-targets a rewritten URI at a different backend, carrying over the
backend's base
+ * path in addition to scheme, host and port.
+ */
+ default URI updateBackendURL(final URI source, final String newBackend)
throws URISyntaxException {
+ final URI newUri = new URI(newBackend);
+ final URIBuilder uriBuilder = new URIBuilder(source);
+ uriBuilder.setScheme(newUri.getScheme());
+ uriBuilder.setHost(newUri.getHost());
+ uriBuilder.setPort(newUri.getPort());
+ final String newBasePath = normalizeBasePath(newUri.getPath());
+ final String oldBasePath = matchConfiguredBasePath(source);
+ if (oldBasePath != null && !oldBasePath.equals(newBasePath)) {
+ final String sourcePath = source.getPath() == null ? "" :
source.getPath();
+ uriBuilder.setPath(newBasePath +
sourcePath.substring(oldBasePath.length()));
+ }
+ return uriBuilder.build();
+ }
+
+ default String matchConfiguredBasePath(final URI source) {
+ if (source.getHost() == null) {
+ return null;
+ }
+ final String sourcePath = source.getPath() == null ? "" :
source.getPath();
+ String best = null;
+ for (String url :
getHaConfigurations().getHaProvider().getURLs(getServiceRole())) {
+ try {
+ final URI poolUri = new URI(url);
+ if (!source.getHost().equals(poolUri.getHost()) ||
effectivePort(source) != effectivePort(poolUri)) {
+ continue;
+ }
+ final String basePath = normalizeBasePath(poolUri.getPath());
+ if (isPathPrefix(sourcePath, basePath) && (best == null ||
basePath.length() > best.length())) {
+ best = basePath;
+ }
+ } catch (URISyntaxException ignore) {
+ // a malformed pool entry cannot be the URL the rewriter used
+ }
+ }
+ return best;
+ }
+
+ static int effectivePort(final URI uri) {
+ final int port = uri.getPort();
+ if (port != -1) {
+ return port;
+ }
+ final String scheme = uri.getScheme();
+ if ("http".equalsIgnoreCase(scheme)) {
+ return 80;
+ }
+ if ("https".equalsIgnoreCase(scheme)) {
+ return 443;
+ }
+ return -1;
+ }
+
+ static String normalizeBasePath(final String path) {
+ if (path == null) {
+ return "";
+ }
+ String normalized = path;
+ while (normalized.endsWith("/")) {
+ normalized = normalized.substring(0, normalized.length() - 1);
+ }
+ return normalized;
+ }
+
+ static boolean isPathPrefix(final String path, final String prefix) {
+ return path.startsWith(prefix)
+ && (path.length() == prefix.length() ||
path.charAt(prefix.length()) == '/');
+ }
+
default void setupUrlHashLookup() {
for (String url :
getHaConfigurations().getHaProvider().getURLs(getServiceRole())) {
String urlHash = hash(url);
@@ -199,46 +272,33 @@ public interface CommonHaDispatch {
backendURI.ifPresent(uri -> ((HttpRequestBase)
outboundRequest).setURI(uri));
}
- /**
- * case where loadbalancing is enabled
- * and we have a HTTP request configured not to use LB
- * use the activeURL
- */
- if (getHaConfigurations().isLoadBalancingEnabled() &&
userAgentDisabled) {
- try {
- ((HttpRequestBase)
outboundRequest).setURI(updateHostURL(outboundRequest.getURI(),
getActiveURL().get()));
- } catch (final URISyntaxException e) {
- LOG.errorSettingActiveUrl();
+ if (getHaConfigurations().isLoadBalancingEnabled()) {
+ if (userAgentDisabled) {
+ /**
+ * case where loadbalancing is enabled
+ * and we have a HTTP request configured not to use LB
+ * use the activeURL
+ */
+ try {
+ ((HttpRequestBase)
outboundRequest).setURI(updateHostURL(outboundRequest.getURI(),
getActiveURL().get()));
+ } catch (final URISyntaxException e) {
+ LOG.errorSettingActiveUrl();
+ }
+ } else if (!backendURI.isPresent()) {
+ String nextURL =
getHaConfigurations().getHaProvider().getActiveURLAndAdvance(getServiceRole());
+ if (nextURL != null) {
+ try {
+ ((HttpRequestBase)
outboundRequest).setURI(updateBackendURL(outboundRequest.getURI(), nextURL));
+ } catch (final URISyntaxException e) {
+ LOG.errorSettingActiveUrl();
+ }
+ }
}
}
return backendURI;
}
- default void shiftActiveURL(boolean userAgentDisabled, Optional<URI>
backendURI) {
- /**
- * 1. Load balance when loadbalancing is enabled and there are no
overrides (disableLB)
- * 2. Loadbalance only when sticky session is enabled but cookie not
detected
- * i.e. when loadbalancing is enabled every request that does not
have BACKEND cookie
- * needs to be loadbalanced. If a request has BACKEND coookie and
Loadbalance=on then
- * there should be no loadbalancing.
- */
- if (getHaConfigurations().isLoadBalancingEnabled() &&
!userAgentDisabled) {
- /* check sticky session enabled */
- if (getHaConfigurations().isStickySessionEnabled()) {
- /* loadbalance only when sticky session enabled and no backend
url cookie */
- if (!backendURI.isPresent()) {
-
getHaConfigurations().getHaProvider().makeNextActiveURLAvailable(getServiceRole());
- } else {
- /* sticky session enabled and backend url cookie is valid
no need to loadbalance */
- /* do nothing */
- }
- } else {
-
getHaConfigurations().getHaProvider().makeNextActiveURLAvailable(getServiceRole());
- }
- }
- }
-
/**
* A helper method that marks an endpoint failed.
* Changes HA Provider state.
@@ -271,8 +331,19 @@ public interface CommonHaDispatch {
inboundRequest.setAttribute(AbstractGatewayFilter.TARGET_REQUEST_URL_ATTRIBUTE_NAME,
null);
// Make sure to remove the ha cookie from the request
inboundRequest = new
StickySessionCookieRemovedRequest(getHaConfigurations().getStickySessionCookieName(),
inboundRequest);
- URI uri = getDispatchUrl(inboundRequest);
- ((HttpRequestBase) outboundRequest).setURI(uri);
+ ((HttpRequestBase)
outboundRequest).setURI(getDispatchUrl(inboundRequest));
+
+ if (getHaConfigurations().isLoadBalancingEnabled() &&
!isUserAgentDisabled(inboundRequest)) {
+ final String nextURL =
getHaConfigurations().getHaProvider().getActiveURLAndAdvance(getServiceRole());
+ if (nextURL != null) {
+ try {
+ ((HttpRequestBase)
outboundRequest).setURI(updateBackendURL(outboundRequest.getURI(), nextURL));
+ } catch (final URISyntaxException e) {
+ LOG.errorSettingActiveUrl();
+ }
+ }
+ }
+
if (getHaConfigurations().getFailoverSleep() > 0) {
try {
Thread.sleep(getHaConfigurations().getFailoverSleep());
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatch.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatch.java
index 5c24f81a6..a4b7df94c 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatch.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatch.java
@@ -31,8 +31,6 @@ import org.apache.knox.gateway.i18n.messages.MessagesFactory;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
-import java.net.URI;
-import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -92,9 +90,8 @@ public class ConfigurableHADispatch extends
ConfigurableDispatch implements Comm
@Override
protected void executeRequestWrapper(HttpUriRequest outboundRequest,
HttpServletRequest inboundRequest, HttpServletResponse outboundResponse) throws
IOException {
boolean userAgentDisabled = isUserAgentDisabled(inboundRequest);
- Optional<URI> backendURI = setBackendUri(outboundRequest,
inboundRequest, userAgentDisabled);
+ setBackendUri(outboundRequest, inboundRequest, userAgentDisabled);
executeRequest(outboundRequest, inboundRequest, outboundResponse);
- shiftActiveURL(userAgentDisabled, backendURI);
}
@Override
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatch.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatch.java
index 136a1721f..40e7f8c18 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatch.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatch.java
@@ -39,7 +39,6 @@ import javax.servlet.FilterConfig;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
-import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -144,14 +143,4 @@ public class SSEHaDispatch extends SSEDispatch implements
CommonHaDispatch {
protected void outboundResponseWrapper(final HttpUriRequest
outboundRequest, final HttpServletRequest inboundRequest, final
HttpServletResponse outboundResponse) {
setKnoxHaCookie(outboundRequest, inboundRequest, outboundResponse,
sslEnabled);
}
-
- @Override
- protected void shiftCallback(HttpUriRequest outboundRequest,
HttpServletRequest inboundRequest) {
- /*
- Due to the async behavior shifting has to take place after a
successful response-received event
- and not in the executeRequest method. This is the same as in a
sync dispatch.
- */
- boolean userAgentDisabled = isUserAgentDisabled(inboundRequest);
- shiftActiveURL(userAgentDisabled, userAgentDisabled ? Optional.empty()
: getBackendFromHaCookie(outboundRequest, inboundRequest));
- }
}
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/HaProvider.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/HaProvider.java
index ff543f68a..228141be3 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/HaProvider.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/HaProvider.java
@@ -72,6 +72,17 @@ public interface HaProvider {
*/
void makeNextActiveURLAvailable(String serviceName);
+ /**
+ * Returns the current active URL for the service and advances the rotation
to the next
+ * URL, as a single atomic operation: concurrent callers are guaranteed to
observe
+ * different values as long as more than one URL is configured.
+ *
+ * @param serviceName the name of the service
+ * @return the URL that was active at the time of the call, or {@code null}
if no URLs
+ * are configured for the service
+ */
+ String getActiveURLAndAdvance(String serviceName);
+
/**
* This method puts gets all the currently
* available URLs for the service.
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/URLManager.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/URLManager.java
index 383cd8d26..53c84c21b 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/URLManager.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/URLManager.java
@@ -35,4 +35,6 @@ public interface URLManager {
void markFailed(String url);
void makeNextActiveURLAvailable();
+
+ String getActiveURLAndAdvance();
}
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/BaseZookeeperURLManager.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/BaseZookeeperURLManager.java
index d3e61abcc..590b09bd1 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/BaseZookeeperURLManager.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/BaseZookeeperURLManager.java
@@ -121,6 +121,19 @@ public abstract class BaseZookeeperURLManager implements
URLManager {
urls.offer(head);
}
+ @Override
+ public synchronized String getActiveURLAndAdvance() {
+ if (urls.isEmpty()) {
+ setURLs(lookupURLs());
+ }
+ String head = urls.poll();
+ if (head == null) {
+ return null;
+ }
+ urls.offer(head);
+ return head;
+ }
+
@Override
public synchronized void setURLs(List<String> urls) {
if ((urls != null) && (!(urls.isEmpty()))) {
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultHaProvider.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultHaProvider.java
index f3ddc929d..c4c7c204e 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultHaProvider.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultHaProvider.java
@@ -125,6 +125,28 @@ public class DefaultHaProvider implements HaProvider {
}
}
+ /*
+ * The pick-and-advance methods deliberately take the READ lock even though
they rotate
+ * the URL queue: their atomicity is provided by the URLManager's own
synchronization,
+ * and the provider-level rwl only guards cross-method consistency with
markFailedURL /
+ * setActiveURL. Taking the write lock here would serialize every
load-balanced request
+ * across all services behind a single gateway-wide lock — and hold it while
ZooKeeper-
+ * backed managers refresh their URL list (a blocking remote call).
+ */
+ @Override
+ public String getActiveURLAndAdvance(String serviceName) {
+ rwl.readLock().lock();
+ try {
+ if (haServices.containsKey(serviceName)) {
+ return haServices.get(serviceName).getActiveURLAndAdvance();
+ }
+ LOG.noActiveUrlFound(serviceName);
+ return null;
+ } finally {
+ rwl.readLock().unlock();
+ }
+ }
+
@Override
public List<String> getURLs(String serviceName) {
if ( haServices.containsKey(serviceName) ) {
diff --git
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultURLManager.java
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultURLManager.java
index 34b2cc8eb..78662fcb1 100644
---
a/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultURLManager.java
+++
b/gateway-provider-ha/src/main/java/org/apache/knox/gateway/ha/provider/impl/DefaultURLManager.java
@@ -102,4 +102,14 @@ public class DefaultURLManager implements URLManager {
String head = urls.poll();
urls.offer(head);
}
+
+ @Override
+ public synchronized String getActiveURLAndAdvance() {
+ String head = urls.poll();
+ if (head == null) {
+ return null;
+ }
+ urls.offer(head);
+ return head;
+ }
}
diff --git
a/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatchTest.java
b/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatchTest.java
index 92f1942dc..404d60111 100644
---
a/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatchTest.java
+++
b/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/ConfigurableHADispatchTest.java
@@ -24,6 +24,13 @@ import java.io.IOException;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.servlet.FilterConfig;
@@ -36,13 +43,19 @@ import org.apache.commons.codec.digest.DigestUtils;
import org.apache.http.Header;
import org.apache.http.HeaderElement;
import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.apache.http.HttpRequest;
import org.apache.http.HttpStatus;
import org.apache.http.StatusLine;
import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.conn.ClientConnectionManager;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.protocol.HttpContext;
import org.apache.knox.gateway.config.GatewayConfig;
import org.apache.knox.gateway.ha.provider.HaDescriptor;
import org.apache.knox.gateway.ha.provider.HaProvider;
@@ -51,6 +64,7 @@ import
org.apache.knox.gateway.ha.provider.impl.DefaultHaProvider;
import org.apache.knox.gateway.ha.provider.impl.HaDescriptorFactory;
import org.apache.knox.gateway.servlet.SynchronousServletOutputStreamAdapter;
import org.easymock.Capture;
+import org.easymock.CaptureType;
import org.easymock.EasyMock;
import org.easymock.IAnswer;
import org.junit.Assert;
@@ -284,4 +298,375 @@ public class ConfigurableHADispatchTest {
Assert.assertEquals(uri1.toString(), provider.getActiveURL(serviceName));
}
+ @Test
+ public void testConcurrentRequestsDoNotShareBackendDuringSlowCall() throws
Exception {
+ String serviceName = "HIVE";
+ HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
+ descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(
+ serviceName, "true", "1", "1000", null, null, "true", "true", null,
null, null));
+ HaProvider provider = new DefaultHaProvider(descriptor);
+ URI uri1 = new URI("http://host1.valid";);
+ URI uri2 = new URI("http://host2.valid";);
+ ArrayList<String> urlList = new ArrayList<>();
+ urlList.add(uri1.toString());
+ urlList.add(uri2.toString());
+ provider.addHaService(serviceName, urlList);
+ final HttpGet outboundRequest1 = new HttpGet(uri1);
+ final HttpGet outboundRequest2 = new HttpGet(uri1);
+
+ ServletContext context = EasyMock.createNiceMock(ServletContext.class);
+ GatewayConfig config = EasyMock.createNiceMock(GatewayConfig.class);
+
EasyMock.expect(context.getAttribute(GatewayConfig.GATEWAY_CONFIG_ATTRIBUTE)).andReturn(config).anyTimes();
+
+ HttpServletRequest inboundRequest1 =
EasyMock.createNiceMock(HttpServletRequest.class);
+ HttpServletRequest inboundRequest2 =
EasyMock.createNiceMock(HttpServletRequest.class);
+
EasyMock.expect(inboundRequest1.getServletContext()).andReturn(context).anyTimes();
+
EasyMock.expect(inboundRequest2.getServletContext()).andReturn(context).anyTimes();
+
EasyMock.expect(inboundRequest1.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).anyTimes();
+
EasyMock.expect(inboundRequest2.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).anyTimes();
+
+ CloseableHttpResponse inboundResponse =
EasyMock.createNiceMock(CloseableHttpResponse.class);
+ StatusLine statusLine = EasyMock.createNiceMock(StatusLine.class);
+ HttpEntity entity = EasyMock.createNiceMock(HttpEntity.class);
+ Header header = EasyMock.createNiceMock(Header.class);
+
EasyMock.expect(inboundResponse.getStatusLine()).andReturn(statusLine).anyTimes();
+
EasyMock.expect(statusLine.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes();
+ EasyMock.expect(inboundResponse.getEntity()).andReturn(entity).anyTimes();
+ EasyMock.expect(inboundResponse.getAllHeaders()).andReturn(new
Header[0]).anyTimes();
+ EasyMock.expect(entity.getContent()).andAnswer(() ->
+ new
ByteArrayInputStream("x".getBytes(StandardCharsets.UTF_8))).anyTimes();
+ EasyMock.expect(entity.getContentType()).andReturn(header).anyTimes();
+ EasyMock.expect(header.getElements()).andReturn(new
HeaderElement[]{}).anyTimes();
+ EasyMock.expect(entity.getContentLength()).andReturn(1L).anyTimes();
+
+ HttpServletResponse outboundResponse1 =
EasyMock.createNiceMock(HttpServletResponse.class);
+ HttpServletResponse outboundResponse2 =
EasyMock.createNiceMock(HttpServletResponse.class);
+ EasyMock.expect(outboundResponse1.getOutputStream()).andAnswer(
+ () -> new SynchronousServletOutputStreamAdapter() {
+ @Override public void write(int b) { /* do nothing */ }
+ }).anyTimes();
+ EasyMock.expect(outboundResponse2.getOutputStream()).andAnswer(
+ () -> new SynchronousServletOutputStreamAdapter() {
+ @Override public void write(int b) { /* do nothing */ }
+ }).anyTimes();
+
+ final CountDownLatch firstCallEntered = new CountDownLatch(1);
+ final CountDownLatch releaseFirstCall = new CountDownLatch(1);
+ CloseableHttpClient httpClient = new CloseableHttpClient() {
+ @Override
+ protected CloseableHttpResponse doExecute(HttpHost target, HttpRequest
request, HttpContext context) throws IOException {
+ if (request == outboundRequest1) {
+ firstCallEntered.countDown();
+ try {
+ if (!releaseFirstCall.await(5, TimeUnit.SECONDS)) {
+ throw new IOException("Slow call was not released in time");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ return inboundResponse;
+ }
+ @Override public void close() { /* no-op */ }
+ @Override public HttpParams getParams() { return new BasicHttpParams(); }
+ @Override public ClientConnectionManager getConnectionManager() { return
null; }
+ };
+
+ EasyMock.replay(inboundRequest1, inboundRequest2, context, config,
+ inboundResponse, statusLine, entity, header,
+ outboundResponse1, outboundResponse2);
+
+ Assert.assertEquals(uri1.toString(), provider.getActiveURL(serviceName));
+
+ ConfigurableHADispatch dispatch = new ConfigurableHADispatch();
+ dispatch.setHttpClient(httpClient);
+ dispatch.setHaProvider(provider);
+ dispatch.setServiceRole(serviceName);
+ dispatch.init();
+ ExecutorService pool = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> slowRequest = pool.submit(() -> {
+ try {
+ dispatch.executeRequestWrapper(outboundRequest1, inboundRequest1,
outboundResponse1);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ Assert.assertTrue("Slow request did not reach the HTTP call in time",
+ firstCallEntered.await(5, TimeUnit.SECONDS));
+
+ dispatch.executeRequestWrapper(outboundRequest2, inboundRequest2,
outboundResponse2);
+
+ Assert.assertNotEquals(
+ "Concurrent second request landed on the same backend as the slow
one — race not fixed",
+ outboundRequest1.getURI().getHost(),
+ outboundRequest2.getURI().getHost());
+ Assert.assertEquals(uri1.getHost(), outboundRequest1.getURI().getHost());
+ Assert.assertEquals(uri2.getHost(), outboundRequest2.getURI().getHost());
+
+ releaseFirstCall.countDown();
+ slowRequest.get(5, TimeUnit.SECONDS);
+ } finally {
+ releaseFirstCall.countDown();
+ pool.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testSetBackendUriCarriesPickedBackendBasePath() throws Exception
{
+ String serviceName = "HIVE";
+ HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
+ descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(
+ serviceName, "true", "1", "0", null, null, "true", "false", null,
null, null));
+ HaProvider provider = new DefaultHaProvider(descriptor);
+ ArrayList<String> urlList = new ArrayList<>();
+ urlList.add("http://host1.valid:8443/cliservice";);
+ urlList.add("http://host2.valid:8443/cliservice2";);
+ provider.addHaService(serviceName, urlList);
+
+ ConfigurableHADispatch dispatch = new ConfigurableHADispatch();
+ dispatch.setHaProvider(provider);
+ dispatch.setServiceRole(serviceName);
+ dispatch.init();
+
+ HttpGet outboundRequest = new HttpGet(new
URI("http://host1.valid:8443/cliservice/query?op=EXECUTE";));
+ provider.makeNextActiveURLAvailable(serviceName);
+
+ HttpServletRequest inboundRequest =
EasyMock.createNiceMock(HttpServletRequest.class);
+ EasyMock.replay(inboundRequest);
+
+ dispatch.setBackendUri(outboundRequest, inboundRequest, false);
+
+ Assert.assertEquals("Picked backend's base path must replace the peeked
backend's",
+ new URI("http://host2.valid:8443/cliservice2/query?op=EXECUTE";),
outboundRequest.getURI());
+ }
+
+ @Test
+ public void testSetBackendUriCarriesPickedBackendBasePath_EmptyFirst()
throws Exception {
+ String serviceName = "HIVE";
+ HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
+ descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(
+ serviceName, "true", "1", "0", null, null, "true", "false", null,
null, null));
+ HaProvider provider = new DefaultHaProvider(descriptor);
+ ArrayList<String> urlList = new ArrayList<>();
+ urlList.add("http://host1.valid:8443";);
+ urlList.add("http://host2.valid:8443/cliservice2";);
+ provider.addHaService(serviceName, urlList);
+
+ ConfigurableHADispatch dispatch = new ConfigurableHADispatch();
+ dispatch.setHaProvider(provider);
+ dispatch.setServiceRole(serviceName);
+ dispatch.init();
+
+ HttpGet outboundRequest = new HttpGet(new URI("http://host1.valid:8443";));
+ provider.makeNextActiveURLAvailable(serviceName);
+
+ HttpServletRequest inboundRequest =
EasyMock.createNiceMock(HttpServletRequest.class);
+ EasyMock.replay(inboundRequest);
+
+ dispatch.setBackendUri(outboundRequest, inboundRequest, false);
+
+ Assert.assertEquals("Picked backend's base path must replace the peeked
backend's",
+ new URI("http://host2.valid:8443/cliservice2";),
outboundRequest.getURI());
+ }
+
+ /*
+ * markEndpointFailed stores the FULL failed request URI (path and query
included) in the
+ * dispatch-level activeURL. The user-agent-disabled branch of setBackendUri
routes against
+ * activeURL via updateBackendURL, which interprets its argument's path as
the backend's
+ * base path — so a stale full request URI must not leak its path into later
requests.
+ */
+ @Test
+ public void testUserAgentDisabledRequestAfterFailoverKeepsRequestPath()
throws Exception {
+ String serviceName = "HIVE";
+ HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
+ descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(
+ serviceName, "true", "1", "0", null, null, "true", "true", null, null,
"agentX"));
+ HaProvider provider = new DefaultHaProvider(descriptor);
+ ArrayList<String> urlList = new ArrayList<>();
+ urlList.add("http://host1.valid";);
+ urlList.add("http://host2.valid";);
+ provider.addHaService(serviceName, urlList);
+
+ ConfigurableHADispatch dispatch = new ConfigurableHADispatch();
+ dispatch.setHaProvider(provider);
+ dispatch.setServiceRole(serviceName);
+ dispatch.init();
+
+ /* a failover on any request stores the full failed request URI in
activeURL */
+ HttpGet failedRequest = new HttpGet(new
URI("http://host1.valid/svc/path?op=EXECUTE";));
+ HttpServletRequest failoverInbound =
EasyMock.createNiceMock(HttpServletRequest.class);
+ EasyMock.replay(failoverInbound);
+ dispatch.markEndpointFailed(failedRequest, failoverInbound);
+ Assert.assertEquals("http://host1.valid/svc/path?op=EXECUTE";,
dispatch.getActiveURL().get());
+
+ /* a later request from a user agent configured to bypass load balancing */
+ HttpGet outboundRequest = new HttpGet(new
URI("http://host2.valid/other?op=LIST";));
+ HttpServletRequest inboundRequest =
EasyMock.createNiceMock(HttpServletRequest.class);
+
EasyMock.expect(inboundRequest.getHeader("User-Agent")).andReturn("agentX").anyTimes();
+ EasyMock.replay(inboundRequest);
+ Assert.assertTrue(dispatch.isUserAgentDisabled(inboundRequest));
+
+ dispatch.setBackendUri(outboundRequest, inboundRequest, true);
+
+ Assert.assertEquals("Stale activeURL's request path leaked into the
outbound URI as a base path",
+ "/other", outboundRequest.getURI().getPath());
+ Assert.assertEquals(new URI("http://host1.valid/other?op=LIST";),
outboundRequest.getURI());
+ }
+
+ /*
+ * With three backends and maxFailoverAttempts=2, a request whose first two
attempts both throw
+ * a connection error must walk the whole rotation host1 -> host2 -> host3
and succeed on the
+ * third, never revisiting a host it already tried. This exercises the real
selection path end
+ * to end: setBackendUri's pick-and-advance for the first attempt, and
markFailedURL rotating
+ * the dead backend to the tail so the re-run rewriter (simulated here via
getRequestURL, which
+ * mirrors the $serviceUrl function peeking provider.getActiveURL) targets
the next healthy one.
+ */
+ @Test
+ public void testRequestRotatesAcrossBackendsOnMultipleFailovers() throws
Exception {
+ final String serviceName = "HIVE";
+ HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
+ descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(
+ serviceName, "true", "2", "0", null, null, "true", "true", null, null,
null));
+ final HaProvider provider = new DefaultHaProvider(descriptor);
+ URI uri1 = new URI("http://host1.valid";);
+ URI uri2 = new URI("http://host2.valid";);
+ URI uri3 = new URI("http://host3.valid";);
+ ArrayList<String> urlList = new ArrayList<>();
+ urlList.add(uri1.toString());
+ urlList.add(uri2.toString());
+ urlList.add(uri3.toString());
+ provider.addHaService(serviceName, urlList);
+
+ /* a real request object so setBackendUri / prepareForFailover actually
mutate the URI */
+ final HttpGet outboundRequest = new HttpGet(uri1);
+
+ /*
+ * Stand in for the URL-rewrite filter: on each (re)dispatch the
rewriter's $serviceUrl
+ * function re-peeks the provider's current active URL. prepareForFailover
nulls the cached
+ * target so getDispatchUrl -> getRequestURL is what re-targets the
failed-over request.
+ */
+ HttpServletRequest inboundRequest =
EasyMock.createNiceMock(HttpServletRequest.class);
+ EasyMock.expect(inboundRequest.getRequestURL())
+ .andAnswer(() -> new
StringBuffer(provider.getActiveURL(serviceName))).anyTimes();
+
EasyMock.expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
+
EasyMock.expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(1)).once();
+
+ /* backend response for the successful third attempt */
+ CloseableHttpResponse inboundResponse =
EasyMock.createNiceMock(CloseableHttpResponse.class);
+ StatusLine statusLine = EasyMock.createNiceMock(StatusLine.class);
+ HttpEntity entity = EasyMock.createNiceMock(HttpEntity.class);
+ Header header = EasyMock.createNiceMock(Header.class);
+ ServletContext context = EasyMock.createNiceMock(ServletContext.class);
+ GatewayConfig config = EasyMock.createNiceMock(GatewayConfig.class);
+
+
EasyMock.expect(inboundResponse.getStatusLine()).andReturn(statusLine).anyTimes();
+
EasyMock.expect(statusLine.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes();
+ EasyMock.expect(inboundResponse.getEntity()).andReturn(entity).anyTimes();
+ EasyMock.expect(inboundResponse.getAllHeaders()).andReturn(new
Header[0]).anyTimes();
+
EasyMock.expect(inboundRequest.getServletContext()).andReturn(context).anyTimes();
+ EasyMock.expect(entity.getContent())
+ .andAnswer(() -> new
ByteArrayInputStream("knox-backend".getBytes(StandardCharsets.UTF_8))).anyTimes();
+ EasyMock.expect(entity.getContentType()).andReturn(header).anyTimes();
+ EasyMock.expect(header.getElements()).andReturn(new
HeaderElement[]{}).anyTimes();
+ EasyMock.expect(entity.getContentLength()).andReturn(4L).anyTimes();
+
EasyMock.expect(context.getAttribute(GatewayConfig.GATEWAY_CONFIG_ATTRIBUTE)).andReturn(config).anyTimes();
+
+ HttpServletResponse outboundResponse =
EasyMock.createNiceMock(HttpServletResponse.class);
+ Capture<Integer> statusCodeCapture =
EasyMock.newCapture(CaptureType.FIRST);
+ outboundResponse.setStatus(EasyMock.captureInt(statusCodeCapture));
+ EasyMock.expectLastCall().anyTimes();
+ EasyMock.expect(outboundResponse.getOutputStream())
+ .andAnswer((IAnswer<SynchronousServletOutputStreamAdapter>) () -> new
SynchronousServletOutputStreamAdapter() {
+ @Override
+ public void write(int b) { /* do nothing */ }
+ }).anyTimes();
+
+ /* record the backend each attempt actually targeted; fail the first two,
succeed on the third */
+ final List<String> attemptedHosts = new ArrayList<>();
+ CloseableHttpClient httpClient = new CloseableHttpClient() {
+ @Override
+ protected CloseableHttpResponse doExecute(HttpHost target, HttpRequest
request, HttpContext ctx) throws IOException {
+ attemptedHosts.add(outboundRequest.getURI().getHost());
+ if (attemptedHosts.size() <= 2) {
+ throw new IOException("unreachable-host");
+ }
+ return inboundResponse;
+ }
+ @Override public void close() { /* no-op */ }
+ @Override public HttpParams getParams() { return new BasicHttpParams(); }
+ @Override public ClientConnectionManager getConnectionManager() { return
null; }
+ };
+
+ EasyMock.replay(inboundRequest, outboundResponse, inboundResponse,
+ statusLine, entity, header, context, config);
+
+ Assert.assertEquals(uri1.toString(), provider.getActiveURL(serviceName));
+ ConfigurableHADispatch dispatch = new ConfigurableHADispatch();
+ dispatch.setHttpClient(httpClient);
+ dispatch.setHaProvider(provider);
+ dispatch.setServiceRole(serviceName);
+ dispatch.init();
+
+ dispatch.executeRequestWrapper(outboundRequest, inboundRequest,
outboundResponse);
+
+ /* both counter reads were consumed => exactly two failovers happened */
+ EasyMock.verify(inboundRequest);
+ Assert.assertEquals("Request must walk the full rotation without
revisiting a backend.",
+ Arrays.asList(uri1.getHost(), uri2.getHost(), uri3.getHost()),
attemptedHosts);
+ Assert.assertEquals("The finally-successful backend must be the outbound
target.",
+ uri3.getHost(), outboundRequest.getURI().getHost());
+ Assert.assertEquals("Each attempt advances the rotation, so after serving
on the last backend the head is the next one.",
+ uri1.toString(), provider.getActiveURL(serviceName));
+ Assert.assertEquals("Expected the request to succeed after failing over
twice.",
+ HttpStatus.SC_OK, statusCodeCapture.getValue().intValue());
+ }
+
+ /*
+ * Failover-path counterpart to the initial-selection reconciliation: if the
URL queue is
+ * rotated (by a concurrent request) between the rewrite's getActiveURL peek
and our
+ * getActiveURLAndAdvance in prepareForFailover, updateBackendURL must
re-target the outbound
+ * request at the backend we actually picked, carrying THAT backend's base
path — not the
+ * stale peeked one. Here the rewrite peeks host1 (/cliservice) but the head
has already
+ * advanced to host2 (/cliservice2) by the time we pick.
+ */
+ @Test
+ public void
testPrepareForFailoverReTargetsWhenQueueRotatedBetweenPeekAndAdvance() throws
Exception {
+ String serviceName = "HIVE";
+ HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
+ descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(
+ serviceName, "true", "1", "0", null, null, "true", "false", null,
null, null));
+ HaProvider provider = new DefaultHaProvider(descriptor);
+ ArrayList<String> urlList = new ArrayList<>();
+ urlList.add("http://host1.valid:8443/cliservice";);
+ urlList.add("http://host2.valid:8443/cliservice2";);
+ provider.addHaService(serviceName, urlList);
+
+ ConfigurableHADispatch dispatch = new ConfigurableHADispatch();
+ dispatch.setHaProvider(provider);
+ dispatch.setServiceRole(serviceName);
+ dispatch.init();
+
+ /* the outbound URI as the rewrite produced it after peeking host1 (host1
base path + tail) */
+ HttpGet outboundRequest = new HttpGet(new
URI("http://host1.valid:8443/cliservice/query?op=EXECUTE";));
+ HttpServletRequest inboundRequest =
EasyMock.createNiceMock(HttpServletRequest.class);
+ /* getDispatchUrl re-derives the peeked target from these on the retry */
+ EasyMock.expect(inboundRequest.getRequestURL())
+ .andReturn(new
StringBuffer("http://host1.valid:8443/cliservice/query";)).anyTimes();
+
EasyMock.expect(inboundRequest.getQueryString()).andReturn("op=EXECUTE").anyTimes();
+ EasyMock.replay(inboundRequest);
+
+ /* a concurrent request advances the head host1 -> host2 between the peek
and our advance */
+ provider.makeNextActiveURLAvailable(serviceName);
+ Assert.assertEquals("http://host2.valid:8443/cliservice2";,
provider.getActiveURL(serviceName));
+
+ dispatch.prepareForFailover(outboundRequest, inboundRequest);
+
+ Assert.assertEquals("Retried request must target the advanced pick with
its own base path, not the peeked backend's",
+ new URI("http://host2.valid:8443/cliservice2/query?op=EXECUTE";),
outboundRequest.getURI());
+ /* selection advanced the queue past the picked backend */
+ Assert.assertEquals("http://host1.valid:8443/cliservice";,
provider.getActiveURL(serviceName));
+ }
}
diff --git
a/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatchTest.java
b/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatchTest.java
index c0ca0f2e9..97028a076 100644
---
a/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatchTest.java
+++
b/gateway-provider-ha/src/test/java/org/apache/knox/gateway/ha/dispatch/SSEHaDispatchTest.java
@@ -86,7 +86,6 @@ public class SSEHaDispatchTest {
@Test
public void testHADispatchURL() throws Exception {
- String serviceName = "SSE";
HaDescriptor descriptor = HaDescriptorFactory.createDescriptor();
descriptor.addServiceConfig(HaDescriptorFactory.createServiceConfig(serviceName,
"true", "1", "1000", null, null, "true", "true", null, null, null, null));
HaProvider provider = new DefaultHaProvider(descriptor);
@@ -344,7 +343,7 @@ public class SSEHaDispatchTest {
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
Capture<Cookie> capturedArgument = Capture.newInstance();
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse,
capturedArgument);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -358,7 +357,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -376,7 +375,7 @@ public class SSEHaDispatchTest {
AsyncContext asyncContext = this.getAsyncContext(latch,
outboundResponse);
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse, null);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -390,7 +389,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -408,7 +407,7 @@ public class SSEHaDispatchTest {
AsyncContext asyncContext = this.getAsyncContext(latch,
outboundResponse);
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse, null);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -422,7 +421,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -454,7 +453,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest);
@@ -473,7 +472,7 @@ public class SSEHaDispatchTest {
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
Capture<Cookie> capturedArgument = Capture.newInstance();
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse,
capturedArgument);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -487,7 +486,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -506,7 +505,7 @@ public class SSEHaDispatchTest {
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
Capture<Cookie> capturedArgument = Capture.newInstance();
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse,
capturedArgument);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -520,7 +519,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doPost(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doPost(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -539,7 +538,7 @@ public class SSEHaDispatchTest {
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
Capture<Cookie> capturedArgument = Capture.newInstance();
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse,
capturedArgument);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -553,7 +552,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doPost(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doPost(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -564,7 +563,7 @@ public class SSEHaDispatchTest {
@Test
public void testFailoverDisabledWithStickySession() throws Exception {
CountDownLatch latch = new CountDownLatch(1);
- HaServiceConfig haServiceConfig =
HaDescriptorFactory.createServiceConfig(serviceName, "true", "1", "1000", null,
null, "true", "true", null, null, "agentX,user1,agentY", null);
+ HaServiceConfig haServiceConfig =
HaDescriptorFactory.createServiceConfig(serviceName, "true", "1", "1000", null,
null, "true", "true", null, "true", "agentX,user1,agentY", null);
SSEHaDispatch sseHaDispatch = this.createDispatch(true,
haServiceConfig);
PrintWriter printWriter = EasyMock.createNiceMock(PrintWriter.class);
HttpServletResponse outboundResponse =
this.getServletResponse(HttpStatus.SC_OK);
@@ -572,7 +571,7 @@ public class SSEHaDispatchTest {
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
Capture<Cookie> capturedArgument = Capture.newInstance();
expect(inboundRequest.getHeader("User-Agent")).andReturn("unknown").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse,
capturedArgument);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -586,7 +585,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
@@ -625,7 +624,7 @@ public class SSEHaDispatchTest {
HttpServletRequest inboundRequest =
this.getHttpServletRequest(asyncContext);
Capture<Cookie> capturedArgument = Capture.newInstance();
expect(inboundRequest.getHeader("User-Agent")).andReturn("agentX").anyTimes();
- expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).once();
+ expect(inboundRequest.getRequestURL()).andReturn(new
StringBuffer(URL.toString())).anyTimes();
expect(inboundRequest.getAttribute("dispatch.ha.failover.counter")).andReturn(new
AtomicInteger(0)).once();
this.expectResponseBodyAndHeader(printWriter, outboundResponse,
capturedArgument);
replay(inboundRequest, asyncContext, outboundResponse, printWriter);
@@ -639,7 +638,7 @@ public class SSEHaDispatchTest {
.content("id:1\ndata:data1\nevent:event1\n\ndata:data2\nevent:event2\nid:2\nretry:1\n:testing\n\n",
StandardCharsets.UTF_8)
.header("response", "header")
.contentType("text/event-stream");
- sseHaDispatch.doGet(new URI("http://unknown-host.invalid";),
inboundRequest, outboundResponse);
+ sseHaDispatch.doGet(new URI("http://unknown-host.invalid/sse";),
inboundRequest, outboundResponse);
latch.await(2L, TimeUnit.SECONDS);
EasyMock.verify(asyncContext, outboundResponse, inboundRequest,
printWriter);
diff --git
a/gateway-spi/src/main/java/org/apache/knox/gateway/sse/SSEDispatch.java
b/gateway-spi/src/main/java/org/apache/knox/gateway/sse/SSEDispatch.java
index 247536d71..5e8ec8e56 100644
--- a/gateway-spi/src/main/java/org/apache/knox/gateway/sse/SSEDispatch.java
+++ b/gateway-spi/src/main/java/org/apache/knox/gateway/sse/SSEDispatch.java
@@ -163,10 +163,6 @@ public class SSEDispatch extends ConfigurableDispatch
implements AsyncDispatch {
return (statusCode >= HttpStatus.SC_OK && statusCode < 300);
}
- protected void shiftCallback(HttpUriRequest outboundRequest,
HttpServletRequest inboundRequest) {
- // No need to shift the URL for non-HA SSE requests
- }
-
protected class SSECharConsumer extends AsyncCharConsumer<SSEResponse> {
private SSEResponse sseResponse;
private final HttpServletResponse outboundResponse;
@@ -192,7 +188,6 @@ public class SSEDispatch extends ConfigurableDispatch
implements AsyncDispatch {
} else {
handleErrorResponse(outboundResponse, url, inboundResponse);
}
- shiftCallback(outboundRequest, inboundRequest);
}
@Override
