Search code examples
spring-bootjava.util.concurrentspring-cloud-sleuthdistributed-tracing

sleuth trace id not getting printed for concurrent calls


In the log, for concurrent calls trace and span id are not getting printed. But for other calls, I can see it's getting printed.

Below is my code.

@RestController
public class GetStudent implements IGetStudent {

    private static final Logger log = LoggerFactory.getLogger(GetStudent.class);
    @Autowired
    private IGetStudentService getStudentService;

    @Override
    @RequestMapping(value = "/getStudent/{studentID}", method = RequestMethod.GET, consumes = "application/json")
    public List<Student> getStudentByID(@PathVariable String studentID) {
        List<Student> studentList = new ArrayList<>();
        log.info("In Controller");
        List<CompletableFuture<Student>> futuresList = new ArrayList<CompletableFuture<Student>>();
        CompletableFuture<Student> addAsy = CompletableFuture
                .supplyAsync(() -> (getStudentService.getStudentByID(studentID)));
        CompletableFuture<Student> addAsy1 = CompletableFuture
                .supplyAsync(() -> (getStudentService.getStudentByID(studentID)));
        futuresList.add(addAsy);
        futuresList.add(addAsy1);
        CompletableFuture<Void> allFutures = CompletableFuture
                .allOf(futuresList.toArray(new CompletableFuture[futuresList.size()]));

        CompletableFuture<List<Student>> allCompletableFuture = allFutures.thenApply(future -> {
            return futuresList.stream().map(completableFuture -> completableFuture.join()).collect(Collectors.toList());
        });
        CompletableFuture<List<Student>> completableFuture = allCompletableFuture.toCompletableFuture();
        try {
            List<Student> finalList = (List<Student>) completableFuture.get();
            for (Student s : finalList) {
                log.info("Student result {}", s);
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        log.info("Calling servicelayer without future");
        studentList.add(getStudentService.getStudentByID(studentID));
        return studentList;
    }
}

Service Layer::

@Service
public class GetStudentService implements IGetStudentService{

    private static final Logger Logger = LoggerFactory.getLogger(GetStudentService.class);
    
    @Override
    public Student getStudentByID(String id) {
        Logger.info("**In Servicelayer*** {}", Thread.currentThread().getName());
        Student student = new Student();
        student.setName("xyx");
        student.setAge("39");   
        return student;
    }
}

Logs:: enter image description here


Solution

  • Your supplyAsync method call is wrong, you need to pass a traced executor service like this

    https://github.com/spring-cloud-samples/brewery/blob/4ccee91ac7a4b7ac7a5b84002c5f3b25e72ffd2d/brewing/src/main/java/io/spring/cloud/samples/brewery/aggregating/IngredientsAggregator.java#L56

    Example:

        private final BeanFactory beanFactory;
    
        @Autowired
        IngredientsAggregator(BeanFactory beanFactory) {
            this.beanFactory = beanFactory;
        }
    
       // TODO: Consider simplifying the case by removing the DB (always matches threshold)
        public Ingredients fetchIngredients(Order order, String processId) throws Exception {
            log.info("Fetching ingredients for order [{}] , processId [{}]", order, processId);
            /**
             * [SLEUTH] ParallelStreams won't work out of the box
             * - example of a completable future with our TraceableExecutorService
             * - makes little business sense here but that's just an example
             */
            CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
                        ingredientsCollector.collectIngredients(order, processId).stream()
                                .filter(ingredient -> ingredient != null)
                                .forEach((Ingredient ingredient) -> {
                                    log.info("Adding an ingredient [{}] for order [{}] , processId [{}]", ingredient);
                                    ingredientWarehouse.addIngredient(ingredient);
                                });
                        return null;
                    }, new TraceableExecutorService(this.beanFactory, Executors.newFixedThreadPool(5), "fetchIngredients"));
            // block to perform the request (as I said the example is stupid)
            completableFuture.get();
            eventGateway.emitEvent(Event.builder().eventType(EventType.INGREDIENTS_ORDERED).processId(processId).build());
            Ingredients ingredients = ingredientWarehouse.getCurrentState();
            return maturingUpdater.updateIfLimitReached(ingredients, processId);
        }