docs: protected DataSourceMetadataUpdateResult updateDataSourceMetadataWithHandle(
        final Handle handle,
        final String dataSource,
        final DataSourceMetadata startMetadata,
        final DataSourceMetadata endMetadata
    ) throws IOException
    {
      Preconditions.checkNotNull(dataSource, "dataSource");
      Preconditions.checkNotNull(startMetadata, "startMetadata");
      Preconditions.checkNotNull(endMetadata, "endMetadata");
  
      final byte[] oldCommitMetadataBytesFromDb = getDataSourceMetadataWithHandleAsBytes(handle, dataSource);
      final String oldCommitMetadataSha1FromDb;
      final DataSourceMetadata oldCommitMetadataFromDb;
  
      if (oldCommitMetadataBytesFromDb == null) {
        oldCommitMetadataSha1FromDb = null;
        oldCommitMetadataFromDb = null;
      } else {
        oldCommitMetadataSha1FromDb = BaseEncoding.base16().encode(
            Hashing.sha1().hashBytes(oldCommitMetadataBytesFromDb).asBytes()
        );
        oldCommitMetadataFromDb = jsonMapper.readValue(oldCommitMetadataBytesFromDb, DataSourceMetadata.class);
      }
  
      final boolean startMetadataMatchesExisting;
  
      if (oldCommitMetadataFromDb == null) {
        startMetadataMatchesExisting = startMetadata.isValidStart();
      } else {
        // Checking against the last committed metadata.
        // Converting the last one into start metadata for checking since only the same type of metadata can be matched.
        // Even though kafka/kinesis indexing services use different sequenceNumber types for representing
        // start and end sequenceNumbers, the below conversion is fine because the new start sequenceNumbers are supposed
        // to be same with end sequenceNumbers of the last commit.
        startMetadataMatchesExisting = startMetadata.asStartMetadata().matches(oldCommitMetadataFromDb.asStartMetadata());
      }
  
      if (!startMetadataMatchesExisting) {
        // Not in the desired start state.
        log.error(
            "Not updating metadata, existing state[%s] in metadata store doesn't match to the new start state[%s].",
            oldCommitMetadataFromDb,
            startMetadata
        );
        return DataSourceMetadataUpdateResult.FAILURE;
      }
  
      // Only endOffsets should be stored in metadata store
      final DataSourceMetadata newCommitMetadata = oldCommitMetadataFromDb == null
                                                   ? endMetadata
                                                   : oldCommitMetadataFromDb.plus(endMetadata);
      final byte[] newCommitMetadataBytes = jsonMapper.writeValueAsBytes(newCommitMetadata);
      final String newCommitMetadataSha1 = BaseEncoding.base16().encode(
          Hashing.sha1().hashBytes(newCommitMetadataBytes).asBytes()
      );
  
      final DataSourceMetadataUpdateResult retVal;
      if (oldCommitMetadataBytesFromDb == null) {
        // SELECT -> INSERT can fail due to races; callers must be prepared to retry.
        final int numRows = handle.createStatement(
            StringUtils.format(
                "INSERT INTO %s (dataSource, created_date, commit_metadata_payload, commit_metadata_sha1) "
                + "VALUES (:dataSource, :created_date, :commit_metadata_payload, :commit_metadata_sha1)",
                dbTables.getDataSourceTable()
            )
        )
                                  .bind("dataSource", dataSource)
                                  .bind("created_date", DateTimes.nowUtc().toString())
                                  .bind("commit_metadata_payload", newCommitMetadataBytes)
                                  .bind("commit_metadata_sha1", newCommitMetadataSha1)
                                  .execute();
  
        retVal = numRows == 1 ? DataSourceMetadataUpdateResult.SUCCESS : DataSourceMetadataUpdateResult.TRY_AGAIN;
      } else {
        // Expecting a particular old metadata; use the SHA1 in a compare-and-swap UPDATE
        final int numRows = handle.createStatement(
            StringUtils.format(
                "UPDATE %s SET "
                + "commit_metadata_payload = :new_commit_metadata_payload, "
                + "commit_metadata_sha1 = :new_commit_metadata_sha1 "
                + "WHERE dataSource = :dataSource AND commit_metadata_sha1 = :old_commit_metadata_sha1",
                dbTables.getDataSourceTable()
            )
        )
                                  .bind("dataSource", dataSource)
                                  .bind("old_commit_metadata_sha1", oldCommitMetadataSha1FromDb)
                                  .bind("new_commit_metadata_payload", newCommitMetadataBytes)
                                  .bind("new_commit_metadata_sha1", newCommitMetadataSha1)
                                  .execute();
  
        retVal = numRows == 1 ? DataSourceMetadataUpdateResult.SUCCESS : DataSourceMetadataUpdateResult.TRY_AGAIN;
      }
  
      if (retVal == DataSourceMetadataUpdateResult.SUCCESS) {
        log.info("Updated metadata from[%s] to[%s].", oldCommitMetadataFromDb, newCommitMetadata);
      } else {
        log.info("Not updating metadata, compare-and-swap failure.");
      }
  
      return retVal;
    }