这个解决方案怎么样？
事实上你用的是retrywhen错误。retrywhen操作符只是发出是否应该重试的信号。每次内部可观察对象发出一个值时，就会开始重试。
在我的示例中，通过onerrorresumnext捕获403异常，该异常捕获onerror发出并订阅可观察的回退。retrywhen after此运算符确保其他错误将重试3次。
仅供参考：每个运算符转换日期只能在不可变数据中工作，因此应避免使用可变列表。

class So65040953 {
    @Test
    fun main() {
        val client = YouTubeClientInterceptor(YouTubeClientStub())
        val scheduler = TestScheduler()

        val test = client.searchRequest("fail")
                // when 403 -> request key and call search again...
                // else propagate error
                .onErrorResumeNext {
                    when (it.message) {
                        "403" -> client.requestApiKey()
                                .flatMapObservable { key -> client.searchRequest("good") }
                        else -> Observable.error(it)
                    }
                }
                // when some error other than 403 happens... retry ever second for three times....
                // error could be thrown by #searchRequest or #requestApiKey
                .retryWhen { throwableObservable ->
                    throwableObservable.take(3).delay(1L, TimeUnit.SECONDS, scheduler)
                }.test()

        scheduler.advanceTimeBy(10, TimeUnit.SECONDS)

        assertThat(client.requestApiKey.get()).isEqualTo(1)
        assertThat(client.searchRequestCount.get()).isEqualTo(2)
        test.assertNoErrors()
                .assertValue(listOf(Video("42")))
    }
}

internal interface YouTubeClient {
    fun searchRequest(query: String): Observable<List<Video>>

    fun requestApiKey(): Single<String>
}

internal class YouTubeClientInterceptor(private val client: YouTubeClient) : YouTubeClient {
    internal val searchRequestCount = AtomicInteger(0)
    internal val requestApiKey = AtomicInteger(0)

    override fun searchRequest(query: String): Observable<List<Video>> {
        searchRequestCount.incrementAndGet()
        return client.searchRequest(query)
    }

    override fun requestApiKey(): Single<String> {
        requestApiKey.incrementAndGet()
        return client.requestApiKey()
    }
}

internal class YouTubeClientStub : YouTubeClient {
    override fun searchRequest(query: String): Observable<List<Video>> {
        println("query: $query")

        return when (query) {
            "fail" -> Observable.error<List<Video>>(RuntimeException("403"))
            "good" -> Observable.just(listOf(Video("42")))
            else -> Observable.empty()
        }
    }

    override fun requestApiKey(): Single<String> {
        println("requestApiKey")

        return Single.just("1-1-1-1")
    }
}

internal data class Video(private val videoId: String)