@Override
  public QueuedCommandStatus enqueueCommand(
      final CommandId commandId,
      final Command command,
      final Producer<CommandId, Command> transactionalProducer
  ) {
    final CommandStatusFuture statusFuture = commandStatusMap.compute(
        commandId,
        (k, v) -> {
          if (v == null) {
            return new CommandStatusFuture(commandId);
          }
          // We should fail registration if a future is already registered, to prevent
          // a caller from receiving a future for a different statement.
          throw new IllegalStateException(
              String.format(
                  "Another command with the same id (%s) is being executed.",
                  commandId)
          );
        }
    );
    try {
      final ProducerRecord<CommandId, Command> producerRecord = new ProducerRecord<>(
          commandTopicName,
          COMMAND_TOPIC_PARTITION,
          commandId,
          command);
      final RecordMetadata recordMetadata =
          transactionalProducer.send(producerRecord).get();
      return new QueuedCommandStatus(recordMetadata.offset(), statusFuture);
    } catch (final Exception e) {
      commandStatusMap.remove(commandId);
      throw new KsqlStatementException(
          "Could not write the statement into the command topic.",
          String.format(
              "Could not write the statement '%s' into the command topic.",
              QueryMask.getMaskedStatement(command.getStatement())
          ),
          QueryMask.getMaskedStatement(command.getStatement()),
          KsqlStatementException.Problem.OTHER,
          e
      );
    }
  }
@Test
  public void shouldCleanupCommandStatusOnProduceError() {
    // Given:
    when(transactionalProducer.send(any(ProducerRecord.class)))
        .thenThrow(new RuntimeException("oops"))
        .thenReturn(testFuture);

    final Exception e = assertThrows(
        KsqlException.class,
        () -> commandStore.enqueueCommand(commandId, command, transactionalProducer)
    );

    assertThat(e.getMessage(), containsString(
        "Could not write the statement into the command topic"
    ));

    // When:
    commandStore.enqueueCommand(commandId, command, transactionalProducer);

    // Then: did not throw.
  }