feat(ingest): add Zerobus Ingest subplugin for push-based ingestion into Unity Catalog Delta tables

## Summary

Add a new `io.kestra.plugin.databricks.zerobus` subpackage exposing a `WriteRecords` task that pushes data directly into a Unity Catalog Delta table via the [Databricks Zerobus Ingest](https://docs.databricks.com/aws/en/ingestion/zerobus-ingest) REST API.

Zerobus Ingest is Databricks' serverless push-based ingestion layer: no message bus, no cluster to manage, instant write to a Delta table behind a single HTTP POST. It is distinct from all existing subplugins (which are cluster/job/SQL/DBFS/CLI-oriented) and complements the `sql.Query` task by covering the *write* path.

---

## Motivation

Kestra workflows routinely produce events and micro-batches that need to land in Databricks without spinning up a cluster or a Kafka topic. Today users must either shell out to `curl` via `DatabricksCLI` or route data through an intermediate message bus.

A first-class `zerobus.WriteRecords` task lets teams ingest directly from any upstream Kestra task (HTTP triggers, file downloads, API polling, transform outputs) with retry semantics, observable metrics, and the same authentication model as the rest of the plugin.

---

## Full YAML flow example

### Example 1 — Inline records

```yaml
id: write_inline_records_to_databricks
namespace: company.team

tasks:
- id: write_records
type: io.kestra.plugin.databricks.zerobus.WriteRecords
host: "{{ secret('DATABRICKS_HOST') }}"
authentication:
token: "{{ secret('DATABRICKS_TOKEN') }}"
workspaceId: "{{ secret('DATABRICKS_WORKSPACE_ID') }}"
region: us-east-1
catalog: main
schema: events
table: user_events
records:
- userId: usr_001
event: page_view
timestamp: "2024-01-15T10:30:00Z"
- userId: usr_002
event: purchase
timestamp: "2024-01-15T10:31:00Z"
amount: 49.99
```

### Example 2 — End-to-end pipeline: HTTP API → Databricks Delta table

```yaml
id: ingest_api_events_to_databricks
namespace: company.team

tasks:
- id: fetch_events
type: io.kestra.plugin.core.http.Download
uri: https://api.example.com/events
headers:
Authorization: "Bearer {{ secret('API_TOKEN') }}"

- id: write_to_delta
type: io.kestra.plugin.databricks.zerobus.WriteRecords
host: "{{ secret('DATABRICKS_HOST') }}"
authentication:
token: "{{ secret('DATABRICKS_TOKEN') }}"
workspaceId: "{{ secret('DATABRICKS_WORKSPACE_ID') }}"
region: us-east-1
catalog: main
schema: events
table: raw_events
from: "{{ outputs.fetch_events.uri }}"
```

---

## Proposed API

**New package:** `io.kestra.plugin.databricks.zerobus`

**New class:** `WriteRecords extends AbstractTask implements RunnableTask`

### Properties

| Property | Type | Group | Description |
|---|---|---|---|
| `catalog` | `Property` | `main` | Unity Catalog catalog name |
| `schema` | `Property` | `main` | Schema (database) name |
| `table` | `Property` | `main` | Target Delta table name |
| `workspaceId` | `Property` | `connection` | Numeric workspace ID (from login URL `?o=`). Used to build the Zerobus server endpoint. |
| `region` | `Property` | `connection` | Cloud region (e.g. `us-east-1`). Combined with `workspaceId` to build the endpoint: `.zerobus..cloud.databricks.com`. |
| `records` | `Property>>` | `main` | Inline records as a list of JSON objects. Mutually exclusive with `from`. |
| `from` | `Property` | `main` | URI of a Kestra internal storage file (Ion or JSON lines). Mutually exclusive with `records`. |

Authentication reuses `AbstractTask`'s existing `authentication` block (PAT token or OAuth2 client credentials).

### Output

| Field | Type | Description |
|---|---|---|
| `recordsCount` | `long` | Number of records successfully sent |

### Metrics

- `Counter` `records.count` — records pushed per execution

---

## Technical Implementation Notes

**REST API call** (stateless, no gRPC needed for a task context):
```
POST https://.zerobus..cloud.databricks.com
/zerobus/v1/tables/../insert
Content-Type: application/json
Authorization: Bearer

[{"col1": "val1", ...}, ...]
```

- PAT token is used directly as the Bearer token. OAuth2 `client_credentials` flow mints a token at `/oidc/v1/token` (same pattern already used in `AbstractTask`).
- For the `from` input path, records should be read from Kestra internal storage (Ion → `Map` deserialization, consistent with how `sql.Query` writes output), streamed and batched for large payloads.
- Target table must be pre-created in Unity Catalog with a defined schema; the task should validate the response status and surface a clear error on 4xx/5xx.

**`package-info.java`** declaring `@PluginSubGroup`:
```java
@PluginSubGroup(
title = "Zerobus Ingest",
description = "Push data directly into Unity Catalog Delta tables via Zerobus Ingest.",
categories = { PluginSubGroup.PluginCategory.DATA }
)
package io.kestra.plugin.databricks.zerobus;
```

---

## Acceptance Criteria

- [ ] `WriteRecords` task sends a JSON batch to the Zerobus REST endpoint and returns `recordsCount`
- [ ] Works with both inline `records` and `from` (Kestra internal storage URI)
- [ ] Authentication via PAT (token) and OAuth2 client credentials
- [ ] `package-info.java` registers the subgroup with title, description, and `DATA` category
- [ ] Unit test covering serialization and property rendering
- [ ] Integration test (can be `@Disabled` behind an env-var guard) hitting a real workspace
- [ ] `@Plugin(examples = {...})` annotation with both YAML examples above

---

## Out of Scope (follow-up issues)

- gRPC SDK integration (higher throughput, streaming ACKs) — Java SDK ships JNI bindings
- OpenTelemetry / OTLP ingestion endpoint
- Trigger or polling-based reading back from Zerobus system tables

Start by reviewing AbstractTask's authentication handling and the sql.Query storage-writing pattern, then inspect the existing Databricks subpackages for task and test conventions. Implement the zerobus package, WriteRecords inputs and output, REST request, authentication paths, metrics, and package-info.java registration. Done means both input modes and authentication options are covered, serialization/property-rendering tests pass, and the optional integration test is guarded appropriately.

Written by the indexing model from the issue text.