Fixes potential NPE in getServices and getAllInstances

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/ConsulDiscoveryClient.java

@@ -79,28 +79,34 @@ private void addInstancesToList(List<ServiceInstance> instances, String serviceI
 		ResponseEntity<List<HealthService>> healthServices = client.getHealthServices(serviceId,
 				properties.isQueryPassing(), properties.getAclToken(), tags, queryParams);
 
-		for (HealthService service : healthServices.getBody()) {
+		List<HealthService> body = healthServices.getBody();
+		if (body == null) {
+			throw new IllegalStateException("Consul did not return instances for service " + serviceId
+					+ " (HTTP status " + healthServices.getStatusCode() + ")");
+		}
+
+		for (HealthService service : body) {
 			instances.add(new ConsulServiceInstance(service, serviceId));
 		}
 	}
 
 	public List<ServiceInstance> getAllInstances() {
 		List<ServiceInstance> instances = new ArrayList<>();
-		Map<String, List<String>> catalogServices = client
-			.getCatalogServices(properties.getAclToken(), QueryParams.DEFAULT)
-			.getBody();
-
-		for (String serviceId : catalogServices.keySet()) {
+		for (String serviceId : getServices()) {
 			addInstancesToList(instances, serviceId, null);
 		}
 		return instances;
 	}
 
 	@Override
 	public List<String> getServices() {
-		Map<String, List<String>> catalogServices = client
-			.getCatalogServices(properties.getAclToken(), QueryParams.DEFAULT)
-			.getBody();
+		ResponseEntity<Map<String, List<String>>> response = client.getCatalogServices(properties.getAclToken(),
+				QueryParams.DEFAULT);
+		Map<String, List<String>> catalogServices = response.getBody();
+		if (catalogServices == null) {
+			throw new IllegalStateException(
+					"Consul did not return a service catalog (HTTP status " + response.getStatusCode() + ")");
+		}
 		return new ArrayList<>(catalogServices.keySet());
 	}
 

spring-cloud-consul-discovery/src/main/java/org/springframework/cloud/consul/discovery/reactive/ConsulReactiveDiscoveryClient.java

@@ -62,21 +62,19 @@ public Flux<ServiceInstance> getInstances(String serviceId) {
 		return Flux.defer(() -> {
 			List<ServiceInstance> instances = blockingClient.getInstances(serviceId);
 			return Flux.fromIterable(instances);
-		}).onErrorResume(exception -> {
-			logger.error("Error getting instances from Consul.", exception);
-			return Flux.empty();
-		}).subscribeOn(Schedulers.boundedElastic());
+		})
+			.doOnError(exception -> logger.error("Error getting instances from Consul.", exception))
+			.subscribeOn(Schedulers.boundedElastic());
 	}
 
 	@Override
 	public Flux<String> getServices() {
 		return Flux.defer(() -> {
 			List<String> services = blockingClient.getServices();
 			return services == null ? Flux.empty() : Flux.fromIterable(services);
-		}).onErrorResume(exception -> {
-			logger.error("Error getting services from Consul.", exception);
-			return Flux.empty();
-		}).subscribeOn(Schedulers.boundedElastic());
+		})
+			.doOnError(exception -> logger.error("Error getting services from Consul.", exception))
+			.subscribeOn(Schedulers.boundedElastic());
 	}
 
 	@Override

spring-cloud-consul-discovery/src/test/java/org/springframework/cloud/consul/discovery/reactive/ConsulReactiveDiscoveryClientTests.java

@@ -71,14 +71,22 @@ public void verifyDefaults() {
 	}
 
 	@Test
-	public void shouldReturnEmptyFluxOfServicesWhenConsulFails() {
+	public void shouldPropagateErrorFromServicesWhenConsulFails() {
 		Flux<String> services = client.getServices();
 		when(consulClient.getCatalogServices(any(), any()))
 			.thenThrow(new RuntimeException("Possible runtime exception"));
-		StepVerifier.create(services).expectNextCount(0).expectComplete().verify();
+		StepVerifier.create(services).expectErrorMessage("Possible runtime exception").verify();
 		verify(consulClient).getCatalogServices(any(), any());
 	}
 
+	@Test
+	public void shouldPropagateErrorFromServicesWhenConsulReturnsNoBody() {
+		Flux<String> services = client.getServices();
+		when(consulClient.getCatalogServices(any(), any()))
+			.thenReturn(ResponseEntity.status(500).<Map<String, List<String>>>build());
+		StepVerifier.create(services).expectError(IllegalStateException.class).verify();
+	}
+
 	@Test
 	public void shouldReturnFluxOfServices() {
 		Flux<String> services = client.getServices();
@@ -121,17 +129,27 @@ public void shouldReturnEmptyFluxForNonExistingService() {
 	}
 
 	@Test
-	public void shouldReturnEmptyFluxWhenConsulFails() {
+	public void shouldPropagateErrorFromInstancesWhenConsulFails() {
 		configureCommonProperties();
 		when(consulClient.getHealthServices(eq("existing-service"), any(boolean.class), any(), any(List.class),
 				any(QueryParams.class)))
 			.thenThrow(new RuntimeException("Possible runtime exception"));
 		Flux<ServiceInstance> instances = client.getInstances("existing-service");
-		StepVerifier.create(instances).expectNextCount(0).expectComplete().verify();
+		StepVerifier.create(instances).expectErrorMessage("Possible runtime exception").verify();
 		verify(consulClient).getHealthServices(eq("existing-service"), any(boolean.class), any(), any(List.class),
 				any(QueryParams.class));
 	}
 
+	@Test
+	public void shouldPropagateErrorFromInstancesWhenConsulReturnsNoBody() {
+		configureCommonProperties();
+		when(consulClient.getHealthServices(eq("existing-service"), any(boolean.class), any(), any(List.class),
+				any(QueryParams.class)))
+			.thenReturn(ResponseEntity.status(500).<List<HealthService>>build());
+		Flux<ServiceInstance> instances = client.getInstances("existing-service");
+		StepVerifier.create(instances).expectError(IllegalStateException.class).verify();
+	}
+
 	@Test
 	public void shouldReturnFluxOfServiceInstances() {
 		configureCommonProperties();