Unique constraint violation for process instance waiting in an async start event during rolling update

Setup:
Rolling update scenario: engine node on 7.13.0 and engine node on 7.12.x

Steps to reproduce:

1. Deploy version 1 of a process containing a start event with camunda:async="true"; make sure that the job execution fails so that the process instance is waiting in the async start event
2. Start a process instance on 7.13.0 engine and pass at least one variable to the process instance

   runtimeService
     .startProcessInstanceByKey(
       "key",
       Variables.createVariables().putValue("foo", "bar")
     );
   

3. Execute the job on 7.12.10 engine

Expected behavior:
The job should be executed successfully without any exception.

Observed behavior:
The job fails with unique constrain violation exception.

16:33:32.406 [camundaTaskExecutor-2] ERROR org.camunda.bpm.engine.context - ENGINE-16004 Exception while closing command context: ENGINE-03004 Exception while executing Database Operation 'INSERT HistoricVariableInstanceEntity[6]' with message '
### Error flushing statements.  Cause: org.apache.ibatis.executor.BatchExecutorException: org.camunda.bpm.engine.impl.persistence.entity.HistoricVariableInstanceEntity.insertHistoricVariableInstance (batch index #1) failed. Cause: org.h2.jdbc.JdbcBatchUpdateException: Eindeutiger Index oder Primärschlüssel verletzt: "PUBLIC.PRIMARY_KEY_4E ON PUBLIC.ACT_HI_VARINST(ID_) VALUES 1"
Unique index or primary key violation: "PUBLIC.PRIMARY_KEY_4E ON PUBLIC.ACT_HI_VARINST(ID_) VALUES 1"; SQL statement:
insert into ACT_HI_VARINST (
      ID_,
      PROC_DEF_KEY_,
      PROC_DEF_ID_,
      ROOT_PROC_INST_ID_,
      PROC_INST_ID_,
      EXECUTION_ID_,
      ACT_INST_ID_,
      TENANT_ID_,
      CASE_DEF_KEY_,
      CASE_DEF_ID_,
      CASE_INST_ID_,
      CASE_EXECUTION_ID_,
      TASK_ID_,
      NAME_,
      REV_,
      VAR_TYPE_,
      CREATE_TIME_,
      REMOVAL_TIME_,
      BYTEARRAY_ID_,
      DOUBLE_,
      LONG_,
      TEXT_,
      TEXT2_,
      STATE_
    ) values (
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?,
      ?
    ) [23505-199]
...

Hints:

• With CAM-10978 we store the historic variables in the database right away (7.13 node), but in prior versions, the historic variables are stored during the execution of the job (7.12 node). That leads to the exception when the data is committed for a second time. The rolling update scenario should be covered for the historic variables stored on async start event with respect of the changes made in CAM-10978.

Solution Idea

• On a 7.12 engine, we need to know if a variable was created in 7.13 or higher (so that in this case we skip firing the history events)
• We already have a solution for the other direction (telling if a variable was created in 7.12 or earlier): https://github.com/camunda/camunda-bpm-platform/blob/7.14.0/engine/src/main/java/org/camunda/bpm/engine/impl/pvm/runtime/LegacyBehavior.java#L656. This based on the process definition id field that was added in 7.13, so that it is NULL for all pre-713 values
• We can use that in 7.12 as follows:
  • When we select the variables of an execution, we already do a select * (see VariableInstance.xml selectVariablesByExecutionId, so the new process definition id field is selected if it exists
  • We can then try to map it to a POJO property. This requires a custom type handler in Mybatis to avoid an exception if the field does not exist (see e.g. https://stackoverflow.com/questions/3599861/how-can-i-determine-if-the-column-name-exist-in-the-resultset for how to determine if a result set has a certain column)
  • If the mapping is successful and we have a non-null process definition id value, then we know that the variable was created in 7.13 or higher and can skip firing the history events again
  • To make the fix minimal, we can use a custom result map that applies only to the selectVariablesByExecutionId mapping, so that we never perform this logic on other variable-selecting statements where it is not needed (e.g. API queries)