From a67a0779067f2651f8ea162a6ac08d63907c407c Mon Sep 17 00:00:00 2001
From: Stephane Maldini <stephane.maldini@gmail.com>
Date: Tue, 24 Oct 2017 11:01:44 -0700
Subject: [PATCH] fix overflow issue with hystrix timeout operator (#2391)

- The fix adds a necessary buffer between the rx-rs adapter and
the timeout operator introduced in AbstractCommand. Unfortunately,
the hystrix operator does not comply with RS backpressure requirements.
It will always request an unbounded demand to a source regardless of
the command consumer demand. With this commit we protect RS adapter and
downstream flow from overflow. Ideally a rework of the operator should
be planned.
---
 spring-cloud-netflix-core/pom.xml             |  5 ++++
 .../netflix/hystrix/HystrixCommands.java      |  2 +-
 .../netflix/hystrix/HystrixCommandsTests.java | 23 +++++++++++++++----
 3 files changed, 24 insertions(+), 6 deletions(-)

diff --git a/spring-cloud-netflix-core/pom.xml b/spring-cloud-netflix-core/pom.xml
index e269c330..9142c4a7 100644
--- a/spring-cloud-netflix-core/pom.xml
+++ b/spring-cloud-netflix-core/pom.xml
@@ -44,6 +44,11 @@
 			<artifactId>reactor-core</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>io.reactivex</groupId>
 			<artifactId>rxjava-reactive-streams</artifactId>
diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java
index 1aceeeef..c6e1611b 100644
--- a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java
+++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/hystrix/HystrixCommands.java
@@ -104,7 +104,7 @@ public class HystrixCommands {
 			if (this.eager) {
 				observable = command.observe();
 			} else {
-				observable = command.toObservable();
+				observable = command.toObservable().onBackpressureBuffer();
 			}
 			return RxReactiveStreams.toPublisher(observable);
 		}
diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java
index 17063e99..06c09a13 100644
--- a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java
+++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/hystrix/HystrixCommandsTests.java
@@ -16,18 +16,19 @@
 
 package org.springframework.cloud.netflix.hystrix;
 
+import java.time.Duration;
 import java.util.List;
 
+import com.netflix.hystrix.exception.HystrixRuntimeException;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
-
-import com.netflix.hystrix.exception.HystrixRuntimeException;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
 
 public class HystrixCommandsTests {
 
@@ -75,6 +76,18 @@ public class HystrixCommandsTests {
 		assertThat(list).hasSize(2).contains("1", "2");
 	}
 
+	@Test
+	public void fluxWorksDeferredRequest() {
+		StepVerifier.create(HystrixCommands.from(Flux.just("1", "2"))
+		                                   .commandName("multiflux")
+		                                   .build(), 1)
+		            .expectNext("1")
+		            .thenAwait(Duration.ofSeconds(1))
+		            .thenRequest(1)
+		            .expectNext("2")
+		            .verifyComplete();
+	}
+
 	@Test
 	public void eagerFluxWorks() {
 		List<String> list = HystrixCommands.from( Flux.just("1", "2"))