EmbeddedKafkaBroker cannot be injected in tests with spring-test context

[celcius112]

In a JUnit 5 test using a spring-test context, it is not possible to autowire the EmbeddedKafkaBroker because there is a resolution clash between EmbeddedKafkaCondition's and SpringExtension's parameter resolution.

For instance:

@EmbeddedKafka(partitions = 1, brokerProperties = "listeners=PLAINTEXT://localhost:9092")
@SpringBootTest(webEnvironment = RANDOM_PORT)
public class MyTest {
  @AfterAll
  static void afterAll(@Autowired EmbeddedKafkaBroker embeddedKafkaBroker) {
    embeddedKafkaBroker.destroy();
  }
}

will fail with error:

org.junit.jupiter.api.extension.ParameterResolutionException: Discovered multiple competing ParameterResolvers for parameter [org.springframework.kafka.test.EmbeddedKafkaBroker embeddedKafkaBroker] in method [static void fr.bpifrance.notification.AbstractKafkaIT.afterAll(org.springframework.kafka.test.EmbeddedKafkaBroker)]: org.springframework.kafka.test.condition.EmbeddedKafkaCondition@f2bee8dd, org.springframework.test.context.junit.jupiter.SpringExtension@5562e7c1

[garyrussell]

@AfterAll is a static method, so injecting the broker here is not correct.

However, the EmbeddedKafkaCondition is not generally involved, since we detect the presence of a spring context and inject the broker into the context instead of creating one in the condition.

I will look when I get back to work next week, but this error is likely an unexpected side effect of trying to inject the broker into a static method (which is wrong).

[celcius112]

I don't think being static is important for JUnit's parameter resolution. It is entirely possible to inject a parameter in a static method with JUnit 5. To prove this, this code will also fail with the same error:

  @AfterEach
  void afterEach(@Autowired EmbeddedKafkaBroker embeddedKafkaBroker) {

  }

However it is true that in my case injecting in a static method prevent me from autowiring it at the class level which would circumvent the issue:

@EmbeddedKafka(partitions = 1, brokerProperties = "listeners=PLAINTEXT://localhost:9092")
@SpringBootTest(webEnvironment = RANDOM_PORT)
public class MyTest {
  
  @Autowired 
  private EmbeddedKafkaBroker embeddedKafkaBroker

  @AfterAll
  static void afterAll() { 
    // compilation error
    embeddedKafkaBroker.destroy();
  }
}

[celcius112]

As for EmbeddedKafkaCondition it is unfortunately automatically involved because of @EmbeddedKafka. In summary:

• @EmbeddedKafka applies EmbeddedKafkaCondition who understands a parameter of type EmbeddedKafkaBroker
• @SpringBootTest applies SpringExtension who understands a parameter annotated by @Autowired

Since both extensions will understand the parameter we are trying to resolve, JUnit 5 will force the failure.

(Sorry disturbing you during your vacations 😃)

[garyrussell]

I don't think being static is important for JUnit's parameter resolution.

Correct - we (I) just didn't anticipate someone doing something like this; as I said in the other issue, shutting down the broker is not enough; the test listener containers need to be shut down too.

In any case, using @Autowired in a static method is just not appropriate.

[celcius112]

Ok let's forget about my example 😛.
A non-static method in a spring-test context, like a @Test or @BeforeEach, will also not be able to inject the broker by parameter.

[garyrussell]

... will also not be able to inject the broker by parameter.

Good point, if you leave off @Autowired, the parameter won't be considered by SpringExtension because it only considers parameters so annotated (unless you change the default auto wire mode on a @TestConstructor or via a property). However, then, the EmbeddedKafkaCondition won't resolve it either, because instead of creating a broker in the condition, it leaves it to the @EmbeddedKafkaBroker to register it with application context instead.

...
Caused by: java.lang.IllegalStateException: Could not find embedded broker instance
	at org.springframework.util.Assert.state(Assert.java:76)
	at org.springframework.kafka.test.condition.EmbeddedKafkaCondition.resolveParameter(EmbeddedKafkaCondition.java:76)
	at org.junit.jupiter.engine.execution.ExecutableInvoker.resolveParameter(ExecutableInvoker.java:216)
	... 56 more

We need to fix supportsParameter() to return false if the condition does not, itself, create a broker.