Mock S3AsyncClient的CompletableFuture解决单元测试空指针异常

我在用AWS Java 2 SDK编写单元测试，测试代码调用了S3AsyncClient的selectObjectContent方法（该方法返回CompletableFuture），但测试时在future.get()行抛出空指针异常。

待测试方法 #

public <T> Collection<T> queryWithS3Select(
    List<String> s3Keys,
    String s3SelectQuery,
    InputSerialization inputSerialization,
    Class<T> modelObject,
    Comparator<T> comparator
) throws ExecutionException, InterruptedException, IOException {
    TreeSet<T> collection = new TreeSet<>(comparator);

    List<SelectObjectContentRequest> selectObjectContentRequest =
        buildS3SelectRequests(s3Keys, s3SelectQuery, inputSerialization);

    S3SelectContentHandler s3SelectContentHandler = new S3SelectContentHandler();

    StringBuilder selectionResult = new StringBuilder();

    for (SelectObjectContentRequest socr : selectObjectContentRequest) {
        CompletableFuture<Void> future = s3AsyncClient.selectObjectContent(socr, s3SelectContentHandler);
        future.get();

        s3SelectContentHandler.getReceivedEvents().forEach(e -> {
            if (e.sdkEventType() == SelectObjectContentEventStream.EventType.RECORDS) {
                RecordsEvent response = (RecordsEvent) e;
                selectionResult.append(response.payload().asUtf8String());
            }
        });
    }

    JsonParser parser = objectMapper.createParser(selectionResult.toString());
    collection.addAll(Lists.newArrayList(objectMapper.readValues(parser, modelObject)));
    return collection;
}

单元测试代码 #

@Mock
private S3AsyncClient s3AsyncClient;

@Test
public void itShouldReturnQueryResults() throws IOException, ExecutionException, InterruptedException {
    List<String> keysToQuery = List.of("key1", "key2");

    InputSerialization inputSerialization = InputSerialization.builder()
            .json(JSONInput.builder().type(JSONType.DOCUMENT).build())
            .compressionType(String.valueOf(CompressionType.GZIP))
            .build();

    Comparator<S3SelectObject> comparator =
            Comparator.comparing((S3SelectObject e) -> e.getStartTime());


    underTest.queryWithS3Select(keysToQuery, S3_SELECT_QUERY, inputSerialization, S3SelectObject.class, comparator );

}

S3SelectContentHandler代码 #

public class S3SelectContentHandler implements SelectObjectContentResponseHandler {

    private SelectObjectContentResponse response;
    private List<SelectObjectContentEventStream> receivedEvents = new ArrayList<>();
    private Throwable exception;

    @Override
    public void responseReceived(SelectObjectContentResponse response) {
        this.response = response;
    }

    @Override
    public void onEventStream(SdkPublisher<SelectObjectContentEventStream> publisher) {
        publisher.subscribe(receivedEvents::add);
    }

    @Override
    public void exceptionOccurred(Throwable throwable) {
        exception = throwable;
    }

    @Override
    public void complete() {}

    public List<SelectObjectContentEventStream> getReceivedEvents() {
        return receivedEvents;
    }
}

空指针异常的核心原因是：mock的S3AsyncClient未配置返回值，调用selectObjectContent时返回了null，导致future.get()触发NPE。需要通过Mock框架（以Mockito为例）配置方法返回有效CompletableFuture，同时模拟S3 Select的事件流触发处理器回调。

1. 配置Mock返回有效CompletableFuture #

通过Mockito的when方法，让selectObjectContent返回已完成的CompletableFuture<Void>，同时手动触发S3SelectContentHandler的回调逻辑，注入模拟的测试事件。

2. 修改后的测试代码示例 #

@Mock
private S3AsyncClient s3AsyncClient;

@InjectMocks
private YourServiceClass underTest; // 替换为你的实际服务类

@Test
public void itShouldReturnQueryResults() throws IOException, ExecutionException, InterruptedException {
    List<String> keysToQuery = List.of("key1", "key2");

    InputSerialization inputSerialization = InputSerialization.builder()
            .json(JSONInput.builder().type(JSONType.DOCUMENT).build())
            .compressionType(String.valueOf(CompressionType.GZIP))
            .build();

    Comparator<S3SelectObject> comparator =
            Comparator.comparing(S3SelectObject::getStartTime);

    // 构造模拟的测试JSON数据，匹配S3SelectObject的结构
    String testJsonData = "{\"startTime\": \"2024-01-01\"}\n{\"startTime\": \"2024-01-02\"}";
    RecordsEvent mockRecordsEvent = RecordsEvent.builder()
            .payload(SdkBytes.fromUtf8String(testJsonData))
            .build();

    // 配置S3AsyncClient的mock行为
    when(s3AsyncClient.selectObjectContent(any(SelectObjectContentRequest.class), any(SelectObjectContentResponseHandler.class)))
            .thenAnswer(invocation -> {
                // 获取传入的处理器实例
                SelectObjectContentResponseHandler handler = invocation.getArgument(1);
                // 触发responseReceived回调（可选，根据业务逻辑是否需要）
                handler.responseReceived(SelectObjectContentResponse.builder().build());
                // 触发事件流，传入模拟的RecordsEvent
                handler.onEventStream(SdkPublisher.of(mockRecordsEvent));
                // 触发complete回调
                handler.complete();
                // 返回已完成的CompletableFuture，避免NPE
                return CompletableFuture.completedFuture(null);
            });

    // 执行测试方法
    Collection<S3SelectObject> result = underTest.queryWithS3Select(keysToQuery, S3_SELECT_QUERY, inputSerialization, S3SelectObject.class, comparator);

    // 断言结果符合预期
    assertEquals(2, result.size());
}

关键说明 #

• 返回已完成的Future：使用CompletableFuture.completedFuture(null)确保future.get()不会抛出NPE，同时无需等待异步执行。
• 触发处理器回调：必须手动调用responseReceived、onEventStream、complete方法，否则receivedEvents列表为空，后续JSON解析会失败。
• 模拟业务数据：构造符合S3SelectObject结构的JSON字符串，保证objectMapper能正确解析实例。

内容的提问来源于stack exchange，提问作者bostonjava