This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/release/v1.2/pr-7103-c3aa6e4ba36bdd0cdc9f76bd7896c024dda392b3 in repository https://gitbox.apache.org/repos/asf/texera.git
commit ea12e35bf9ae81d23c600313bb30b7167d8aaba7 Author: Yicong Huang <[email protected]> AuthorDate: Fri Jul 31 09:59:22 2026 -0400 fix(file-service, v1.2): retry LakeFS health check on startup (#7103) ### What changes were proposed in this PR? Backport of #5647 to `release/v1.2`, cherry-picked from 263093aaa54b34996013c717e4bdf7f8902230d0. Follows the Direct Backport Push convention; opened as a PR (rather than a direct push) as part of a backport-coverage audit for fixes merged to `main` since early June that were never labeled for backport. ### Any related issues, documentation, discussions? Backport of #5647. Originally linked #5646. ### How was this PR tested? Release-branch CI runs on this PR. The cherry-pick applied cleanly onto `release/v1.2`; no manual conflict resolution was needed. ### Was this PR authored or co-authored using generative AI tooling? Yes — backport prepared with Claude Code (mechanical cherry-pick; the change itself is #5647 by its original author). Co-authored-by: Matthew B. <[email protected]> Co-authored-by: Xuan Gu <[email protected]> --- .../core/storage/util/LakeFSStorageClient.scala | 55 ++++++++++++++-- .../storage/util/LakeFSStorageClientSpec.scala | 73 ++++++++++++++++++++++ 2 files changed, 123 insertions(+), 5 deletions(-) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala index 613255173e..548579cde5 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClient.scala @@ -19,6 +19,7 @@ package org.apache.texera.amber.core.storage.util +import com.typesafe.scalalogging.LazyLogging import io.lakefs.clients.sdk._ import io.lakefs.clients.sdk.model.ResetCreation.TypeEnum import io.lakefs.clients.sdk.model._ @@ -33,11 +34,16 @@ import scala.jdk.CollectionConverters._ * LakeFSFileStorage provides high-level file storage operations using LakeFS, * similar to Git operations for version control and file management. */ -object LakeFSStorageClient { +object LakeFSStorageClient extends LazyLogging { // Maximum number of results per LakeFS API request (pagination page size) private val PageSize = 1000 + // Health-check retry settings: retry with exponential backoff before giving up. + // 5 attempts starting at 200ms (200, 400, 800, 1600ms) caps total wait at ~3s. + private val HealthCheckMaxAttempts = 5 + private val HealthCheckInitialDelayMillis = 200L + private lazy val apiClient: ApiClient = { val client = new ApiClient() client.setApiKey(StorageConfig.lakefsPassword) @@ -69,11 +75,50 @@ object LakeFSStorageClient { private val branchName: String = "main" def healthCheck(): Unit = { - try { + retryWithBackoff(HealthCheckMaxAttempts, HealthCheckInitialDelayMillis) { this.healthCheckApi.healthCheck().execute() - } catch { - case e: Exception => - throw new RuntimeException(s"Failed to connect to lake fs server: ${e.getMessage}") + } + } + + /** + * Runs `operation`, retrying on failure with exponential backoff (the delay + * doubles after each failed attempt) until it succeeds or `maxAttempts` is + * reached. The final failure is rethrown with the last exception as its cause. + * If interrupted while waiting, restores the interrupt status and fails fast. + * + * `sleep` is injectable so the backoff can be exercised in tests without real waiting. + */ + private[util] def retryWithBackoff( + maxAttempts: Int, + initialDelayMillis: Long, + sleep: Long => Unit = Thread.sleep + )(operation: => Unit): Unit = { + var attempt = 1 + var delayMillis = initialDelayMillis + while (true) { + try { + operation + return + } catch { + case ie: InterruptedException => + // Restore the interrupt status and fail fast rather than retrying. + Thread.currentThread().interrupt() + throw new RuntimeException("Interrupted while waiting to retry lake fs health check", ie) + case e: Exception => + if (attempt >= maxAttempts) { + throw new RuntimeException( + s"Failed to connect to lake fs server after $maxAttempts attempts: ${e.getMessage}", + e + ) + } + logger.warn( + s"LakeFS not reachable (attempt $attempt/$maxAttempts): ${e.getMessage}. " + + s"Retrying in ${delayMillis}ms..." + ) + sleep(delayMillis) + attempt += 1 + delayMillis *= 2 + } } } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala new file mode 100644 index 0000000000..56506cee1f --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/util/LakeFSStorageClientSpec.scala @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.core.storage.util + +import org.scalatest.flatspec.AnyFlatSpec + +import scala.collection.mutable.ListBuffer + +class LakeFSStorageClientSpec extends AnyFlatSpec { + + "retryWithBackoff" should "run the operation once and not sleep when it succeeds immediately" in { + var attempts = 0 + val delays = ListBuffer.empty[Long] + LakeFSStorageClient.retryWithBackoff(5, 200L, delays += _) { + attempts += 1 + } + assert(attempts == 1) + assert(delays.isEmpty) + } + + it should "retry until success and double the delay after each failed attempt" in { + var attempts = 0 + val delays = ListBuffer.empty[Long] + LakeFSStorageClient.retryWithBackoff(5, 200L, delays += _) { + attempts += 1 + if (attempts < 3) throw new RuntimeException("transient") + } + assert(attempts == 3) + assert(delays.toList == List(200L, 400L)) + } + + it should "give up after maxAttempts and preserve the last failure as the cause" in { + var attempts = 0 + val cause = new RuntimeException("still down") + val ex = intercept[RuntimeException] { + LakeFSStorageClient.retryWithBackoff(3, 200L, _ => ()) { + attempts += 1 + throw cause + } + } + assert(attempts == 3) + assert(ex.getMessage.contains("after 3 attempts")) + assert(ex.getCause eq cause) + } + + it should "fail fast and restore the interrupt status when interrupted" in { + val ex = intercept[RuntimeException] { + LakeFSStorageClient.retryWithBackoff(5, 200L, _ => ()) { + throw new InterruptedException("interrupted") + } + } + // Thread.interrupted() both reads and clears the flag, so the interrupt was restored. + assert(Thread.interrupted()) + assert(ex.getCause.isInstanceOf[InterruptedException]) + } +}
