unit testing - How to test ListenableFuture Callbacks in spock

Question

I asked a question a few days ago regarding stubbing the future response from the kafka.send() method. this was answered and explained correctly by @kriegaex here Though I faced another issue, on how can i test the onSuccess and onFailure callbacks of this future response. here's the code under testing.

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

public class KakfaService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final LogService logService;

    public KakfaService(KafkaTemplate kafkaTemplate, LogService logService){
        this.kafkaTemplate = kafkaTemplate;
        this.logService = logService;
    }

    public void sendMessage(String topicName, String message) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

            @Override
            public void onSuccess(SendResult<String, String> result) {
              LogDto logDto = new LogDto();
              logDto.setStatus(StatusEnum.SUCCESS);
              logService.create(logDto)
            }
            @Override
            public void onFailure(Throwable ex) {
              LogDto logDto = new LogDto();
              logDto.setStatus(StatusEnum.FAILED);
              logService.create(logDto)
            }
        });
    }
}

and here's the tests code

import com…….KafkaService
import com…….LogService
import org.apache.kafka.clients.producer.RecordMetadata
import org.apache.kafka.common.TopicPartition
import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.support.SendResult
import org.springframework.util.concurrent.ListenableFuture
import org.springframework.util.concurrent.ListenableFutureCallback
import org.springframework.util.concurrent.SettableListenableFuture
import spock.lang.Specification

public class kafaServiceTest extends Specification {

    private KafkaTemplate<String, String> kafkaTemplate;
    private KafkaService kafaService;
    private SendResult<String, String> sendResult;
    private SettableListenableFuture<SendResult<?, ?>> future;
    private RecordMetadata recordMetadata
    private String topicName
    private String message


    def setup() {
        topicName = "test.topic"
        message = "test message"
        sendResult = Mock(SendResult.class);
        future = new SettableListenableFuture<>();
        recordMetadata = new RecordMetadata(new TopicPartition(topicName, 1), 1L, 0L, 0L, 0L, 0, 0);

        kafkaTemplate = Mock(KafkaTemplate.class)

        logService = Mock(LogService.class)
        kafkaSservice = new KafkaSservice(kafkaTemplate, logService);
    }

    def "Test success send message method"() {
        given:
        sendResult.getRecordMetadata() >> recordMetadata
        ListenableFutureCallback listenableFutureCallback = Mock(ListenableFutureCallback.class);
        listenableFutureCallback.onFailure(Mock(Throwable.class))
        future.addCallback(listenableFutureCallback)

        when:
        kafkaService.sendMessage(topicName, message)

        then:
        1 * kafkaTemplate.send(_ as String, _ as String) >> future
        // test success of failed callbacks
    }
}

I've tried this following articles and got me nowhere, I might be misunderstand to usage of this tool.

• https://stackoverflow.com/a/56677098/13258992
• https://stackoverflow.com/a/56677098/13258992
• https://www.baeldung.com/mockito-callbacks
• https://www.javatips.net/api/org.springframework.util.concurrent.successcallback

UPDATE: PARTAILLY WORKING

I was able to hit the onSuccess and onFailure on the callback by using future.set(sendResult) and future.setException(new Throwable()) respectively (thanks to @GarryRussell answer here). but the problem is verifying the behavior on the onSuccess and onFailure method. for example I have a log object entity where I save the status (success or failed), assertion on this behavior always returns true. here's the updated test code for the success scenario.

    def "Test success send message method"() {
        given:
        sendResult.getRecordMetadata() >> recordMetadata
        future.set(sendResult)

        when:
        kafkaService.sendMessage(topicName, message)

        then:
        1 * kafkaTemplate.send(_ as String, _ as String) >> future
        1 * logService.create(_) >> {arguments ->
            final LogDto logDto = arguments.get(0)
            // this assert below should fail
            assert logDto.getStatus() == LogStatus.FAILED 
        }
    }

one more thing that I observe is that when I run the code covarage, theres still a red code indication on the closing curly braces for onSuccess and onFailure callback methods.

Answer 1

General comments

In addition to my comments and because you seem to be a beginner in test automation, especially mock testing, some general advice:

• Tests are not mainly a quality check tool, that's only a desirable side effect.
• Instead, they are a design tool for your application, especially when using TDD. I.e. writing tests helps you refactor your code for simplicity, elegance, readability, maintainability, testability (you might want to read about clean code and software craftsmanship):
  • The tests feed back into the application code, i.e. if it is difficult to test something, you should refactor the code.
  • If you have good test coverage, you can also refactor fearlessly, i.e. if your refactoring breaks existing application logic, your automatic tests will immediately detect it and you can fix a small glitch before it becomes a big mess.
• One typical type of refactoring is removing complexity from methods by factoring out nested layers of logic into layered helper methods or even into specific classes taking care of a certain aspect. It makes the code easier to understand and also easier to test.
• Get yourself acquainted with the Dependency Injection (DI) design pattern. The general principle is called Inversion of Control (IoC).

Having said that, I like to mention that one typical anti pattern in software development leading to problematic application design and bad testability is if classes and methods create their own dependencies inline instead of permitting (or even requiring) the user to inject them.

Answer for question asked

Your situation is a good example: You want to verify that your ListenableFutureCallback callback hooks are being called as expected, but you cannot because that object is created inside the sendMessage method as an anonymous subclass and assigned to a local variable. Local = untestable in an easy way and without dirty tricks like abusing the log service to test a side effect of those callback hooks. Just imagine what would happen if the methods would not log anymore or only based on a specific log level or debug condition: The test would break.

So why don't you factor out the callback instance creation into a special service or at least into a method? The method does not even need to be public, protected or package-scoped would suffice - just not private because you cannot mock private methods.

Here is my MCVE for you. I removed some complexity by replacing your log service by direct console logging in order to demonstrate that you don't need to verify any side effects there.

package de.scrum_master.stackoverflow.q61100974;

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

public class KafkaService {
  private KafkaTemplate<String, String> kafkaTemplate;

  public KafkaService(KafkaTemplate kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }

  public void sendMessage(String topicName, String message) {
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topicName, message);
    future.addCallback(createCallback());
  }

  protected ListenableFutureCallback<SendResult<String, String>> createCallback() {
    return new ListenableFutureCallback<SendResult<String, String>>() {
      @Override
      public void onSuccess(SendResult<String, String> result) {
        System.out.print("Success -> " + result);
      }

      @Override
      public void onFailure(Throwable ex) {
        System.out.print("Failed -> " + ex);
      }
    };
  }
}

package de.scrum_master.stackoverflow.q61100974

import org.springframework.kafka.core.KafkaTemplate
import org.springframework.kafka.support.SendResult
import org.springframework.util.concurrent.ListenableFuture
import org.springframework.util.concurrent.ListenableFutureCallback
import org.springframework.util.concurrent.SettableListenableFuture
import spock.lang.Specification

class KafkaServiceTest extends Specification {

  KafkaTemplate<String, String> kafkaTemplate = Mock()
  ListenableFutureCallback callback = Mock()

  // Inject mock template into spy (wrapping the real service) so we can verify interactions on it later
  KafkaService kafkaService = Spy(constructorArgs: [kafkaTemplate]) {
    // Make newly created helper method return mock callback so we can verify interactions on it later
    createCallback() >> callback
  }

  SendResult<String, String> sendResult = Stub()
  String topicName = "test.topic"
  String message = "test message"
  ListenableFuture<SendResult<String, String>> future = new SettableListenableFuture<>()

  def "sending message succeeds"() {
    given:
    future.set(sendResult)

    when:
    kafkaService.sendMessage(topicName, message)

    then:
    1 * kafkaTemplate.send(topicName, message) >> future
    1 * callback.onSuccess(_)
  }

  def "sending message fails"() {
    given:
    future.setException(new Exception("uh-oh"))

    when:
    kafkaService.sendMessage(topicName, message)

    then:
    1 * kafkaTemplate.send(topicName, message) >> future
    1 * callback.onFailure(_)
  }
}

Please note with regard to the test:

• We are using a Spy on the KafkaService, i.e. a special type of partial mock wrapping an original instance.
• On this spy we stub the new method createCallback() in order to inject a mock callback into the class. This allows us to verify later if interactions such as onSuccess(_) or onFailure(_) have been called on it as expected.
• There is no need to mock or instantiate any of RecordMetadata or TopicPartition.

Enjoy! :-)

---

Update: Some more remarks:

• Spies work, but whenever I use a spy I have an uneasy feeling. Maybe because...
• factoring out methods into protected helper methods is a simple way of enabling the spy to stub the method or to test the method separately. But many developers frown upon the practice of making methods visible (even if just protected and not public) only(?) because it makes the code easier to test. I disagree mostly because as I said: Tests are a design tool and smaller and more focused methods are better to understand, maintain and re-use. That the helper method cannot be private due to the need to stub it, is not so nice sometimes. On the other hand, a protected helper method enables us to override it in a production subclass, so there is one more advantage unrelated to testing.
• So what is the alternative? As I said above, you can extract the code into a focused extra class (inner static class or separate) instead of an extra method. That class can be unit-tested separately and be mocked and injected without having to use a spy. But then of course you need to expose an interface for injecting a collaborator instance via constructor or setter.

There is no perfect solution all developers would agree on. I showed you one that I think is pretty much clean and mentioned another one.