LOGGER.info( "Drained queue of pipeline {} for merge.", pipelineReplica.id() );
operatorTupleQueueManager.releaseDefaultOperatorTupleQueue( regionId, replicaIndex, firstOperatorDef.id() );
}
final NonBlockingTupleQueueDrainerPool drainerPool = new NonBlockingTupleQueueDrainerPool( config, firstOperatorDef );
final Supplier<TuplesImpl> outputSupplier = ( isLastMergedPipeline && ( pipelineReplica.getOperatorCount() == 1 ) )
<BUG>? new NonCachedTuplesImplSupplier( firstOperatorDef.inputPortCount() )
: new CachedTuplesImplSupplier( firstOperatorDef.inputPortCount() );</BUG>
LOGGER.info( "Duplicating first operator {} of {} to {}", firstOperatorDef.id(), pipelineReplica.id(), newPipelineReplicaId );
return firstOperator.duplicate( newPipelineReplicaId, firstOperatorQueue, drainerPool, outputSupplier );
}

LOGGER.info( "Drained queue of pipeline {} for merge.", pipelineReplica.id() );
operatorTupleQueueManager.releaseDefaultOperatorTupleQueue( regionId, replicaIndex, firstOperatorDef.id() );
}
final NonBlockingTupleQueueDrainerPool drainerPool = new NonBlockingTupleQueueDrainerPool( config, firstOperatorDef );
final Supplier<TuplesImpl> outputSupplier = ( isLastMergedPipeline && ( pipelineReplica.getOperatorCount() == 1 ) )
? OutputSupplierUtils.newInstance( lastOperatorOutputSupplierClass,
firstOperatorDef.inputPortCount() )
: new CachedTuplesImplSupplier( firstOperatorDef.inputPortCount() );
LOGGER.info( "Duplicating first operator {} of {} to {}", firstOperatorDef.id(), pipelineReplica.id(), newPipelineReplicaId );
return firstOperator.duplicate( newPipelineReplicaId, firstOperatorQueue, drainerPool, outputSupplier );
}
