NullPointerException when sending out a response with no request metadata

[npepinpe]

Description

The GrpcResponseWriter currently always expects a pending request for every response it sends out, and will explode with NullPointerException if this is not the case. Unfortunately, in the Zeebe engine, we sometimes send out responses anyway because the transport there simply ignores responses with no request metadata, i.e. request ID and stream ID are both the SBE null value, or TypedRecord#hasRequestMetadata returns false.

We should either mimic this behavior, or fix it on the Zeebe side. This currently prevents users from using any intermediate signal throw event.

See: camunda/camunda#15649
Support: https://jira.camunda.com/browse/SUPPORT-19727

Expected behaviour

No NPEs are thrown, or the issue is fixed on the Zeebe side (no responses are sent without request metadata). It would likely be good to anyway catch it here.

Reproduction steps

Test code

/*
 * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
 * one or more contributor license agreements. See the NOTICE file distributed
 * with this work for additional information regarding copyright ownership.
 * Licensed under the Zeebe Community License 1.1. You may not use this file
 * except in compliance with the Zeebe Community License 1.1.
 */

package io.camunda.zeebe.process.test.engine;

import io.camunda.zeebe.client.ZeebeClient;
import io.camunda.zeebe.model.bpmn.Bpmn;
import io.camunda.zeebe.process.test.api.ZeebeTestEngine;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

@ExtendWith(PrintRecordStreamExtension.class)
public class EngineTest {
  private ZeebeTestEngine zeebeEngine;
  private ZeebeClient zeebeClient;

  @BeforeEach
  void setupGrpcServer() {
    zeebeEngine = EngineFactory.create();
    zeebeEngine.start();
    zeebeClient = zeebeEngine.createClient();
  }

  @AfterEach
  void tearDown() {
    zeebeEngine.stop();
    zeebeClient.close();
  }

  // regression:
  @Test
  void shouldNotThrowNpeOnPublishMessageWhichBroadcastsSignal()
      throws InterruptedException, TimeoutException {
    // given
    final var process =
        Bpmn.createExecutableProcess("process")
            .startEvent()
            .intermediateCatchEvent(
                "start", b -> b.message(m -> m.name("start").zeebeCorrelationKey("=businessKey")))
            .intermediateThrowEvent("signal", b -> b.signal("signal"))
            .intermediateCatchEvent(
                "wait", b -> b.message(m -> m.name("wait").zeebeCorrelationKey("=businessKey")))
            .endEvent()
            .done();
    zeebeClient.newDeployResourceCommand().addProcessModel(process, "process.bpmn").send().join();
    final var pi =
        zeebeClient
            .newCreateInstanceCommand()
            .bpmnProcessId("process")
            .latestVersion()
            .variables(Map.of("businessKey", "42"))
            .send()
            .join()
            .getProcessInstanceKey();
    zeebeEngine.waitForIdleState(Duration.ofSeconds(600));

    // when
    zeebeClient.newPublishMessageCommand().messageName("start").correlationKey("42").send().join();
    zeebeEngine.waitForIdleState(Duration.ofSeconds(600));

    // then

  }
}

[nicpuppa]

ZPA Triage:

• Size is medium because we need to investigate more on this
• Priority upcoming because ZPT is unusable for Signal Events

[ana-vinogradova-camunda]

Assigning to myself as I was working on a fix from Zeebe side.

[ana-vinogradova-camunda]

@npepinpe this issue was fixed on Zeebe side in this PR.
Is there anything else that should be done for this issue?

[npepinpe]

Yes, to prevent future cases, we should also here not try to send a response if there is no associated request. For example, in Zeebe, we know that we should send a response only if the record's metadata (i.e. RecordMetadata) has non-null requestStreamId and requestId properties. So here, for example, in GrpcResponseWriter, in tryWriteResponse, we might want to add the same check - if the requestStreamId or requestId are null (respectively RecordMetadataEncoder.requestStreamIdNullValue() and RecordMetadataEncoder.requestIdNullValue()), then we don't even try to send anything.