Advice precedence problem when one @Around advice does not proceed

You can add the @Order annotation on the two aspects to define the ordering. If it's not provided, the execution order of aspects is undefined. For example, if the Log aspect should always be executed as the last advice, no matter what other advices there are, give it e.g. an order of 10000, and all other advices lower numbers (lower numbers are executed first).

Documentation: https://docs.spring.io/spring-framework/docs/current/reference/html/core.html#aop-ataspectj-advice-ordering

Because of this problem, you need to use @Order(value = Ordered.HIGHEST_PRECEDENCE) on PollableStreamListenerAspect. Quite weird indeed, but then it works as you wish it to. IMO the issue ought to be fixed in Spring, though. Having to use this workaround is ugly and only works if your aspect calling proceed() asynchronously actually does have the highest precedence, which is not always the case. As an alternative, you could use native AspectJ and its own concept of declaring advice precedence, which is independent of Spring internals.

Here is a simplified version of your application as an MCVE:

Annotations:

package de.scrum_master.spring.q67155048;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface CustomLogging {}

package de.scrum_master.spring.q67155048;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface PollableStreamListener {}

Component with method carrying both annotations:

package de.scrum_master.spring.q67155048;

import org.springframework.stereotype.Component;

@Component
public class MyComponent {
  private int counter = 0;

  @PollableStreamListener
  @CustomLogging
  public void submitDeletion() {
    try {
      System.out.println("  Starting processing #" + ++counter);
      Thread.sleep(1000);
      System.out.println("  Finished processing #" + counter);
    }
    catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Aspects:

package de.scrum_master.spring.q67155048;

import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.springframework.stereotype.Component;

@Aspect
@Component
public class CustomLoggingAspect {
  @Before("@annotation(de.scrum_master.spring.q67155048.CustomLogging)")
  public void addCustomLogging() {
    System.out.println("Logging");
  }
}

package de.scrum_master.spring.q67155048;

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Aspect
@Component
@Order(value = Ordered.HIGHEST_PRECEDENCE)
public class PollableStreamListenerAspect {
  public static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(1);
  private volatile boolean paused = false;

  @Around("@annotation(de.scrum_master.spring.q67155048.PollableStreamListener)")
  public void receiveMessage(ProceedingJoinPoint joinPoint) throws Throwable {
    System.out.println("Receiving message");
    if (!paused) {
      // The separate thread is not busy with a previous message, so process this message:
      Runnable runnable = () -> {
        try {
          paused = true;
          joinPoint.proceed();
        }
        catch (Throwable throwable) {
          throwable.printStackTrace();
        }
        finally {
          paused = false;
        }
      };
      EXECUTOR_SERVICE.submit(runnable);
    }
    else {
      System.out.println("  Re-queue");
    }
  }
}

Driver application:

The application calls the target method every 500 ms, but execution takes 1,000 ms. So we expect to see ~50% of the calls being re-queued without any logging in that case, because the higher precedence aspect does not proceed to the target method.

package de.scrum_master.spring.q67155048;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;

@SpringBootApplication
@Configuration
public class DemoApplication {
  public static void main(String[] args) throws InterruptedException {
    try (ConfigurableApplicationContext appContext = SpringApplication.run(DemoApplication.class, args)) {
      doStuff(appContext);
    }
  }

  private static void doStuff(ConfigurableApplicationContext appContext) throws InterruptedException {
    MyComponent myComponent = appContext.getBean(MyComponent.class);
    for (int i = 0; i < 10; i++) {
      myComponent.submitDeletion();
      Thread.sleep(500);
    }
    // This is just to make the application exit cleanly
    PollableStreamListenerAspect.EXECUTOR_SERVICE.shutdown();
  }
}

Console log:

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
 :: Spring Boot ::        (v2.1.8.RELEASE)

2021-04-20 12:56:03.675  INFO 13560 --- [           main] d.s.spring.q67155048.DemoApplication     : Starting DemoApplication on Xander-Ultrabook with PID 13560 (C:\Users\alexa\Documents\java-src\spring-aop-playground\target\classes started by alexa in C:\Users\alexa\Documents\java-src\spring-aop-playground)
...
2021-04-20 12:56:07.666  INFO 13560 --- [           main] d.s.spring.q67155048.DemoApplication     : Started DemoApplication in 4.65 seconds (JVM running for 7.181)
Receiving message
Logging
  Starting processing #1
Receiving message
  Re-queue
  Finished processing #1
Receiving message
Logging
  Starting processing #2
Receiving message
  Re-queue
  Finished processing #2
Receiving message
Logging
  Starting processing #3
Receiving message
  Re-queue
  Finished processing #3
Receiving message
Logging
  Starting processing #4
Receiving message
  Re-queue
  Finished processing #4
Receiving message
Logging
  Starting processing #5
Receiving message
  Re-queue
  Finished processing #5
2021-04-20 12:56:12.767  INFO 13560 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'applicationTaskExecutor'
...

See? We count 10x "Receiving message", but only 5x "Re-queue" and 5x "Logging". Please note that I numbered the processing calls because they are asynchronous. That way it is easier to follow when they start and finish.

Update in response to user comment:

I have updated my MCVE in order to reproduce your problem with the parameter binding. The new or changed files are:

package de.scrum_master.spring.q67155048;

public class Message<T> {
  private T content;

  public Message(T content) {
    this.content = content;
  }

  @Override
  public String toString() {
    return "Message{" +
      "content=" + content +
      '}';
  }
}

package de.scrum_master.spring.q67155048;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;

@SpringBootApplication
@Configuration
public class DemoApplication {
  public static void main(String[] args) throws InterruptedException {
    try (ConfigurableApplicationContext appContext = SpringApplication.run(DemoApplication.class, args)) {
      doStuff(appContext);
    }
  }

  private static void doStuff(ConfigurableApplicationContext appContext) throws InterruptedException {
    MyComponent myComponent = appContext.getBean(MyComponent.class);
    Message<String> message = new Message<>("Hi there!");
    for (int i = 0; i < 10; i++) {
      myComponent.submitDeletion(message);
      Thread.sleep(500);
    }
    // This is just to make the application exit cleanly
    PollableStreamListenerAspect.EXECUTOR_SERVICE.shutdown();
  }
}

package de.scrum_master.spring.q67155048;

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Aspect
@Component
@Order(value = Ordered.HIGHEST_PRECEDENCE)
public class PollableStreamListenerAspect {
  public static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(1);
  private volatile boolean paused = false;

  @Around("@annotation(de.scrum_master.spring.q67155048.PollableStreamListener) && args(dataCapsule, ..)")
  public void receiveMessage(ProceedingJoinPoint joinPoint, Object dataCapsule) throws Throwable {
    System.out.println("Receiving message");
    if (!paused) {
      // The separate thread is not busy with a previous message, so process this message:
      Runnable runnable = () -> {
        try {
          paused = true;
          System.out.println("dataCapsule = " + dataCapsule);
          joinPoint.proceed();
        }
        catch (Throwable throwable) {
          throwable.printStackTrace();
        }
        finally {
          paused = false;
        }
      };
      EXECUTOR_SERVICE.submit(runnable);
    }
    else {
      System.out.println("  Re-queue");
    }
  }
}

This yields, according to your own experience:

Exception in thread "main" java.lang.IllegalStateException: Required to bind 2 arguments, but only bound 1 (JoinPointMatch was NOT bound in invocation)
    at org.springframework.aop.aspectj.AbstractAspectJAdvice.argBinding(AbstractAspectJAdvice.java:605)
    at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:633)
    at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:70)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
    at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:688)
    at de.scrum_master.spring.q67155048.MyComponent$$EnhancerBySpringCGLIB$$4baa410d.submitDeletion(<generated>)
    at de.scrum_master.spring.q67155048.DemoApplication.doStuff(DemoApplication.java:21)
    at de.scrum_master.spring.q67155048.DemoApplication.main(DemoApplication.java:13)

You are hitting this problem, and I have commented on the closed Spring issue #16956 about it, hoping to get it repoened and someone to fix it.

For now, your workaround is not to use the elegant AOP argument binding but to fetch the parameter manually using JoinPoint.getArgs():

package de.scrum_master.spring.q67155048;

import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Aspect
@Component
@Order(value = Ordered.HIGHEST_PRECEDENCE)
public class PollableStreamListenerAspect {
  public static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(1);
  private volatile boolean paused = false;

  //@Around("@annotation(de.scrum_master.spring.q67155048.PollableStreamListener) && args(dataCapsule, ..)")
  @Around("@annotation(de.scrum_master.spring.q67155048.PollableStreamListener) && execution(* *(*, ..))")
  //public void receiveMessage(ProceedingJoinPoint joinPoint, Object dataCapsule) throws Throwable {
  public void receiveMessage(ProceedingJoinPoint joinPoint) throws Throwable {
    Object dataCapsule = joinPoint.getArgs()[0];
    System.out.println("Receiving message");
    if (!paused) {
      // The separate thread is not busy with a previous message, so process this message:
      Runnable runnable = () -> {
        try {
          paused = true;
          System.out.println("dataCapsule = " + dataCapsule);
          joinPoint.proceed();
        }
        catch (Throwable throwable) {
          throwable.printStackTrace();
        }
        finally {
          paused = false;
        }
      };
      EXECUTOR_SERVICE.submit(runnable);
    }
    else {
      System.out.println("  Re-queue");
    }
  }
}

Now it works again like this:

Receiving message
dataCapsule = Message{content=Hi there!}
Logging
  Starting processing #1, message = Message{content=Hi there!}
Receiving message
  Re-queue
Receiving message
  Re-queue
  Finished processing #1, message = Message{content=Hi there!}
Receiving message
dataCapsule = Message{content=Hi there!}
Logging
  Starting processing #2, message = Message{content=Hi there!}
Receiving message
  Re-queue
  Finished processing #2, message = Message{content=Hi there!}
Receiving message
dataCapsule = Message{content=Hi there!}
Logging
  Starting processing #3, message = Message{content=Hi there!}
Receiving message
  Re-queue
  Finished processing #3, message = Message{content=Hi there!}
Receiving message
dataCapsule = Message{content=Hi there!}
Logging
  Starting processing #4, message = Message{content=Hi there!}
Receiving message
  Re-queue
  Finished processing #4, message = Message{content=Hi there!}
Receiving message
dataCapsule = Message{content=Hi there!}
Logging
  Starting processing #5, message = Message{content=Hi there!}