Backfill test about reactor edge case

[codefromthecrypt]

From #1562 (comment)

@simonbasle @robotmrv I noticed in #1126 we started handling both onNext and onComplete, to finish a span. The code that handles onComplete when onNext isn't called is dead code. At least in our tests it isn't invoked.

I would like to remove the onComplete without onNext edge case as it isn't tested and also it is very confusing. Is there any reason why we should be so defensive?

When you look at the Flux or Mono level, not the Subscriber level, pipelines which end results appear empty are not that uncommon. Any reactive method that returns a Mono<Void> for instance...

That said, Sleuth works by wrapping Subscribers, which are the steps in one activation of the pipeline. The cases where such a step is empty are probably fewer. For instance, the Subscriber generated by ignoreElements(), despite producing a Flux that appears empty to the end consumer, still sees onNext() calls from its source. It just doesn't propagate them. If there is another operator after that, its Subscriber will only see an onComplete().

Another way of triggering this code path is by having an empty source:

Flux.empty()
    .map(i -> i)
    .filter(i -> true)
    .flatMap(i -> Mono.error(new RuntimeException("won't be thrown because no onNext"));

Hope this helps improving the tests/understanding when onComplete might matter...

It is probably true that the vast majority of Subscribers will not onComplete() without an onNext(), but when one considers the whole sequence (ie. the Flux or Mono) this cannot be considered an edge case. All Mon

[codefromthecrypt]

We also need a test case for onNext code that might issue a cancel after its first element. #1572 (review)

[codefromthecrypt]

another note about empty source #1572 (comment)

[codefromthecrypt]

TODO: also make a test that shows if possible spans are not started until the mono representing the call is subscribed to. There was a bug around this recently.

[olssonoskar]

Scope decoration seems to be order dependent if using Project reactors recommended approach to blocking calls. It seems that everything below .subscribeOn(Schedulers.boundedElastic()) in the chain will have MDC decorated while for everything above it will be empty.

Mono blockingWrapper = Mono.fromCallable(() -> { 
    return /* blocking code */ 
})

// Not decorated, seems to be no in-flight span (calling Tracer.nextSpan in blocking code resulted in new root)
return blockingWrapper.subscribeOn(Schedulers.boundedElastic()); 

// MDC seems to be decorated during fromCallable scope with expected fields
return Mono.just("").subscribeOn(Schedulers.boundedElastic()).then(blockingWrapper)

// MDC is again empty
return Mono.just("").then(blockingWrapper).subscribeOn(Schedulers.boundedElastic())

Since MDC eventually is decorated, I believe the context is passed correctly, it is just a question of when decoration is and when it should be triggered.

[codefromthecrypt]

related question about how to continue a trace in reactor. In this case, they are trying to do this..

Single<MyObject> methohThaCallAnotherRemoteService() {

        return Single.fromCallable( () -> { ...

            final ResponseEntity<MyObject> response = restTemplate.postForEntity...

            return response.getBody();

        })
        .subscribeOn(Schedulers.io()); // to be run on separate threads / 

https://stackoverflow.com/questions/61659898/make-spring-sleuth-not-to-create-a-new-trace-id-span-for-new-thread
#1650

[codefromthecrypt]

@simonbasle @robotmrv can you review the various problems and give some advice maybe? we're getting killed in support about various continuity problems in reactive commands and since there are so many ways to do things, it might be nice to have a suggestion we can put into docs

[simonbasle]

@adriancole that stackoverflow question is using RxJava Single, I'm not sure how that interacts with Sleuth instrumentation? Spring will use adapters to bridge the Single into a Mono, so that might introduce further complexity...

[marcingrzejszczak]

With the spring.sleuth.reacto.instrumentation-type: manual instrumentation mode (version 3.0.0) that shouldn't be a problem anymore cause it's up to the user to retrieve the TraceContext and manually pass it to all the required operators. Example (#1585)

spring.sleuth.baggage.remote-fields=test-key
spring.sleuth.reactor.instrumentation-type=manual

code

package com.example.demo;

import brave.baggage.BaggageField;
import brave.propagation.TraceContext;
import reactor.core.publisher.Mono;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.sleuth.instrument.web.WebFluxSleuthOperators;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ServerWebExchange;

import static reactor.core.scheduler.Schedulers.boundedElastic;

@RestController
@SpringBootApplication
public class DemoApplication2 {

	public static void main(String... args) {
		SpringApplication.run(DemoApplication2.class, args);
	}

	@GetMapping("/test")
	Mono<ResponseEntity<String>> test(ServerWebExchange exchange){
		return Mono.fromSupplier(() -> getExtraField(exchange))
			.subscribeOn(boundedElastic())
				.map(ResponseEntity::ok);
	}

	private String getExtraField(ServerWebExchange exchange) {
		TraceContext traceContext = WebFluxSleuthOperators.currentTraceContext(exchange);
		String value = BaggageField.getByName(traceContext, "test-key").getValue(traceContext);
		return value != null ? value : "no value for key";
	}
}

[spring-projects-issues]

If you would like us to look at this issue, please provide the requested information. If the information is not provided within the next 7 days this issue will be closed.

[spring-projects-issues]

Closing due to lack of requested feedback. If you would like us to look at this issue, please provide the requested information and we will re-open the issue.