airbytehq / airbytehq/airbyte

[destination-databricks] WorkspaceClient built without an HTTP timeout — one hung Files API call stalls a sync for 600s

Đang mở
#85,069 1 bình luận 0 reaction 0 người được giao Xem trên GitHub
area/connectors autoteam community connectors/destination/databricks needs-triage team/extensibility type/bug
Ngôn ngữ chính
Python
Star
22.1k
Fork
5.3k
Chỉ số merge pull request
Chỉ số pull request đang chờ

Mô tả

### Connector Name

destination-databricks

### Connector Version

4.0.2

### What step the error happened?

During the sync

### Relevant information

**Summary:** `DatabricksBeanFactory.kt` sets explicit timeouts on the JDBC `DataSource` but none on the
`WorkspaceClient`. Files API calls therefore inherit the Databricks Java SDK's 300-second default socket
timeout. When a pooled connection goes dead, a single `files().delete()` / `createDirectory()` /
`upload()` blocks for ~600s and fails the whole sync — while the destination is healthy and the SQL path
keeps working in the same process, concurrently.

### The asymmetry, in your code

In `DatabricksBeanFactory.kt`, `databricksDataSource()` sets (L25-26, L75-78):

```kotlin
private const val SOCKET_TIMEOUT_SECONDS = 3600
private const val TEMPORARILY_UNAVAILABLE_RETRY_TIMEOUT_SECONDS = 300
...
props["SocketTimeout"] = SOCKET_TIMEOUT_SECONDS.toString()
// Helps the driver retries connecting to a paused/resuming warehouse
props["TemporarilyUnavailableRetryTimeout"] =
TEMPORARILY_UNAVAILABLE_RETRY_TIMEOUT_SECONDS.toString()
```

That is the hardening from #74732 — *"Add JDBC timeout configuration to prevent indefinite hangs"*,
merged for 3.3.8, filed against a production sync that *"hung for 20+ hours with no error surfaced to the
user."* It originally landed in `DatabricksConnectorClientsFactory.kt` and now lives in
`DatabricksBeanFactory.kt`.

But `workspaceClient()` — the next function in the same file (L100-117) — sets only host and auth:

```kotlin
DatabricksConfig()
.setAuthType("oauth-m2m")
.setHost("https://${config.hostname}")
.setClientId(config.authType.clientId)
.setClientSecret(config.authType.secret)
// no .setHttpTimeoutSeconds(...)
```

So the SQL path is protected against exactly this failure mode and the Files API path is not — which
matches our logs precisely: `COPY INTO` survives, `upload` / `createDirectory` / `delete` hang.

### Where the 300s comes from

`com.databricks.sdk.core.commons.CommonsHttpClient.makeDefaultRequestConfig` (databricks-sdk-java
v0.110.0, L168-181):

```java
int timeoutSeconds = 300;
if (databricksConfig != null && databricksConfig.getHttpTimeoutSeconds() != null) {
timeoutSeconds = databricksConfig.getHttpTimeoutSeconds();
}
...
.setConnectionRequestTimeout(timeout)
.setConnectTimeout(timeout)
.setSocketTimeout(timeout)
```

The observed hang is 600.2s — the 300s default, retried once by Apache HttpClient's `RetryExec`. The
thread is stuck in `DefaultHttpResponseParser.parseHead`, so the request left the client and the first
response byte never came back.

### Why this is a connector gap, not our network

- The connection was in active use **7 seconds** before the call that hung. `COPY INTO` for a staging
file succeeded; the `files().delete()` cleaning up that same file was issued 7s later and never got a
response.
- **The same destination process kept talking to the same host successfully throughout the hang** — inside
the 600s it ran five `SELECT column_name ...` queries and marked two other streams complete.
- **Other sync pods on the same node, same egress IP, hitting the same workspace, completed Files API
uploads and `COPY INTO` throughout the blackhole** — one logged 110 successful staging operations inside
that window. Neither the workspace nor the shared network path was down; one individual pooled
connection went dark.
- Our egress load balancer has TCP reset enabled, so an idle-reap would surface as `Connection reset`,
not a silent 600s wait. Raising the outbound idle timeout changed nothing.
- SNAT ports were at ~2% utilisation, so this is not port exhaustion.
- Not warehouse contention: statement latency is flat regardless of how many syncs are running, and this
also happens when it is the only sync on the instance.
- Not specific to one source or one call site: we see the same signature with unrelated sources, and on
`delete`, `createDirectory` and `upload`.

### Operators have no workaround

`DatabricksSpecification.kt` has no timeout field, and in the SDK `httpTimeoutSeconds` is declared
`@ConfigAttribute()` with no `env` (`DatabricksConfig.java:135`). `ConfigAttributeAccessor.getEnv()`
returns `""` when `env` is empty (L38-43), so no environment variable can set it either. The only lever
is connector code.

### Proposed fix

Mirrors #74732, applied to the other client:

```diff
private const val SOCKET_TIMEOUT_SECONDS = 3600
private const val TEMPORARILY_UNAVAILABLE_RETRY_TIMEOUT_SECONDS = 300
+// Bounds Files API calls (upload/createDirectory/delete). Without this they inherit the
+// SDK's 300s default, so one dead pooled connection stalls a sync for 600s+.
+private const val HTTP_TIMEOUT_SECONDS = 60

fun workspaceClient(config: DatabricksConfiguration): WorkspaceClient {
val databricksConfig =
when (config.authType) {
is PersonalAccessTokenConfiguration -> {
DatabricksConfig()
.setAuthType("pat")
.setHost("https://${config.hostname}")
.setToken(config.authType.personalAccessToken)
+ .setHttpTimeoutSeconds(HTTP_TIMEOUT_SECONDS)
}
is OAuthConfiguration -> {
DatabricksConfig()
.setAuthType("oauth-m2m")
.setHost("https://${config.hostname}")
.setClientId(config.authType.clientId)
.setClientSecret(config.authType.secret)
+ .setHttpTimeoutSeconds(HTTP_TIMEOUT_SECONDS)
}
}
return WorkspaceClient(databricksConfig)
}
```

On the value: `httpTimeoutSeconds` maps to Apache's **socket** timeout, which bounds the gap between
packets rather than total request duration. 60s therefore does not cap large staging-file uploads — an
actively transferring upload resets the clock continuously — it only kills genuinely silent connections.
Happy to use a different number, or to plumb it through the spec instead, if you'd prefer.

Separately, it may be worth asking the SDK team for `validateAfterInactivity` or a stale-connection
eviction policy on the pooled connection manager (`CommonsHttpClient` sets `setMaxTotal(100)` and
`setDefaultMaxPerRoute(20)` with no validation configured), which would stop the dead connection being
handed out at all rather than just bounding the damage. The timeout above is the minimal connector-side
fix.

### Impact

The sync fails after burning ~20 minutes on a single call. Attempt-level retry does eventually recover —
the next attempt of the same job failed identically, the one after succeeded with no code or config
change — so it presents as a sync that occasionally takes hours rather than as a hard failure, which
makes it easy to miss.

### Relevant log output

```shell
01:12:20 INFO DatabricksAirbyteClient — COPY INTO ... (succeeds)
FROM '/Volumes///_staging//.avro'
01:12:27 INFO DatabricksInsertBuffer — flush begins; files().delete() issued
for that same staging file
01:16:15 INFO [destination] Executing query: SELECT column_name ... <- succeeds
01:16:18 INFO [destination] Executing query: SELECT column_name ... <- succeeds
01:16:30 INFO [destination] Destination complete for two other streams
... still no response to the DELETE, 600s total ...
01:22:27 ERROR i.a.i.d.d.w.l.DatabricksInsertBuffer(flush):112
Failed to flush 10472 record(s) for .

com.databricks.sdk.core.DatabricksError: Read timed out
at com.databricks.sdk.core.ApiClient.executeInner(ApiClient.java:271)
at com.databricks.sdk.core.ApiClient.execute(ApiClient.java:221)
at com.databricks.sdk.core.ApiClient.execute(ApiClient.java:208)
at com.databricks.sdk.service.files.FilesImpl.delete(FilesImpl.java:51)
at com.databricks.sdk.service.files.FilesAPI.delete(FilesAPI.java:59)
at com.databricks.sdk.service.files.FilesAPI.delete(FilesAPI.java:54)
at io.airbyte.integrations.destination.databricks.client.DatabricksAirbyteClient.deleteStagedFile(DatabricksAirbyteClient.kt:212)
at io.airbyte.integrations.destination.databricks.write.load.DatabricksInsertBuffer.flush(DatabricksInsertBuffer.kt:108)
at io.airbyte.integrations.destination.databricks.dataflow.DatabricksAggregate.flush(DatabricksAggregate.kt:20)
at io.airbyte.cdk.load.dataflow.stages.FlushStage.apply(FlushStage.kt:17)
Caused by: java.net.SocketTimeoutException: Read timed out
at org.apache.http.impl.conn.DefaultHttpResponseParser.parseHead(DefaultHttpResponseParser.java:138)
at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:89)

[databricks-sdk-java-0.110.0.jar, bulk-cdk-core-load-1.0.13.jar, httpclient-4.5.14.jar]
```

### Contribute

- [x] Yes, I want to contribute

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Đánh giá

Issue này chưa được đánh giá.

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.