core/src/main/java/com/orientechnologies/orient/core/storage/impl/local/OAbstractPaginatedStorage.java
private List<ORecordOperation> commit(final OTransactionInternal transaction, final boolean allocated) {
    // XXX: At this moment, there are two implementations of the commit method. One for regular client transactions and one for
    // implicit micro-transactions. The implementations are quite identical, but operate on slightly different data. If you change
    // this method don't forget to change its counterpart:
    //
    //  OAbstractPaginatedStorage.commit(com.orientechnologies.orient.core.storage.impl.local.OMicroTransaction)

    try {
      checkOpenness();
      checkLowDiskSpaceRequestsAndReadOnlyConditions();

      txBegun.incrementAndGet();

      final ODatabaseDocumentInternal database = transaction.getDatabase();
      final OIndexManager indexManager = database.getMetadata().getIndexManager();
      final TreeMap<String, OTransactionIndexChanges> indexOperations = getSortedIndexOperations(transaction);

      database.getMetadata().makeThreadLocalSchemaSnapshot();

      final Collection<ORecordOperation> recordOperations = transaction.getRecordOperations();
      final TreeMap<Integer, OCluster> clustersToLock = new TreeMap<>();
      final Map<ORecordOperation, Integer> clusterOverrides = new IdentityHashMap<>(8);

      final Set<ORecordOperation> newRecords = new TreeSet<>(COMMIT_RECORD_OPERATION_COMPARATOR);

      for (final ORecordOperation recordOperation : recordOperations) {
        if (recordOperation.type == ORecordOperation.CREATED || recordOperation.type == ORecordOperation.UPDATED) {
          final ORecord record = recordOperation.getRecord();
          if (record instanceof ODocument) {
            ((ODocument) record).validate();
          }
        }

        if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.DELETED) {
          final int clusterId = recordOperation.getRecord().getIdentity().getClusterId();
          clustersToLock.put(clusterId, getClusterById(clusterId));
        } else if (recordOperation.type == ORecordOperation.CREATED) {
          newRecords.add(recordOperation);

          final ORecord record = recordOperation.getRecord();
          final ORID rid = record.getIdentity();

          int clusterId = rid.getClusterId();

          if (record.isDirty() && clusterId == ORID.CLUSTER_ID_INVALID && record instanceof ODocument) {
            // TRY TO FIX CLUSTER ID TO THE DEFAULT CLUSTER ID DEFINED IN SCHEMA CLASS

            final OImmutableClass class_ = ODocumentInternal.getImmutableSchemaClass(((ODocument) record));
            if (class_ != null) {
              clusterId = class_.getClusterForNewInstance((ODocument) record);
              clusterOverrides.put(recordOperation, clusterId);
            }
          }

          clustersToLock.put(clusterId, getClusterById(clusterId));
        }
      }

      final List<ORecordOperation> result = new ArrayList<>(8);
      stateLock.acquireReadLock();
      try {
        if (modificationLock) {
          final List<ORID> recordLocks = new ArrayList<>();
          for (final ORecordOperation recordOperation : recordOperations) {
            if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.DELETED) {
              recordLocks.add(recordOperation.getRID());
            }
          }
          final Set<ORID> locked = transaction.getLockedRecords();
          if (locked != null) {
            recordLocks.removeAll(locked);
          }
          Collections.sort(recordLocks);
          for (final ORID rid : recordLocks) {
            acquireWriteLock(rid);
          }
        }
        try {
          checkOpenness();

          makeStorageDirty();

          boolean rollback = false;
          startStorageTx(transaction);
          try {
            final OAtomicOperation atomicOperation = OAtomicOperationsManager.getCurrentOperation();

            lockClusters(clustersToLock);

            checkReadOnlyConditions();

            final Map<ORecordOperation, OPhysicalPosition> positions = new IdentityHashMap<>(8);
            for (final ORecordOperation recordOperation : newRecords) {
              final ORecord rec = recordOperation.getRecord();

              if (allocated) {
                if (rec.getIdentity().isPersistent()) {
                  positions.put(recordOperation, new OPhysicalPosition(rec.getIdentity().getClusterPosition()));
                } else {
                  throw new OStorageException("Impossible to commit a transaction with not valid rid in pre-allocated commit");
                }
              } else if (rec.isDirty() && !rec.getIdentity().isPersistent()) {
                final ORecordId rid = (ORecordId) rec.getIdentity().copy();
                final ORecordId oldRID = rid.copy();

                final Integer clusterOverride = clusterOverrides.get(recordOperation);
                final int clusterId = clusterOverride == null ? rid.getClusterId() : clusterOverride;

                final OCluster cluster = getClusterById(clusterId);

                assert atomicOperation.getCounter() == 1;
                OPhysicalPosition physicalPosition = cluster.allocatePosition(ORecordInternal.getRecordType(rec));
                assert atomicOperation.getCounter() == 1;
                rid.setClusterId(cluster.getId());

                if (rid.getClusterPosition() > -1) {
                  // CREATE EMPTY RECORDS UNTIL THE POSITION IS REACHED. THIS IS THE CASE WHEN A SERVER IS OUT OF SYNC
                  // BECAUSE A TRANSACTION HAS BEEN ROLLED BACK BEFORE TO SEND THE REMOTE CREATES. SO THE OWNER NODE DELETED
                  // RECORD HAVING A HIGHER CLUSTER POSITION
                  while (rid.getClusterPosition() > physicalPosition.clusterPosition) {
                    assert atomicOperation.getCounter() == 1;
                    physicalPosition = cluster.allocatePosition(ORecordInternal.getRecordType(rec));
                    assert atomicOperation.getCounter() == 1;
                  }

                  if (rid.getClusterPosition() != physicalPosition.clusterPosition) {
                    throw new OConcurrentCreateException(rid, new ORecordId(rid.getClusterId(), physicalPosition.clusterPosition));
                  }
                }
                positions.put(recordOperation, physicalPosition);

                rid.setClusterPosition(physicalPosition.clusterPosition);

                transaction.updateIdentityAfterCommit(oldRID, rid);
              }
            }

            lockRidBags(clustersToLock, indexOperations, indexManager);

            checkReadOnlyConditions();

            for (final ORecordOperation recordOperation : recordOperations) {
              assert atomicOperation.getCounter() == 1;
              commitEntry(recordOperation, positions.get(recordOperation), database.getSerializer());
              assert atomicOperation.getCounter() == 1;
              result.add(recordOperation);
            }

            lockIndexes(indexOperations);

            checkReadOnlyConditions();

            commitIndexes(indexOperations, atomicOperation);
          } catch (final IOException | RuntimeException e) {
            rollback = true;
            if (e instanceof RuntimeException) {
              throw ((RuntimeException) e);
            } else {
              throw OException.wrapException(new OStorageException("Error during transaction commit"), e);
            }
          } finally {
            if (rollback) {
              rollback(transaction);
            } else {
              endStorageTx(transaction, recordOperations);
            }

            this.transaction.set(null);
          }
        } finally {
          atomicOperationsManager.ensureThatComponentsUnlocked();
          database.getMetadata().clearThreadLocalSchemaSnapshot();
        }
      } finally {
        try {
          if (modificationLock) {
            final List<ORID> recordLocks = new ArrayList<>();
            for (final ORecordOperation recordOperation : recordOperations) {
              if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.DELETED) {
                recordLocks.add(recordOperation.getRID());
              }
            }

            final Set<ORID> locked = transaction.getLockedRecords();
            if (locked != null) {
              recordLocks.removeAll(locked);
            }

            for (final ORID rid : recordLocks) {
              releaseWriteLock(rid);
            }
          }
        } finally {
          stateLock.releaseReadLock();
        }
      }

      if (OLogManager.instance().isDebugEnabled()) {
        OLogManager.instance()
            .debug(this, "%d Committed transaction %d on database '%s' (result=%s)", Thread.currentThread().getId(),
                transaction.getId(), database.getName(), result);
      }

      return result;
    } catch (final RuntimeException ee) {
      throw logAndPrepareForRethrow(ee);
    } catch (final Error ee) {
      handleJVMError(ee);
      OAtomicOperationsManager.alarmClearOfAtomicOperation();
      throw logAndPrepareForRethrow(ee);
    } catch (final Throwable t) {
      throw logAndPrepareForRethrow(t);
    }
  }
private List<ORecordOperation> commit(final OTransactionInternal transaction, final boolean allocated) {
    // XXX: At this moment, there are two implementations of the commit method. One for regular client transactions and one for
    // implicit micro-transactions. The implementations are quite identical, but operate on slightly different data. If you change
    // this method don't forget to change its counterpart:
    //
    //  OAbstractPaginatedStorage.commit(com.orientechnologies.orient.core.storage.impl.local.OMicroTransaction)

    try {
      checkOpenness();
      checkLowDiskSpaceRequestsAndReadOnlyConditions();

      txBegun.incrementAndGet();

      final ODatabaseDocumentInternal database = transaction.getDatabase();
      final OIndexManager indexManager = database.getMetadata().getIndexManager();
      final TreeMap<String, OTransactionIndexChanges> indexOperations = getSortedIndexOperations(transaction);

      database.getMetadata().makeThreadLocalSchemaSnapshot();

      final Collection<ORecordOperation> recordOperations = transaction.getRecordOperations();
      final TreeMap<Integer, OCluster> clustersToLock = new TreeMap<>();
      final Map<ORecordOperation, Integer> clusterOverrides = new IdentityHashMap<>(8);

      final Set<ORecordOperation> newRecords = new TreeSet<>(COMMIT_RECORD_OPERATION_COMPARATOR);

      for (final ORecordOperation recordOperation : recordOperations) {
        if (recordOperation.type == ORecordOperation.CREATED || recordOperation.type == ORecordOperation.UPDATED) {
          final ORecord record = recordOperation.getRecord();
          if (record instanceof ODocument) {
            ((ODocument) record).validate();
          }
        }

        if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.DELETED) {
          final int clusterId = recordOperation.getRecord().getIdentity().getClusterId();
          clustersToLock.put(clusterId, getClusterById(clusterId));
        } else if (recordOperation.type == ORecordOperation.CREATED) {
          newRecords.add(recordOperation);

          final ORecord record = recordOperation.getRecord();
          final ORID rid = record.getIdentity();

          int clusterId = rid.getClusterId();

          if (record.isDirty() && clusterId == ORID.CLUSTER_ID_INVALID && record instanceof ODocument) {
            // TRY TO FIX CLUSTER ID TO THE DEFAULT CLUSTER ID DEFINED IN SCHEMA CLASS

            final OImmutableClass class_ = ODocumentInternal.getImmutableSchemaClass(((ODocument) record));
            if (class_ != null) {
              clusterId = class_.getClusterForNewInstance((ODocument) record);
              clusterOverrides.put(recordOperation, clusterId);
            }
          }

          clustersToLock.put(clusterId, getClusterById(clusterId));
        }
      }

      final List<ORecordOperation> result = new ArrayList<>(8);
      stateLock.acquireReadLock();
      try {
        if (modificationLock) {
          final List<ORID> recordLocks = new ArrayList<>();
          for (final ORecordOperation recordOperation : recordOperations) {
            if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.DELETED) {
              recordLocks.add(recordOperation.getRID());
            }
          }
          final Set<ORID> locked = transaction.getLockedRecords();
          if (locked != null) {
            recordLocks.removeAll(locked);
          }
          Collections.sort(recordLocks);
          for (final ORID rid : recordLocks) {
            acquireWriteLock(rid);
          }
        }
        try {
          checkOpenness();

          makeStorageDirty();

          boolean rollback = false;
          startStorageTx(transaction);
          try {
            final OAtomicOperation atomicOperation = OAtomicOperationsManager.getCurrentOperation();

            lockClusters(clustersToLock);

            checkReadOnlyConditions();

            final Map<ORecordOperation, OPhysicalPosition> positions = new IdentityHashMap<>(8);
            for (final ORecordOperation recordOperation : newRecords) {
              final ORecord rec = recordOperation.getRecord();

              if (allocated) {
                if (rec.getIdentity().isPersistent()) {
                  positions.put(recordOperation, new OPhysicalPosition(rec.getIdentity().getClusterPosition()));
                } else {
                  throw new OStorageException("Impossible to commit a transaction with not valid rid in pre-allocated commit");
                }
              } else if (rec.isDirty() && !rec.getIdentity().isPersistent()) {
                final ORecordId rid = (ORecordId) rec.getIdentity().copy();
                final ORecordId oldRID = rid.copy();

                final Integer clusterOverride = clusterOverrides.get(recordOperation);
                final int clusterId = clusterOverride == null ? rid.getClusterId() : clusterOverride;

                final OCluster cluster = getClusterById(clusterId);

                assert atomicOperation.getCounter() == 1;
                OPhysicalPosition physicalPosition = cluster.allocatePosition(ORecordInternal.getRecordType(rec));
                assert atomicOperation.getCounter() == 1;
                rid.setClusterId(cluster.getId());

                if (rid.getClusterPosition() > -1) {
                  // CREATE EMPTY RECORDS UNTIL THE POSITION IS REACHED. THIS IS THE CASE WHEN A SERVER IS OUT OF SYNC
                  // BECAUSE A TRANSACTION HAS BEEN ROLLED BACK BEFORE TO SEND THE REMOTE CREATES. SO THE OWNER NODE DELETED
                  // RECORD HAVING A HIGHER CLUSTER POSITION
                  while (rid.getClusterPosition() > physicalPosition.clusterPosition) {
                    assert atomicOperation.getCounter() == 1;
                    physicalPosition = cluster.allocatePosition(ORecordInternal.getRecordType(rec));
                    assert atomicOperation.getCounter() == 1;
                  }

                  if (rid.getClusterPosition() != physicalPosition.clusterPosition) {
                    throw new OConcurrentCreateException(rid, new ORecordId(rid.getClusterId(), physicalPosition.clusterPosition));
                  }
                }
                positions.put(recordOperation, physicalPosition);

                rid.setClusterPosition(physicalPosition.clusterPosition);

                transaction.updateIdentityAfterCommit(oldRID, rid);
              }
            }

            lockRidBags(clustersToLock, indexOperations, indexManager);

            checkReadOnlyConditions();

            for (final ORecordOperation recordOperation : recordOperations) {
              assert atomicOperation.getCounter() == 1;
              commitEntry(recordOperation, positions.get(recordOperation), database.getSerializer());
              assert atomicOperation.getCounter() == 1;
              result.add(recordOperation);
            }

            lockIndexes(indexOperations);

            checkReadOnlyConditions();

            commitIndexes(indexOperations, atomicOperation);
          } catch (final IOException | RuntimeException e) {
            rollback = true;
            if (e instanceof RuntimeException) {
              throw ((RuntimeException) e);
            } else {
              throw OException.wrapException(new OStorageException("Error during transaction commit"), e);
            }
          } finally {
            if (rollback) {
              rollback(transaction);
            } else {
              endStorageTx(transaction, recordOperations);
            }

            this.transaction.set(null);
          }
        } finally {
          atomicOperationsManager.ensureThatComponentsUnlocked();
          database.getMetadata().clearThreadLocalSchemaSnapshot();
        }
      } finally {
        try {
          if (modificationLock) {
            final List<ORID> recordLocks = new ArrayList<>();
            for (final ORecordOperation recordOperation : recordOperations) {
              if (recordOperation.type == ORecordOperation.UPDATED || recordOperation.type == ORecordOperation.DELETED) {
                recordLocks.add(recordOperation.getRID());
              }
            }

            final Set<ORID> locked = transaction.getLockedRecords();
            if (locked != null) {
              recordLocks.removeAll(locked);
            }

            for (final ORID rid : recordLocks) {
              releaseWriteLock(rid);
            }
          }
        } finally {
          stateLock.releaseReadLock();
        }
      }

      if (OLogManager.instance().isDebugEnabled()) {
        OLogManager.instance()
            .debug(this, "%d Committed transaction %d on database '%s' (result=%s)", Thread.currentThread().getId(),
                transaction.getId(), database.getName(), result);
      }

      return result;
    } catch (final RuntimeException ee) {
      throw logAndPrepareForRethrow(ee);
    } catch (final Error ee) {
      handleJVMError(ee);
      OAtomicOperationsManager.alarmClearOfAtomicOperation();
      throw logAndPrepareForRethrow(ee);
    } catch (final Throwable t) {
      throw logAndPrepareForRethrow(t);
    }
  }
The commit operation can be run in 3 different conditions, embedded commit, pre-allocated commit, other node commit.
<bold>Embedded commit</bold> is the basic commit where the operation is run in embedded or server side, the transaction arrive
with invalid rids that get allocated and committed.
<bold>pre-allocated commit</bold> is the commit that happen after an preAllocateRids call is done, this is usually run by the
coordinator of a tx in distributed.
<bold>other node commit</bold> is the commit that happen when a node execute a transaction of another node where all the rids
are already allocated in the other node.

@param transaction the transaction to commit
@param allocated   true if the operation is pre-allocated commit

@return The list of operations applied by the transaction