feat(source): create source stream with retry during recovery

feat(source): create source stream with retry during recovery #45616