fix the 1.27 forward compatibility with 1.28
We detected some cases when the  1.27 submitter and finisher don't handle correctly the transfers submitted by 1.28. Fix this

diff --git a/lib/rucio/core/request.py b/lib/rucio/core/request.py
index 8d8287739..f537f1c02 100644
--- a/lib/rucio/core/request.py
+++ b/lib/rucio/core/request.py
@@ -49,7 +49,7 @@ from six import string_types
 from sqlalchemy import and_, or_, func, update, inspect
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.sql.expression import asc, true
+from sqlalchemy.sql.expression import asc, true, null
 from sqlalchemy.schema import Table, Column
 
 from rucio.common.cache import make_region_memcached
@@ -89,12 +89,12 @@ REGION = make_region_memcached(expiration_time=120)
 TRANSFERS_HOP_TABLE = None
 
 
-def __delete_from_transfers_hops_table_if_it_exists(request_id, session, logger):
+def get_transfer_hops_table(session):
     """
-    This function implements forward compatibility of the current 1.27 rucio release line
-    with the 1.28 rucio release which introduces a new table.
-    The goal is to allow parallel execution of 1.27 and 1.28 during the migration phase.
+    Retrieve the transfers_hop table, which doesn't exist in 1.27, but exists in 1.28 rucio.
+    This is to enable forward compatibility for running rucio 1.27 and 1.28 together on the same database.
     """
+
     global TRANSFERS_HOP_TABLE
 
     if not TRANSFERS_HOP_TABLE:
@@ -126,12 +126,21 @@ def __delete_from_transfers_hops_table_if_it_exists(request_id, session, logger)
                     "primary_key": columns,
                 }
             TRANSFERS_HOP_TABLE = DeclarativeObj
+    return TRANSFERS_HOP_TABLE
+
 
-    if TRANSFERS_HOP_TABLE:
+def __delete_from_transfers_hops_table_if_it_exists(request_id, session, logger):
+    """
+    This function implements forward compatibility of the current 1.27 rucio release line
+    with the 1.28 rucio release which introduces a new table.
+    The goal is to allow parallel execution of 1.27 and 1.28 during the migration phase.
+    """
+    transfers_hop_table = get_transfer_hops_table(session)
+    if transfers_hop_table:
         logger(logging.DEBUG, "Deleting from transfer_hops table for request %s", request_id)
-        session.query(TRANSFERS_HOP_TABLE).filter(or_(TRANSFERS_HOP_TABLE.request_id == request_id,
-                                                      TRANSFERS_HOP_TABLE.next_hop_request_id == request_id,
-                                                      TRANSFERS_HOP_TABLE.initial_request_id == request_id)).delete()
+        session.query(transfers_hop_table).filter(or_(transfers_hop_table.request_id == request_id,
+                                                      transfers_hop_table.next_hop_request_id == request_id,
+                                                      transfers_hop_table.initial_request_id == request_id)).delete()
 
 
 def should_retry_request(req, retry_protocol_mismatches):
@@ -341,7 +350,7 @@ def queue_requests(requests, session=None, logger=logging.log):
 @read_session
 def get_next(request_type, state, limit=100, older_than=None, rse_id=None, activity=None,
              total_workers=0, worker_number=0, mode_all=False, hash_variable='id',
-             activity_shares=None, transfertool=None, session=None):
+             activity_shares=None, include_dependent=True, transfertool=None, session=None):
     """
     Retrieve the next requests matching the request type and state.
     Workers are balanced via hashing to reduce concurrency on database.
@@ -357,6 +366,7 @@ def get_next(request_type, state, limit=100, older_than=None, rse_id=None, activ
     :param mode_all:          If set to True the function returns everything, if set to False returns list of dictionaries  {'request_id': x, 'external_host': y, 'external_id': z}.
     :param hash_variable:     The variable to use to perform the partitioning. By default it uses the request id.
     :param activity_shares:   Activity shares dictionary, with number of requests
+    :param include_dependent: If true, includes transfers which have a previous hop dependency on other transfers
     :param transfertool:      The transfer tool as specified in rucio.cfg.
     :param session:           Database session to use.
     :returns:                 Request as a dictionary.
@@ -394,6 +404,16 @@ def get_next(request_type, state, limit=100, older_than=None, rse_id=None, activ
                                                  .filter(models.Request.request_type.in_(request_type))\\
                                                  .order_by(asc(models.Request.updated_at))
 
+        if not include_dependent:
+            # filter out transfers which depend on some other "previous hop" requests.
+            # In particular, this is used to avoid multiple finishers trying to archive different
+            # transfers from the same path and thus having concurrent deletion of same rows from
+            # the transfer_hop table.
+            transfers_hop_table = get_transfer_hops_table(session)
+            if transfers_hop_table:
+                query = query.outerjoin(transfers_hop_table, transfers_hop_table.next_hop_request_id == models.Request.id) \\
+                    .filter(transfers_hop_table.next_hop_request_id == null())
+
         if isinstance(older_than, datetime.datetime):
             query = query.filter(models.Request.updated_at < older_than)
 
diff --git a/lib/rucio/core/transfer.py b/lib/rucio/core/transfer.py
index 7793170e5..31645c2e5 100644
--- a/lib/rucio/core/transfer.py
+++ b/lib/rucio/core/transfer.py
@@ -53,7 +53,7 @@ from typing import TYPE_CHECKING
 from dogpile.cache.api import NoValue
 from sqlalchemy import and_, update
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.sql.expression import false
+from sqlalchemy.sql.expression import false, null
 
 from rucio.common import constants
 from rucio.common.cache import make_region_memcached
@@ -69,7 +69,7 @@ from rucio.core import did, message as message_core, request as request_core
 from rucio.core.config import get as core_config_get
 from rucio.core.monitor import record_counter, record_timer
 from rucio.core.replica import add_replicas, tombstone_from_delay, update_replica_state
-from rucio.core.request import get_request_by_did, queue_requests, set_request_state
+from rucio.core.request import get_request_by_did, queue_requests, set_request_state, get_transfer_hops_table
 from rucio.core.rse import get_rse_name, get_rse_vo, list_rses
 from rucio.core.rse_expression_parser import parse_expression
 from rucio.db.sqla import models, filter_thread_work
@@ -1684,6 +1684,11 @@ def __list_transfer_requests_and_source_replicas(
         .filter(models.RSE.deleted == false()) \\
         .order_by(models.Request.created_at)
 
+    transfers_hop_table = get_transfer_hops_table(session)
+    if transfers_hop_table:
+        sub_requests = sub_requests.outerjoin(transfers_hop_table, transfers_hop_table.next_hop_request_id == models.Request.id) \\
+            .filter(transfers_hop_table.next_hop_request_id == null())
+
     if not ignore_availability:
         sub_requests = sub_requests.filter(models.RSE.availability.in_((2, 3, 6, 7)))
 
diff --git a/lib/rucio/daemons/conveyor/finisher.py b/lib/rucio/daemons/conveyor/finisher.py
index ba42006db..1a6e71f22 100644
--- a/lib/rucio/daemons/conveyor/finisher.py
+++ b/lib/rucio/daemons/conveyor/finisher.py
@@ -88,6 +88,7 @@ def run_once(bulk, db_bulk, suspicious_patterns, retry_protocol_mismatches, hear
                                      total_workers=total_workers,
                                      worker_number=worker_number,
                                      mode_all=True,
+                                     include_dependent=False,
                                      hash_variable='rule_id')
         record_timer('daemons.conveyor.finisher.get_next', (time.time() - time1) * 1000)
         time2 = time.time()
['lib/rucio/daemons/conveyor/finisher.py', 'lib/rucio/core/request.py', 'lib/rucio/core/transfer.py']