You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2025/11/02 13:49:43 UTC
(fluss) branch release-0.8 updated (a876cf80e -> 029a39d2c)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
from a876cf80e [docs] Add labels to versions under releasing
new f83567000 [hotfix] Use literal directly instead of PARTITION_GENERATE_LEGCY_NAME to avoid NoSuchField error (#1880)
new 2c318ac5d [server] Tablet server should register lakeCatalogDynamicLoader in dynamicConfigManager (#1879)
new 7e620e3dc [hotfix][client] Fix potential metadata inconsistency by deduplicating getBucketLocation call (#1897)
new 5da6cdd3f [hotfix] fix sidebar naming conversion (#1898)
new 8775613a9 [hotfix][client] Use more efficient `EnumMap` (#1882)
new 0bcc92884 [docs] upgrade 0.7 to 0.8 about sasl authorization and default database. (#1871)
new d0012bab1 [Docs] Add upgrade notes for fluss package changes (#1869)
new 710d59122 [server] Added Retry Handler to Address Snapshotting Test Asynchrony (#1881)
new 3b01d82f0 [docs] Update docs/website images for 0.8 release (#1905)
new 8e3e758be [hotfix] Set paimon warehouse in FlinkCatalogITCase to make it consistent (#1902)
new 699bc1de1 [lake/lance] add Flink memory usage note (#1909)
new 6f1715fc8 [client] Add support for inner class and is/has methods for boolean (#1908)
new 13a16e868 remove unwanted emojis to make document standard (#1865)
new 163e3a422 [helm] Move helm charts from `docker/helm` into dedicated `helm` directory in project root (#1917)
new 46e4fabf2 [docs] Add docs for Deploying with helm charts (#1862)
new 151a49f97 [doc] Add docs for delta join support with Flink 2.1 (#1875)
new 029a39d2c [docs] Add documentation for updating cluster configs and table configs (#1870)
The 17 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../fluss/client/converter/ConverterCommons.java | 4 +-
.../apache/fluss/client/converter/PojoType.java | 110 ++++--
.../fluss/client/table/scanner/log/LogFetcher.java | 6 +-
.../fluss/client/admin/FlussAdminITCase.java | 44 ++-
.../fluss/client/converter/PojoTypeTest.java | 163 ++++++++
.../fluss/flink/catalog/FlinkCatalogITCase.java | 15 +
.../fluss/lake/paimon/utils/PaimonConversions.java | 11 +-
.../apache/fluss/server/tablet/TabletServer.java | 1 +
.../apache/fluss/server/replica/ReplicaTest.java | 24 +-
{docker/helm => helm}/.helmignore | 0
{docker/helm => helm}/Chart.yaml | 0
{docker/helm => helm}/DISCLAIMER | 0
{docker/helm => helm}/LICENSE | 0
{docker/helm => helm}/NOTICE | 0
{docker/helm => helm}/README.md | 0
{docker/helm => helm}/templates/_helpers.tpl | 0
{docker/helm => helm}/templates/configmap.yaml | 0
.../helm => helm}/templates/sts-coordinator.yaml | 0
{docker/helm => helm}/templates/sts-tablet.yaml | 0
.../helm => helm}/templates/svc-coordinator.yaml | 0
{docker/helm => helm}/templates/svc-tablet.yaml | 0
{docker/helm => helm}/values.yaml | 0
website/docs/assets/architecture.png | Bin 331654 -> 351273 bytes
website/docs/assets/delta_join.jpg | Bin 0 -> 85431 bytes
website/docs/assets/deployment_overview.png | Bin 185222 -> 200087 bytes
website/docs/assets/streamhouse.png | Bin 250947 -> 181137 bytes
website/docs/assets/tiered-storage.png | Bin 146759 -> 163033 bytes
website/docs/engine-flink/datastream.mdx | 2 +-
website/docs/engine-flink/ddl.md | 29 ++
website/docs/engine-flink/delta-joins.md | 186 ++++++++++
website/docs/engine-flink/options.md | 12 +-
website/docs/install-deploy/deploying-with-helm.md | 408 +++++++++++++++++++++
.../maintenance/operations/graceful-shutdown.md | 11 +-
.../maintenance/operations/updating-configs.md | 48 +++
.../maintenance/operations/upgrade-notes-0.8.md | 69 +++-
website/docs/maintenance/operations/upgrading.md | 2 +-
website/docs/quickstart/_category_.json | 2 +-
.../integrate-data-lakes/iceberg.md | 46 +--
.../integrate-data-lakes/lance.md | 2 +
.../integrate-data-lakes/paimon.md | 2 +-
website/src/css/custom.css | 2 -
website/static/img/fluss.png | Bin 198902 -> 219843 bytes
42 files changed, 1110 insertions(+), 89 deletions(-)
create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
rename {docker/helm => helm}/.helmignore (100%)
rename {docker/helm => helm}/Chart.yaml (100%)
rename {docker/helm => helm}/DISCLAIMER (100%)
rename {docker/helm => helm}/LICENSE (100%)
rename {docker/helm => helm}/NOTICE (100%)
rename {docker/helm => helm}/README.md (100%)
rename {docker/helm => helm}/templates/_helpers.tpl (100%)
rename {docker/helm => helm}/templates/configmap.yaml (100%)
rename {docker/helm => helm}/templates/sts-coordinator.yaml (100%)
rename {docker/helm => helm}/templates/sts-tablet.yaml (100%)
rename {docker/helm => helm}/templates/svc-coordinator.yaml (100%)
rename {docker/helm => helm}/templates/svc-tablet.yaml (100%)
rename {docker/helm => helm}/values.yaml (100%)
create mode 100644 website/docs/assets/delta_join.jpg
create mode 100644 website/docs/engine-flink/delta-joins.md
create mode 100644 website/docs/install-deploy/deploying-with-helm.md
create mode 100644 website/docs/maintenance/operations/updating-configs.md
(fluss) 08/17: [server] Added Retry Handler to Address Snapshotting Test Asynchrony (#1881)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 710d5912202cfc2124dfe12ba3bd0dc23f813a34
Author: Rion Williams <ri...@broadcom.com>
AuthorDate: Thu Oct 30 03:29:11 2025 -0500
[server] Added Retry Handler to Address Snapshotting Test Asynchrony (#1881)
[server] Added Retry Handler to Address Snapshotting Test Asynchrony
(cherry picked from commit 37f46ddd5413de94ce4407682990b38d7f5d56bd)
---
.../apache/fluss/server/replica/ReplicaTest.java | 24 +++++++++++++++++++---
1 file changed, 21 insertions(+), 3 deletions(-)
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
index bc1365ab1..51ab24347 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaTest.java
@@ -503,7 +503,7 @@ final class ReplicaTest extends ReplicaTestBase {
putRecordsToLeader(kvReplica, kvRecords);
// trigger first snapshot
- scheduledExecutorService.triggerNonPeriodicScheduledTask();
+ triggerSnapshotTaskWithRetry(scheduledExecutorService, 5);
kvSnapshotStore.waitUntilSnapshotComplete(tableBucket, 0);
// put more data and create second snapshot
@@ -514,7 +514,7 @@ final class ReplicaTest extends ReplicaTestBase {
putRecordsToLeader(kvReplica, kvRecords);
// trigger second snapshot
- scheduledExecutorService.triggerNonPeriodicScheduledTask();
+ triggerSnapshotTaskWithRetry(scheduledExecutorService, 5);
kvSnapshotStore.waitUntilSnapshotComplete(tableBucket, 1);
// put more data and create third snapshot (this will be the broken one)
@@ -525,7 +525,7 @@ final class ReplicaTest extends ReplicaTestBase {
putRecordsToLeader(kvReplica, kvRecords);
// trigger third snapshot
- scheduledExecutorService.triggerNonPeriodicScheduledTask();
+ triggerSnapshotTaskWithRetry(scheduledExecutorService, 5);
CompletedSnapshot snapshot2 = kvSnapshotStore.waitUntilSnapshotComplete(tableBucket, 2);
// verify that snapshot2 is the latest one before we break it
@@ -766,4 +766,22 @@ final class ReplicaTest extends ReplicaTestBase {
isScheduled = false;
}
}
+
+ /** A helper function with support for retries for flaky triggering operations. */
+ private static void triggerSnapshotTaskWithRetry(
+ ManuallyTriggeredScheduledExecutorService scheduledExecutorService, int maxRetries)
+ throws Exception {
+ for (int i = 0; i < maxRetries; i++) {
+ try {
+ scheduledExecutorService.triggerNonPeriodicScheduledTask();
+ return;
+ } catch (java.util.NoSuchElementException e) {
+ if (i == maxRetries - 1) {
+ throw e;
+ }
+
+ Thread.sleep(50);
+ }
+ }
+ }
}
(fluss) 07/17: [Docs] Add upgrade notes for fluss package changes (#1869)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit d0012bab12cd011245dd03555ee80fecb0bd5970
Author: CaoZhen <ca...@outlook.com>
AuthorDate: Thu Oct 30 16:26:15 2025 +0800
[Docs] Add upgrade notes for fluss package changes (#1869)
(cherry picked from commit 23058ad0388f283b33f330f407ecd1d0f5b49bcc)
---
.../maintenance/operations/upgrade-notes-0.8.md | 58 ++++++++++++++++++++++
1 file changed, 58 insertions(+)
diff --git a/website/docs/maintenance/operations/upgrade-notes-0.8.md b/website/docs/maintenance/operations/upgrade-notes-0.8.md
index a2697dd03..2f537ac5d 100644
--- a/website/docs/maintenance/operations/upgrade-notes-0.8.md
+++ b/website/docs/maintenance/operations/upgrade-notes-0.8.md
@@ -23,6 +23,64 @@ However, we **strongly recommend upgrading to Java 11 or higher** to ensure comp
๐ **If youโre using Fluss with Apache Flink**:
Please also upgrade your Flink deployment to **Java 11 or above**. All Flink versions currently supported by Fluss are fully compatible with Java 11.
+## Java client: Package and GroupId Migration
+
+๐ง **Action Required**: All Java client users must update their dependencies and import statements to continue using Fluss.
+
+### What Changed
+
+| Aspect | Fluss v0.7 | Fluss v0.8 |
+|-------------|------------------------------|----------------------------|
+| **GroupId** | `com.alibaba.fluss` | `org.apache.fluss` |
+| **Package** | `com.alibaba.fluss` | `org.apache.fluss` |
+
+**Note**: This change does not affect the Fluss client API.
+
+### Impact Assessment
+
+**All Java client users** will need to update their dependencies and import statements:
+
+1. **Build Dependencies**: Update your Maven/Gradle dependencies to use the new groupId
+2. **Import Statements**: Update all import statements from `com.alibaba.fluss.*` to `org.apache.fluss.*`
+
+### Migration Steps
+
+1. **Update Maven Dependencies**:
+ ```xml
+ <!-- Old dependency -->
+ <dependency>
+ <groupId>com.alibaba.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>0.7.x</version>
+ </dependency>
+
+ <!-- New dependency -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>0.8.0</version>
+ </dependency>
+ ```
+
+2. **Update Import Statements**:
+ ```java
+ // Old imports
+ import com.alibaba.fluss.client.Connection;
+ import com.alibaba.fluss.client.ConnectionFactory;
+
+ // New imports
+ import org.apache.fluss.client.Connection;
+ import org.apache.fluss.client.ConnectionFactory;
+ ```
+
+### Why This Change Was Made
+
+This change represents an important milestone in Fluss's journey to becoming an Apache project. The migration to `org.apache.fluss`:
+
+- Aligns with the Apache Software Foundation's naming conventions
+- Establishes a clear, independent identity for the project
+- Ensures long-term stability and governance under the Apache umbrella
+
## Metrics Updates
We have updated the report level for some metrics and also removed some metrics, this greatly reduces the metrics amount and improves the performance.
(fluss) 15/17: [docs] Add docs for Deploying with helm charts (#1862)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 46e4fabf2711e6735c16f909c33f1fe415f76e46
Author: David <13...@users.noreply.github.com>
AuthorDate: Sun Nov 2 13:01:29 2025 +0100
[docs] Add docs for Deploying with helm charts (#1862)
(cherry picked from commit 30468322fd3120fbbc7d5bd12ef88930b09b92b7)
---
website/docs/install-deploy/deploying-with-helm.md | 408 +++++++++++++++++++++
1 file changed, 408 insertions(+)
diff --git a/website/docs/install-deploy/deploying-with-helm.md b/website/docs/install-deploy/deploying-with-helm.md
new file mode 100644
index 000000000..a33134363
--- /dev/null
+++ b/website/docs/install-deploy/deploying-with-helm.md
@@ -0,0 +1,408 @@
+---
+title: "Deploying with Helm Charts"
+sidebar_position: 5
+---
+
+# Deploying with Helm Charts
+
+This page provides instructions for deploying a Fluss cluster on Kubernetes using Helm charts. The chart creates a distributed streaming storage system with CoordinatorServer and TabletServer components.
+
+## Prerequisites
+
+Before installing the Fluss Helm chart, ensure you have:
+
+- Kubernetes
+- Helm
+- **For Local Development**: Minikube and Docker (see [Local Development with Minikube](#running-fluss-locally-with-minikube))
+
+:::note
+A Fluss cluster deployment requires a running ZooKeeper ensemble. To provide flexibility in deployment and enable reuse of existing infrastructure,
+the Fluss Helm chart does not include a bundled ZooKeeper cluster. If you donโt already have a ZooKeeper running,
+the installation documentation provides instructions for deploying one using Bitnamiโs Helm chart.
+:::
+
+## Supported Versions
+
+| Component | Minimum Version | Recommended Version |
+|-----------|----------------|-------------------|
+| [Kubernetes](https://kubernetes.io) | v1.19+ | v1.25+ |
+| [Helm](https://helm.sh) | v3.8.0+ | v3.18.6+ |
+| [ZooKeeper](https://zookeeper.apache.org) | v3.6+ | v3.8+ |
+| [Apache Fluss](https://fluss.apache.org/docs/) (Container Image) | $FLUSS_VERSION$ | $FLUSS_VERSION$ |
+| [Minikube](https://minikube.sigs.k8s.io) (Local Development) | v1.25+ | v1.32+ |
+| [Docker](https://docs.docker.com/) (Local Development) | v20.10+ | v24.0+ |
+
+## Installation
+
+### Running Fluss locally with Minikube
+
+For local testing and development, you can deploy Fluss on Minikube. This is ideal for development, testing, and learning purposes.
+
+#### Prerequisites
+
+- Docker container runtime
+- At least 4GB RAM available for Minikube
+- At least 2 CPU cores available
+
+#### Start Minikube
+
+```bash
+# Start Minikube with recommended settings for Fluss
+minikube start
+
+# Verify cluster is ready
+kubectl cluster-info
+```
+
+#### Configure Docker Environment (Optional)
+
+To build images directly in Minikube you need to configure the Docker CLI to use Minikube's internal Docker daemon:
+
+```bash
+# Configure shell to use Minikube's Docker daemon
+eval $(minikube docker-env)
+```
+
+To build custom images please refer to [Custom Container Images](#custom-container-images).
+
+### Installing the chart on a cluster
+
+This installation process is generally working for a distributed Kubernetes cluster or a Minikube setup.
+
+### Step 1: Deploy ZooKeeper (Optional if ZooKeeper is existing)
+
+To start Zookeeper use Bitnamiโs chart or your own deployment. If you have an existing Zookeeper cluster, you can skip this step. Example with Bitnamiโs chart:
+
+```bash
+# Add Bitnami repository
+helm repo add bitnami https://charts.bitnami.com/bitnami
+helm repo update
+
+# Deploy ZooKeeper
+helm install zk bitnami/zookeeper
+```
+### Step 2: Deploy Fluss
+
+#### Install from Helm repo
+
+```bash
+helm repo add fluss https://downloads.apache.org/incubator/fluss/helm-chart
+helm repo update
+helm install helm-repo/fluss
+```
+
+#### Install from Local Chart
+
+```bash
+helm install fluss ./helm
+```
+
+#### Install with Custom Values
+
+You can customize the installation by providing your own `values.yaml` file or setting individual parameters via the `--set` flag. Using a custom values file:
+
+```bash
+helm install fluss ./helm -f my-values.yaml
+```
+
+Or for example to change the ZooKeeper address via the `--set` flag:
+
+```bash
+helm install fluss ./helm \
+ --set configurationOverrides.zookeeper.address=<my-zk-cluster>:2181
+```
+
+### Cleanup
+
+```bash
+# Uninstall Fluss
+helm uninstall fluss
+
+# Uninstall ZooKeeper
+helm uninstall zk
+
+# Delete PVCs
+kubectl delete pvc -l app.kubernetes.io/name=fluss
+
+# Stop Minikube
+minikube stop
+
+# Delete Minikube cluster
+minikube delete
+```
+
+## Architecture Overview
+
+The Fluss Helm chart deploys the following Kubernetes resources:
+
+### Core Components
+- **CoordinatorServer**: 1x StatefulSet with Headless Service for cluster coordination
+- **TabletServer**: 3x StatefulSet with Headless Service for data storage and processing
+- **ConfigMap**: Configuration management for `server.yaml` settings
+- **Services**: Headless services providing stable pod DNS names
+
+### Optional Components
+- **PersistentVolumes**: Data persistence when `persistence.enabled=true`
+
+
+### Step 3: Verify Installation
+
+```bash
+# Check pod status
+kubectl get pods -l app.kubernetes.io/name=fluss
+
+# Check services
+kubectl get svc -l app.kubernetes.io/name=fluss
+
+# View logs
+kubectl logs -l app.kubernetes.io/component=coordinator
+kubectl logs -l app.kubernetes.io/component=tablet
+```
+
+## Configuration Parameters
+
+The following table lists the configurable parameters of the Fluss chart and their default values.
+
+### Global Parameters
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `nameOverride` | Override the name of the chart | `""` |
+| `fullnameOverride` | Override the full name of the resources | `""` |
+
+### Image Parameters
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `image.registry` | Container image registry | `""` |
+| `image.repository` | Container image repository | `fluss` |
+| `image.tag` | Container image tag | `$FLUSS_VERSION$` |
+| `image.pullPolicy` | Container image pull policy | `IfNotPresent` |
+| `image.pullSecrets` | Container image pull secrets | `[]` |
+
+### Application Configuration
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `appConfig.internalPort` | Internal communication port | `9123` |
+| `appConfig.externalPort` | External client port | `9124` |
+
+### Fluss Configuration Overrides
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `configurationOverrides.default.bucket.number` | Default number of buckets for tables | `3` |
+| `configurationOverrides.default.replication.factor` | Default replication factor | `3` |
+| `configurationOverrides.zookeeper.path.root` | ZooKeeper root path for Fluss | `/fluss` |
+| `configurationOverrides.zookeeper.address` | ZooKeeper ensemble address | `zk-zookeeper.{{ .Release.Namespace }}.svc.cluster.local:2181` |
+| `configurationOverrides.remote.data.dir` | Remote data directory for snapshots | `/tmp/fluss/remote-data` |
+| `configurationOverrides.data.dir` | Local data directory | `/tmp/fluss/data` |
+| `configurationOverrides.internal.listener.name` | Internal listener name | `INTERNAL` |
+
+### Persistence Parameters
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `persistence.enabled` | Enable persistent volume claims | `false` |
+| `persistence.size` | Persistent volume size | `1Gi` |
+| `persistence.storageClass` | Storage class name | `nil` (uses default) |
+
+### Resource Parameters
+
+| Parameter | Description | Default |
+|-----------|-------------|---------|
+| `resources.coordinatorServer.requests.cpu` | CPU requests for coordinator | Not set |
+| `resources.coordinatorServer.requests.memory` | Memory requests for coordinator | Not set |
+| `resources.coordinatorServer.limits.cpu` | CPU limits for coordinator | Not set |
+| `resources.coordinatorServer.limits.memory` | Memory limits for coordinator | Not set |
+| `resources.tabletServer.requests.cpu` | CPU requests for tablet servers | Not set |
+| `resources.tabletServer.requests.memory` | Memory requests for tablet servers | Not set |
+| `resources.tabletServer.limits.cpu` | CPU limits for tablet servers | Not set |
+| `resources.tabletServer.limits.memory` | Memory limits for tablet servers | Not set |
+
+
+## Advanced Configuration
+
+### Custom ZooKeeper Configuration
+
+For external ZooKeeper clusters:
+
+```yaml
+configurationOverrides:
+ zookeeper.address: "zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181"
+ zookeeper.path.root: "/my-fluss-cluster"
+```
+
+### Network Configuration
+
+The chart automatically configures listeners for internal cluster communication and external client access:
+
+- **Internal Port (9123)**: Used for inter-service communication within the cluster
+- **External Port (9124)**: Used for client connections
+
+Custom listener configuration:
+
+```yaml
+appConfig:
+ internalPort: 9123
+ externalPort: 9124
+
+configurationOverrides:
+ bind.listeners: "INTERNAL://0.0.0.0:9123,CLIENT://0.0.0.0:9124"
+ advertised.listeners: "CLIENT://my-cluster.example.com:9124"
+```
+
+### Storage Configuration
+
+Configure different storage backends:
+
+```yaml
+configurationOverrides:
+ data.dir: "/data/fluss"
+ remote.data.dir: "s3://my-bucket/fluss-data"
+```
+
+## Upgrading
+
+### Upgrade the Chart
+
+```bash
+# Upgrade to a newer chart version
+helm upgrade fluss ./helm
+
+# Upgrade with new configuration
+helm upgrade fluss ./helm -f values-new.yaml
+```
+
+### Rolling Updates
+
+The StatefulSets support rolling updates. When you update the configuration, pods will be restarted one by one to maintain availability.
+
+## Custom Container Images
+
+### Building Custom Images
+
+To build and use custom Fluss images:
+
+1. Build the project with Maven:
+```bash
+mvn clean package -DskipTests
+```
+
+2. Build the Docker image:
+```bash
+# Copy build artifacts
+cp -r build-target/* docker/fluss/build-target
+
+# Build image
+cd docker
+docker build -t my-registry/fluss:custom-tag .
+```
+
+3. Use in Helm values:
+```yaml
+image:
+ registry: my-registry
+ repository: fluss
+ tag: custom-tag
+ pullPolicy: Always
+```
+
+## Monitoring and Observability
+
+### Health Checks
+
+The chart includes liveness and readiness probes:
+
+```yaml
+livenessProbe:
+ tcpSocket:
+ port: 9124
+ initialDelaySeconds: 10
+ periodSeconds: 3
+ failureThreshold: 100
+
+readinessProbe:
+ tcpSocket:
+ port: 9124
+ initialDelaySeconds: 10
+ periodSeconds: 3
+ failureThreshold: 100
+```
+
+### Logs
+
+Access logs from different components:
+
+```bash
+# Coordinator logs
+kubectl logs -l app.kubernetes.io/component=coordinator -f
+
+# Tablet server logs
+kubectl logs -l app.kubernetes.io/component=tablet -f
+
+# Specific pod logs
+kubectl logs coordinator-server-0 -f
+kubectl logs tablet-server-0 -f
+```
+
+## Troubleshooting
+
+### Common Issues
+
+#### Pod Startup Issues
+
+**Symptoms**: Pods stuck in `Pending` or `CrashLoopBackOff` state
+
+**Solutions**:
+```bash
+# Check pod events
+kubectl describe pod <pod-name>
+
+# Check resource availability
+kubectl describe nodes
+
+# Verify ZooKeeper connectivity
+kubectl exec -it <fluss-pod> -- nc -zv <zookeeper-host> 2181
+```
+
+#### Image Pull Errors
+
+**Symptoms**: `ImagePullBackOff` or `ErrImagePull`
+
+**Solutions**:
+- Verify image repository and tag exist
+- Check pull secrets configuration
+- Ensure network connectivity to registry
+
+
+#### Connection Issues
+
+**Symptoms**: Clients cannot connect to Fluss cluster
+
+**Solutions**:
+```bash
+# Check service endpoints
+kubectl get endpoints
+
+# Test network connectivity
+kubectl exec -it <client-pod> -- nc -zv <fluss-service> 9124
+
+# Verify DNS resolution
+kubectl exec -it <client-pod> -- nslookup <fluss-service>
+```
+
+### Debug Commands
+
+```bash
+# Get all resources
+kubectl get all -l app.kubernetes.io/name=fluss
+
+# Check configuration
+kubectl get configmap fluss-conf-file -o yaml
+
+
+# Get detailed pod information
+kubectl get pods -o wide -l app.kubernetes.io/name=fluss
+```
+
(fluss) 01/17: [hotfix] Use literal directly instead of PARTITION_GENERATE_LEGCY_NAME to avoid NoSuchField error (#1880)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit f835670009adde84ef9031fa6581d659d0201833
Author: yuxia Luo <lu...@alumni.sjtu.edu.cn>
AuthorDate: Wed Oct 29 16:09:53 2025 +0800
[hotfix] Use literal directly instead of PARTITION_GENERATE_LEGCY_NAME to avoid NoSuchField error (#1880)
(cherry picked from commit e51f72bb9c956f65fee34c71b2527dd72eafe8af)
---
.../org/apache/fluss/lake/paimon/utils/PaimonConversions.java | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
index 2c42c471e..1f57ba3b6 100644
--- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
+++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/PaimonConversions.java
@@ -49,6 +49,13 @@ import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.SYSTEM_COLUMNS;
/** Utils for conversion between Paimon and Fluss. */
public class PaimonConversions {
+ // use literal directly, a future paimon version will rename
+ // the variable PARTITION_GENERATE_LEGCY_NAME to PARTITION_GENERATE_LEGACY_NAME, use literal
+ // can help avoid NoSuchField error
+ // todo: after upgrade paimon version, we call fall back to use PARTITION_GENERATE_LEGACY_NAME
+ // again
+ private static final String PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY = "partition.legacy-name";
+
// for fluss config
private static final String FLUSS_CONF_PREFIX = "fluss.";
// for paimon config
@@ -60,7 +67,7 @@ public class PaimonConversions {
static {
PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.BUCKET.key());
PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.BUCKET_KEY.key());
- PAIMON_UNSETTABLE_OPTIONS.add(CoreOptions.PARTITION_GENERATE_LEGCY_NAME.key());
+ PAIMON_UNSETTABLE_OPTIONS.add(PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY);
}
public static RowKind toRowKind(ChangeType changeType) {
@@ -219,7 +226,7 @@ public class PaimonConversions {
private static void setPaimonDefaultProperties(Options options) {
// set partition.legacy-name to false, otherwise paimon will use toString for all types,
// which will cause inconsistent partition value for the same binary value
- options.set(CoreOptions.PARTITION_GENERATE_LEGCY_NAME, false);
+ options.set(PARTITION_GENERATE_LEGACY_NAME_OPTION_KEY, Boolean.FALSE.toString());
}
private static void setFlussPropertyToPaimon(String key, String value, Options options) {
(fluss) 11/17: [lake/lance] add Flink memory usage note (#1909)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 699bc1de164f9623d5ba1fcf731bdb72f6e41cf6
Author: xx789 <34...@qq.com>
AuthorDate: Fri Oct 31 13:49:00 2025 +0800
[lake/lance] add Flink memory usage note (#1909)
(cherry picked from commit 613d35436bd35b17d03fd330795cfd1a43d53931)
---
website/docs/streaming-lakehouse/integrate-data-lakes/lance.md | 2 ++
1 file changed, 2 insertions(+)
diff --git a/website/docs/streaming-lakehouse/integrate-data-lakes/lance.md b/website/docs/streaming-lakehouse/integrate-data-lakes/lance.md
index 082ef04a7..af435973e 100644
--- a/website/docs/streaming-lakehouse/integrate-data-lakes/lance.md
+++ b/website/docs/streaming-lakehouse/integrate-data-lakes/lance.md
@@ -72,6 +72,8 @@ Additionally, when following the [Start Datalake Tiering Service](maintenance/ti
> **NOTE**: Fluss v0.8 only supports tiering log tables to Lance.
+> **NOTE**: The Lance connector leverages Arrow Java library, which operates on off-heap memory. To prevent `java.lang.OutOfMemoryError: Direct buffer memory` error in Flink Task Manager, please increase the value of `taskmanager.memory.task.off-heap.size` in `<FLINK_HOME>/conf/config.yaml` to at least `'512m'` (e.g., `taskmanager.memory.task.off-heap.size: 512m`). You may need to adjust this value higher (such as `'1g'`) depending on your workload and data size.
+
Then, the datalake tiering service continuously tiers data from Fluss to Lance. The parameter `table.datalake.freshness` controls the frequency that Fluss writes data to Lance tables. By default, the data freshness is 3 minutes.
You can also specify Lance table properties when creating a datalake-enabled Fluss table by using the `lance.` prefix within the Fluss table properties clause.
(fluss) 13/17: remove unwanted emojis to make document standard (#1865)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 13a16e8682aab5fc4686eb6d0817154b2e5e64b4
Author: MehulBatra <66...@users.noreply.github.com>
AuthorDate: Fri Oct 31 13:22:02 2025 +0530
remove unwanted emojis to make document standard (#1865)
(cherry picked from commit 0b5ec3ba4b54a1889588ffd3ef1983ecd0695180)
---
.../integrate-data-lakes/iceberg.md | 46 +++++++++++-----------
.../integrate-data-lakes/paimon.md | 2 +-
2 files changed, 24 insertions(+), 24 deletions(-)
diff --git a/website/docs/streaming-lakehouse/integrate-data-lakes/iceberg.md b/website/docs/streaming-lakehouse/integrate-data-lakes/iceberg.md
index a51c967c0..4e299ebe1 100644
--- a/website/docs/streaming-lakehouse/integrate-data-lakes/iceberg.md
+++ b/website/docs/streaming-lakehouse/integrate-data-lakes/iceberg.md
@@ -13,9 +13,9 @@ To integrate Fluss with Iceberg, you must enable lakehouse storage and configure
> **NOTE**: Iceberg requires JDK11 or later. Please ensure that both your Fluss deployment and the Flink cluster used for tiering services are running on JDK11+.
-## โ๏ธ Configure Iceberg as LakeHouse Storage
+## Configure Iceberg as LakeHouse Storage
-### ๐ง Configure Iceberg in Cluster Configurations
+### Configure Iceberg in Cluster Configurations
To configure Iceberg as the lakehouse storage, you must configure the following configurations in `server.yaml`:
```yaml
@@ -27,13 +27,13 @@ datalake.iceberg.type: hadoop
datalake.iceberg.warehouse: /tmp/iceberg
```
-#### ๐ง Configuration Processing
+#### Configuration Processing
Fluss processes Iceberg configurations by stripping the `datalake.iceberg.` prefix and uses the stripped configurations (without the prefix `datalake.iceberg.`) to initialize the Iceberg catalog.
This approach enables passing custom configurations for Iceberg catalog initialization. Check out the [Iceberg Catalog Properties](https://iceberg.apache.org/docs/1.9.1/configuration/#catalog-properties) for more details on available catalog configurations.
-#### ๐ Supported Catalog Types
+#### Supported Catalog Types
Fluss supports all Iceberg-compatible catalog types:
@@ -56,7 +56,7 @@ datalake.iceberg.catalog-impl: <your_iceberg_catalog_impl_class_name>
datalake.iceberg.catalog-impl: org.apache.iceberg.snowflake.SnowflakeCatalog
```
-#### ๐ง Prerequisites
+#### Prerequisites
##### 1. Hadoop Dependencies Configuration
@@ -95,19 +95,19 @@ Fluss only bundles catalog implementations included in the `iceberg-core` module
The Iceberg version that Fluss bundles is based on `1.9.1`. Please ensure the JARs you add are compatible with `Iceberg-1.9.1`.
-#### โ ๏ธ Important Notes
+#### Important Notes
- Ensure all JAR files are compatible with Iceberg 1.9.1
- If using an existing Hadoop environment, it's recommended to use the `HADOOP_CLASSPATH` environment variable
- Configuration changes take effect after restarting the Fluss service
-### ๐ Start Tiering Service to Iceberg
+### Start Tiering Service to Iceberg
To tier Fluss's data to Iceberg, you must start the datalake tiering service. For guidance, you can refer to [Start The Datalake Tiering Service](maintenance/tiered-storage/lakehouse-storage.md#start-the-datalake-tiering-service). Although the example uses Paimon, the process is also applicable to Iceberg.
-#### ๐ง Prerequisites: Hadoop Dependencies
+#### Prerequisites: Hadoop Dependencies
-**โ ๏ธ Important**: Iceberg has a strong dependency on Hadoop. You must ensure Hadoop-related classes are available in the classpath before starting the tiering service.
+**Important**: Iceberg has a strong dependency on Hadoop. You must ensure Hadoop-related classes are available in the classpath before starting the tiering service.
##### Option 1: Use Existing Hadoop Environment (Recommended)
@@ -144,7 +144,7 @@ export HADOOP_HOME=$(pwd)/hadoop-3.3.5
export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`
```
-#### ๐ง Prepare Required JARs
+#### Prepare Required JARs
Follow the dependency management guidelines below for the [Prepare required jars](maintenance/tiered-storage/lakehouse-storage.md#prepare-required-jars) step:
@@ -176,7 +176,7 @@ iceberg-aws-bundle-1.9.1.jar
failsafe-3.3.2.jar
```
-#### ๐ Start Datalake Tiering Service
+#### Start Datalake Tiering Service
When following the [Start Datalake Tiering Service](maintenance/tiered-storage/lakehouse-storage.md#start-datalake-tiering-service) guide, use Iceberg-specific configurations as parameters when starting the Flink tiering job:
@@ -188,7 +188,7 @@ When following the [Start Datalake Tiering Service](maintenance/tiered-storage/l
--datalake.iceberg.warehouse /tmp/iceberg
```
-#### โ ๏ธ Important Notes
+#### Important Notes
- Ensure all JAR files are compatible with Iceberg 1.9.1
- Verify that all required dependencies are in the `${FLINK_HOME}/lib` directory
@@ -202,7 +202,7 @@ When a Fluss table is created or altered with the option `'table.datalake.enable
The schema of the Iceberg table matches that of the Fluss table, except for the addition of three system columns at the end: `__bucket`, `__offset`, and `__timestamp`.
These system columns help Fluss clients consume data from Iceberg in a streaming fashion, such as seeking by a specific bucket using an offset or timestamp.
-### ๐ง Basic Configuration
+### Basic Configuration
Here is an example using Flink SQL to create a table with data lake enabled:
@@ -224,7 +224,7 @@ CREATE TABLE fluss_order_with_lake (
);
```
-### โ๏ธ Iceberg Table Properties
+### Iceberg Table Properties
You can also specify Iceberg [table properties](https://iceberg.apache.org/docs/latest/configuration/#table-properties) when creating a datalake-enabled Fluss table by using the `iceberg.` prefix within the Fluss table properties clause.
@@ -249,7 +249,7 @@ CREATE TABLE fluss_order_with_lake (
);
```
-### ๐ Primary Key Tables
+### Primary Key Tables
Primary key tables in Fluss are mapped to Iceberg tables with:
@@ -289,7 +289,7 @@ CREATE TABLE user_profiles (
SORTED BY (__offset ASC);
```
-### ๐ Log Tables
+### Log Tables
The table mapping for Fluss log tables varies depending on whether the bucket key is specified or not.
@@ -360,7 +360,7 @@ CREATE TABLE order_events (
SORTED BY (__offset ASC);
```
-### ๐๏ธ Partitioned Tables
+### Partitioned Tables
For Fluss partitioned tables, Iceberg first partitions by Fluss partition keys, then follows the above rules:
@@ -394,7 +394,7 @@ CREATE TABLE daily_sales (
SORTED BY (__offset ASC);
```
-### ๐ System Columns
+### System Columns
All Iceberg tables created by Fluss include three system columns:
@@ -406,7 +406,7 @@ All Iceberg tables created by Fluss include three system columns:
## Read Tables
-### ๐ฟ๏ธ Reading with Apache Flink
+### Reading with Apache Flink
When a table has the configuration `table.datalake.enabled = 'true'`, its data exists in two layers:
@@ -444,7 +444,7 @@ Key behavior for data retention:
- **Expired Fluss log data** (controlled by `table.log.ttl`) remains accessible via Iceberg if previously tiered
- **Cleaned-up partitions** in partitioned tables (controlled by `table.auto-partition.num-retention`) remain accessible via Iceberg if previously tiered
-### ๐ Reading with Other Engines
+### Reading with Other Engines
Since data tiered to Iceberg from Fluss is stored as standard Iceberg tables, you can use any Iceberg-compatible engine. Below is an example using [StarRocks](https://docs.starrocks.io/docs/data_source/catalog/iceberg/iceberg_catalog/):
@@ -504,7 +504,7 @@ When integrating with Iceberg, Fluss automatically converts between Fluss data t
## Maintenance and Optimization
-### ๐ฆ Auto Compaction
+### Auto Compaction
The table option `table.datalake.auto-compaction` (disabled by default) provides per-table control over automatic compaction.
When enabled for a specific table, compaction is automatically triggered during write operations to that table by the tiering service.
@@ -528,7 +528,7 @@ CREATE TABLE example_table (
- **Storage**: Optimizes storage usage by removing duplicate data
- **Maintenance**: Automatically handles data organization
-### ๐ Snapshot Metadata
+### Snapshot Metadata
Fluss adds specific metadata to Iceberg snapshots for traceability:
@@ -578,7 +578,7 @@ For partitioned tables, the metadata structure includes partition information:
| `offset` | Offset within the partition's log | `3`, `1000` |
-## ๐ซ Current Limitations
+## Current Limitations
- **Complex Types**: Array, Map, and Row types are not supported
- **Multiple bucket keys**: Not supported until Iceberg implements multi-argument partition transforms
diff --git a/website/docs/streaming-lakehouse/integrate-data-lakes/paimon.md b/website/docs/streaming-lakehouse/integrate-data-lakes/paimon.md
index a27d534af..6e1462435 100644
--- a/website/docs/streaming-lakehouse/integrate-data-lakes/paimon.md
+++ b/website/docs/streaming-lakehouse/integrate-data-lakes/paimon.md
@@ -176,7 +176,7 @@ The following table shows the mapping between [Fluss data types](table-design/da
| BINARY | BINARY |
| BYTES | BYTES |
-## ๐ Snapshot Metadata
+## Snapshot Metadata
Fluss adds specific metadata to Paimon snapshots for traceability:
(fluss) 02/17: [server] Tablet server should register lakeCatalogDynamicLoader in dynamicConfigManager (#1879)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 2c318ac5d513ebe235621bb6dffaf86e93daed73
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Thu Oct 30 14:14:27 2025 +0800
[server] Tablet server should register lakeCatalogDynamicLoader in dynamicConfigManager (#1879)
(cherry picked from commit dbf6f61d77ff38a069785a00da16c9c907d8f641)
---
.../fluss/client/admin/FlussAdminITCase.java | 44 ++++++++++++++++------
.../apache/fluss/server/tablet/TabletServer.java | 1 +
2 files changed, 33 insertions(+), 12 deletions(-)
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 860d63f36..8e2b6a329 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -86,9 +86,12 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.fluss.config.ConfigOptions.DATALAKE_FORMAT;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
+import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_FORMAT;
import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.apache.fluss.testutils.common.CommonTestUtils.waitUntil;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -1070,7 +1073,7 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
}
@Test
- void testDynamicConfigs() throws ExecutionException, InterruptedException {
+ void testDynamicConfigs() throws Exception {
assertThat(
FLUSS_CLUSTER_EXTENSION
.getCoordinatorServer()
@@ -1092,20 +1095,37 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
assertConfigEntry(
DATALAKE_FORMAT.key(), null, ConfigEntry.ConfigSource.DYNAMIC_SERVER_CONFIG);
- // Delete dynamic configs to use the initial value(from server.yaml)
admin.alterClusterConfigs(
- Collections.singletonList(
+ Arrays.asList(
+ new AlterConfig(
+ DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.SET),
new AlterConfig(
- DATALAKE_FORMAT.key(), null, AlterConfigOpType.DELETE)))
+ "datalake.paimon.warehouse",
+ "test-warehouse",
+ AlterConfigOpType.SET)))
.get();
- assertThat(
- FLUSS_CLUSTER_EXTENSION
- .getCoordinatorServer()
- .getCoordinatorService()
- .getDataLakeFormat())
- .isEqualTo(PAIMON);
- assertConfigEntry(
- DATALAKE_FORMAT.key(), "paimon", ConfigEntry.ConfigSource.INITIAL_SERVER_CONFIG);
+ TablePath tablePath = TablePath.of("test_db", "test_table");
+ createTable(
+ tablePath,
+ TableDescriptor.builder()
+ .schema(DEFAULT_SCHEMA)
+ .property(TABLE_DATALAKE_ENABLED, true)
+ .build(),
+ true);
+
+ waitUntil(
+ () -> {
+ TableInfo tableInfo = admin.getTableInfo(tablePath).get();
+ Map<String, String> tableProperties = tableInfo.getProperties().toMap();
+ return tableProperties.containsKey(TABLE_DATALAKE_FORMAT.key())
+ && tableProperties.containsKey("table.datalake.paimon.warehouse")
+ && PAIMON.toString()
+ .equals(tableProperties.get(TABLE_DATALAKE_FORMAT.key()))
+ && "test-warehouse"
+ .equals(tableProperties.get("table.datalake.paimon.warehouse"));
+ },
+ Duration.ofMinutes(1),
+ "Get lakehouse info");
}
private void assertConfigEntry(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index 7c02f993c..eef925b50 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -203,6 +203,7 @@ public class TabletServer extends ServerBase {
MetadataManager metadataManager =
new MetadataManager(zkClient, conf, lakeCatalogDynamicLoader);
this.dynamicConfigManager = new DynamicConfigManager(zkClient, conf, false);
+ dynamicConfigManager.register(lakeCatalogDynamicLoader);
dynamicConfigManager.startup();
this.metadataCache = new TabletServerMetadataCache(metadataManager);
(fluss) 09/17: [docs] Update docs/website images for 0.8 release (#1905)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 3b01d82f0cb847980536233a45b1ee9883c19dda
Author: Michael Koepf <47...@users.noreply.github.com>
AuthorDate: Thu Oct 30 13:45:27 2025 +0000
[docs] Update docs/website images for 0.8 release (#1905)
* Update images for 0.8 release
* Add Lance to tiered storage
(cherry picked from commit 2f422b3ce365037fde0527d621dc75f0edb72736)
---
website/docs/assets/architecture.png | Bin 331654 -> 351273 bytes
website/docs/assets/deployment_overview.png | Bin 185222 -> 200087 bytes
website/docs/assets/streamhouse.png | Bin 250947 -> 181137 bytes
website/docs/assets/tiered-storage.png | Bin 146759 -> 163033 bytes
website/static/img/fluss.png | Bin 198902 -> 219843 bytes
5 files changed, 0 insertions(+), 0 deletions(-)
diff --git a/website/docs/assets/architecture.png b/website/docs/assets/architecture.png
index d6cb9ee71..cd80e8502 100644
Binary files a/website/docs/assets/architecture.png and b/website/docs/assets/architecture.png differ
diff --git a/website/docs/assets/deployment_overview.png b/website/docs/assets/deployment_overview.png
index 928b511e5..380e34cf2 100644
Binary files a/website/docs/assets/deployment_overview.png and b/website/docs/assets/deployment_overview.png differ
diff --git a/website/docs/assets/streamhouse.png b/website/docs/assets/streamhouse.png
index a59ec390e..3b6339de4 100644
Binary files a/website/docs/assets/streamhouse.png and b/website/docs/assets/streamhouse.png differ
diff --git a/website/docs/assets/tiered-storage.png b/website/docs/assets/tiered-storage.png
index 9ca1e1d00..2d9f43749 100644
Binary files a/website/docs/assets/tiered-storage.png and b/website/docs/assets/tiered-storage.png differ
diff --git a/website/static/img/fluss.png b/website/static/img/fluss.png
index 2cf5f9ce3..7873e416c 100644
Binary files a/website/static/img/fluss.png and b/website/static/img/fluss.png differ
(fluss) 16/17: [doc] Add docs for delta join support with Flink 2.1 (#1875)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 151a49f9799456625acd35ec43c1c05308a838c0
Author: Xuyang <xy...@163.com>
AuthorDate: Sun Nov 2 21:01:06 2025 +0800
[doc] Add docs for delta join support with Flink 2.1 (#1875)
(cherry picked from commit aa4afe84af58ba0f00b7fc000f6864def7c62d30)
---
website/docs/assets/delta_join.jpg | Bin 0 -> 85431 bytes
website/docs/engine-flink/datastream.mdx | 2 +-
website/docs/engine-flink/delta-joins.md | 186 +++++++++++++++++++++++++++++++
website/docs/engine-flink/options.md | 2 +-
4 files changed, 188 insertions(+), 2 deletions(-)
diff --git a/website/docs/assets/delta_join.jpg b/website/docs/assets/delta_join.jpg
new file mode 100644
index 000000000..baa460880
Binary files /dev/null and b/website/docs/assets/delta_join.jpg differ
diff --git a/website/docs/engine-flink/datastream.mdx b/website/docs/engine-flink/datastream.mdx
index 0ef721a9d..c9aaa3798 100644
--- a/website/docs/engine-flink/datastream.mdx
+++ b/website/docs/engine-flink/datastream.mdx
@@ -1,6 +1,6 @@
---
title: "DataStream API"
-sidebar_position: 6
+sidebar_position: 7
---
# DataStream API
diff --git a/website/docs/engine-flink/delta-joins.md b/website/docs/engine-flink/delta-joins.md
new file mode 100644
index 000000000..8304ac39e
--- /dev/null
+++ b/website/docs/engine-flink/delta-joins.md
@@ -0,0 +1,186 @@
+---
+sidebar_label: Delta Joins
+title: Flink Delta Joins
+sidebar_position: 6
+---
+
+# The Delta Join
+Beginning with **Apache Flink 2.1**, a new operator called [Delta Join](https://cwiki.apache.org/confluence/display/FLINK/FLIP-486%3A+Introduce+A+New+DeltaJoin) was introduced.
+Compared to traditional streaming joins, the delta join operator significantly reduces the amount of state that needs to be maintained during execution. This improvement helps mitigate several common issues associated with large state sizes, including:
+
+- Excessive computing resource and storage consumption
+- Long checkpointing durations
+- Extended recovery times after failures or restarts
+- Heavy state bootstrap overhead after changing job logic
+
+Starting with **Apache Fluss 0.8**, streaming join jobs running on **Flink 2.1 or later** will be automatically optimized into **delta joins** whenever applicable. This optimization happens transparently at query planning time, requiring no manual configuration.
+
+## How Delta Join Works
+
+Traditional streaming joins in Flink require maintaining both input sides entirely in state to match records across streams. Delta join, by contrast, uses a **index-key lookup mechanism** to transform the behavior of querying data from the state into querying data from the Fluss source table, thereby avoiding redundant storage of the same data in both the Fluss source table and the state. This drastically reduces state size and improves performance for many streaming analytics and enrich [...]
+
+
+
+## Example: Delta Join in Flink 2.1
+
+Below is an example demonstrating a delta join query supported by Flink 2.1.
+
+#### Create Source and Sink Tables
+
+```sql title="Flink SQL"
+USE CATALOG fluss_catalog;
+```
+
+```sql title="Flink SQL"
+CREATE DATABASE my_db;
+```
+
+```sql title="Flink SQL"
+USE my_db;
+```
+
+#### Create Left Source Table
+```sql title="Flink SQL"
+CREATE TABLE `fluss_catalog`.`my_db`.`left_src` (
+ `city_id` INT NOT NULL,
+ `order_id` INT NOT NULL,
+ `content` VARCHAR NOT NULL,
+ PRIMARY KEY (city_id, order_id) NOT ENFORCED
+) WITH (
+ -- prefix key
+ 'bucket.key' = 'city_id',
+ -- in Flink 2.1, delta join only support append-only source
+ 'table.merge-engine' = 'first_row'
+);
+```
+
+#### Create Right Source Table
+```sql title="Flink SQL"
+CREATE TABLE `fluss_catalog`.`my_db`.`right_src` (
+ `city_id` INT NOT NULL,
+ `city_name` VARCHAR NOT NULL,
+ PRIMARY KEY (city_id) NOT ENFORCED
+) WITH (
+ -- in Flink 2.1, delta join only support append-only source
+ 'table.merge-engine' = 'first_row'
+);
+```
+
+#### Create Sink Table
+```sql title="Flink SQL"
+CREATE TABLE `fluss_catalog`.`my_db`.`snk` (
+ `city_id` INT NOT NULL,
+ `order_id` INT NOT NULL,
+ `content` VARCHAR NOT NULL,
+ `city_name` VARCHAR NOT NULL,
+ PRIMARY KEY (city_id, order_id) NOT ENFORCED
+);
+```
+
+#### Explain the Join Query
+```sql title="Flink SQL"
+EXPLAIN
+INSERT INTO `fluss_catalog`.`my_db`.`snk`
+SELECT T1.`city_id`, T1.`order_id`, T1.`content`, T2.`city_name`
+FROM `fluss_catalog`.`my_db`.`left_src` T1
+Join `fluss_catalog`.`my_db`.`right_src` T2
+ON T1.`city_id` = T2.`city_id`;
+```
+
+If the printed plan includes `DeltaJoin` as shown below, it indicates that the optimizer has successfully transformed the traditional streaming join into a delta join.
+
+```title="Flink Plan"
+== Abstract Syntax Tree ==
+LogicalSink(table=[fluss_catalog.my_db.snk], fields=[city_id, order_id, content, city_name])
++- LogicalProject(city_id=[$0], order_id=[$1], content=[$2], city_name=[$4])
+ +- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalTableScan(table=[[fluss_catalog, my_db, left_src]])
+ +- LogicalTableScan(table=[[fluss_catalog, my_db, right_src]])
+
+== Optimized Physical Plan ==
+Sink(table=[fluss_catalog.my_db.snk], fields=[city_id, order_id, content, city_name])
++- Calc(select=[city_id, order_id, content, city_name])
+ +- DeltaJoin(joinType=[InnerJoin], where=[=(city_id, city_id0)], select=[city_id, order_id, content, city_id0, city_name])
+ :- Exchange(distribution=[hash[city_id]])
+ : +- TableSourceScan(table=[[fluss_catalog, my_db, left_src]], fields=[city_id, order_id, content])
+ +- Exchange(distribution=[hash[city_id]])
+ +- TableSourceScan(table=[[fluss_catalog, my_db, right_src]], fields=[city_id, city_name])
+
+== Optimized Execution Plan ==
+Sink(table=[fluss_catalog.my_db.snk], fields=[city_id, order_id, content, city_name])
++- Calc(select=[city_id, order_id, content, city_name])
+ +- DeltaJoin(joinType=[InnerJoin], where=[(city_id = city_id0)], select=[city_id, order_id, content, city_id0, city_name])
+ :- Exchange(distribution=[hash[city_id]])
+ : +- TableSourceScan(table=[[fluss_catalog, my_db, left_src]], fields=[city_id, order_id, content])
+ +- Exchange(distribution=[hash[city_id]])
+ +- TableSourceScan(table=[[fluss_catalog, my_db, right_src]], fields=[city_id, city_name])
+```
+
+## Understanding Index Keys
+
+Delta Join relies on performing lookups by the join key (e.g., the `city_id` in the above example) on the Fluss source tables. This requires that the Fluss source tables are defined with appropriate index on the join key to enable efficient lookups.
+
+Currently, Fluss only supports to defines a single index key per table, which is also referred to as the **prefix key**. The general secondary index which allows define multiple indexes with arbitrary fields is planned to be supported in the near future releases.
+
+A prefix key defines the portion of a tableโs primary key that can be used for efficient key-based lookups or index pruning.
+
+In Fluss, the option `'bucket.key' = 'city_id'` specifies that data is organized (or bucketed) by `city_id`. When performing a delta join, this allows Flink to quickly locate and read only the subset of records corresponding to the specific prefix key value, rather than scanning or caching the entire table state.
+
+For example:
+- Full primary key: `(city_id, order_id)`
+- Bucket key: `city_id`
+
+This yields an **index** on the prefix key `city_id`, so that you can perform [Prefix Key Lookup](/docs/engine-flink/lookups/#prefix-lookup) by the `city_id`.
+
+In this setup:
+* The delta join operator uses the prefix key (`city_id`) to retrieve only relevant right-side records matching each left-side event.
+* This eliminates the need to hold all records for every city in Flink state, significantly reducing state size.
+
+Prefix keys thus form the foundation for efficient lookups in delta joins, enabling Flink to scale join workloads efficiently even under high throughput.
+
+## Flink Version Support
+
+The delta join feature is introduced since Flink 2.1 and still evolving, and its optimization capabilities vary across Flink versions.
+
+Refer to the [Delta Join Issue](https://issues.apache.org/jira/browse/FLINK-37836) for the most up-to-date information.
+
+
+### Flink 2.1
+
+#### Supported Features
+
+- Support for optimizing a dual-stream join in the straightforward pattern of `insert only source -> join -> upsert sink` into a delta join.
+
+#### Limitations
+
+- The primary key or the prefix key of the tables must be included as part of the equivalence conditions in the join.
+- The join must be a INNER join.
+- The downstream nodes of the join can accept duplicate changes, such as a sink that provides UPSERT mode without `upsertMaterialize`.
+- All join inputs should be INSERT-ONLY streams.
+ - This is why the option `'table.merge-engine' = 'first_row'` is added to the source table DDL.
+- All upstream nodes of the join should be `TableSourceScan` or `Exchange`.
+
+### Flink 2.2 (upcoming)
+
+#### Supported Features
+
+- Support for optimizing a dual-stream join from CDC sources that do not include delete messages into a delta join.
+ - Disable delete on the source table to guarantee there is no delete message in the table, by adding the option `'table.delete.behavior' = 'IGNORE'` or `'DISABLE'` on the table.
+ - The source table is no more required to be a `first_row` merge engine table since this version.
+- Support `Project` and `Filter` between source and delta join.
+- Support cache in delta join.
+
+#### Limitations
+
+- The primary key or the prefix lookup key of the tables must be included as part of the equivalence conditions in the join.
+- The join must be a INNER join.
+- The downstream nodes of the join can accept duplicate changes, such as a sink that provides UPSERT mode.
+- When consuming a CDC stream, the join key used in the delta join must be part of the primary key.
+- All filters must be applied on the upsert key, and neither filters nor projections should contain non-deterministic functions.
+
+## Future Plan
+
+The Fluss and Flink community are actively working together on enhancing delta join. Planned improvements include:
+- Support for additional join types, such as LEFT JOIN and FULL OUTER JOIN.
+- Support for multi-way joins involving more than two streams. This also requires Fluss to support defining multiple secondary index keys on a single table.
+- Support for more complex query patterns to optimize a wider range of join scenarios.
\ No newline at end of file
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
index 9c71e7077..3cf7490e4 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -1,6 +1,6 @@
---
title: Connector Options
-sidebar_position: 7
+sidebar_position: 8
---
# Connector Options
(fluss) 10/17: [hotfix] Set paimon warehouse in FlinkCatalogITCase to make it consistent (#1902)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 8e3e758be0314143c4a9523aed71b48383023e6f
Author: yuxia Luo <lu...@alumni.sjtu.edu.cn>
AuthorDate: Thu Oct 30 22:59:57 2025 +0800
[hotfix] Set paimon warehouse in FlinkCatalogITCase to make it consistent (#1902)
(cherry picked from commit fcd2378014377341baa0755e59d0a45591cc7bcc)
---
.../apache/fluss/flink/catalog/FlinkCatalogITCase.java | 15 +++++++++++++++
1 file changed, 15 insertions(+)
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
index 062c6eea7..88e16e996 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java
@@ -20,6 +20,7 @@ package org.apache.fluss.flink.catalog;
import org.apache.fluss.cluster.ServerNode;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.exception.InvalidAlterTableException;
import org.apache.fluss.exception.InvalidConfigException;
import org.apache.fluss.exception.InvalidTableException;
@@ -47,6 +48,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import java.nio.file.Files;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -76,11 +78,23 @@ abstract class FlinkCatalogITCase {
.setClusterConf(initClusterConf())
.build();
+ protected static String paimonWarehousePath;
+
static Configuration initClusterConf() {
Configuration clusterConf = new Configuration();
// use a small check interval to cleanup partitions quickly
clusterConf.set(ConfigOptions.AUTO_PARTITION_CHECK_INTERVAL, Duration.ofSeconds(3));
clusterConf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON);
+ try {
+ paimonWarehousePath =
+ Files.createTempDirectory("fluss-catalog-itcase")
+ .resolve("warehouse")
+ .toString();
+ } catch (Exception e) {
+ throw new FlussRuntimeException("Failed to create warehouse path");
+ }
+ clusterConf.setString("datalake.paimon.warehouse", paimonWarehousePath);
+
return clusterConf;
}
@@ -208,6 +222,7 @@ abstract class FlinkCatalogITCase {
expectedOptions.put("bucket.num", "5");
expectedOptions.put("table.datalake.enabled", "true");
expectedOptions.put("table.datalake.format", "paimon");
+ expectedOptions.put("table.datalake.paimon.warehouse", paimonWarehousePath);
assertOptionsEqual(table.getOptions(), expectedOptions);
// alter table
(fluss) 03/17: [hotfix][client] Fix potential metadata inconsistency by deduplicating getBucketLocation call (#1897)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 7e620e3dc58f482c22ced3d548af088927c6b9d1
Author: yunhong <33...@qq.com>
AuthorDate: Thu Oct 30 15:25:59 2025 +0800
[hotfix][client] Fix potential metadata inconsistency by deduplicating getBucketLocation call (#1897)
(cherry picked from commit 31a9352bd091353c8d3b06f6523e5af382eba8fa)
---
.../java/org/apache/fluss/client/table/scanner/log/LogFetcher.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
index ff67575e2..f2df66aaa 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
@@ -65,6 +65,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
@@ -502,8 +503,9 @@ public class LogFetcher implements Closeable {
}
private Integer getTableBucketLeader(TableBucket tableBucket) {
- if (metadataUpdater.getBucketLocation(tableBucket).isPresent()) {
- BucketLocation bucketLocation = metadataUpdater.getBucketLocation(tableBucket).get();
+ Optional<BucketLocation> bucketLocationOpt = metadataUpdater.getBucketLocation(tableBucket);
+ if (bucketLocationOpt.isPresent()) {
+ BucketLocation bucketLocation = bucketLocationOpt.get();
if (bucketLocation.getLeader() != null) {
return bucketLocation.getLeader();
}
(fluss) 14/17: [helm] Move helm charts from `docker/helm` into dedicated `helm` directory in project root (#1917)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 163e3a422bd31cd9ecf1e8e3563df2d7e1f1a17f
Author: David <13...@users.noreply.github.com>
AuthorDate: Fri Oct 31 12:17:34 2025 +0100
[helm] Move helm charts from `docker/helm` into dedicated `helm` directory in project root (#1917)
(cherry picked from commit 84b06ac6ef18ea629106e77755bee4b1963edd55)
---
{docker/helm => helm}/.helmignore | 0
{docker/helm => helm}/Chart.yaml | 0
{docker/helm => helm}/DISCLAIMER | 0
{docker/helm => helm}/LICENSE | 0
{docker/helm => helm}/NOTICE | 0
{docker/helm => helm}/README.md | 0
{docker/helm => helm}/templates/_helpers.tpl | 0
{docker/helm => helm}/templates/configmap.yaml | 0
{docker/helm => helm}/templates/sts-coordinator.yaml | 0
{docker/helm => helm}/templates/sts-tablet.yaml | 0
{docker/helm => helm}/templates/svc-coordinator.yaml | 0
{docker/helm => helm}/templates/svc-tablet.yaml | 0
{docker/helm => helm}/values.yaml | 0
13 files changed, 0 insertions(+), 0 deletions(-)
diff --git a/docker/helm/.helmignore b/helm/.helmignore
similarity index 100%
rename from docker/helm/.helmignore
rename to helm/.helmignore
diff --git a/docker/helm/Chart.yaml b/helm/Chart.yaml
similarity index 100%
rename from docker/helm/Chart.yaml
rename to helm/Chart.yaml
diff --git a/docker/helm/DISCLAIMER b/helm/DISCLAIMER
similarity index 100%
rename from docker/helm/DISCLAIMER
rename to helm/DISCLAIMER
diff --git a/docker/helm/LICENSE b/helm/LICENSE
similarity index 100%
rename from docker/helm/LICENSE
rename to helm/LICENSE
diff --git a/docker/helm/NOTICE b/helm/NOTICE
similarity index 100%
rename from docker/helm/NOTICE
rename to helm/NOTICE
diff --git a/docker/helm/README.md b/helm/README.md
similarity index 100%
rename from docker/helm/README.md
rename to helm/README.md
diff --git a/docker/helm/templates/_helpers.tpl b/helm/templates/_helpers.tpl
similarity index 100%
rename from docker/helm/templates/_helpers.tpl
rename to helm/templates/_helpers.tpl
diff --git a/docker/helm/templates/configmap.yaml b/helm/templates/configmap.yaml
similarity index 100%
rename from docker/helm/templates/configmap.yaml
rename to helm/templates/configmap.yaml
diff --git a/docker/helm/templates/sts-coordinator.yaml b/helm/templates/sts-coordinator.yaml
similarity index 100%
rename from docker/helm/templates/sts-coordinator.yaml
rename to helm/templates/sts-coordinator.yaml
diff --git a/docker/helm/templates/sts-tablet.yaml b/helm/templates/sts-tablet.yaml
similarity index 100%
rename from docker/helm/templates/sts-tablet.yaml
rename to helm/templates/sts-tablet.yaml
diff --git a/docker/helm/templates/svc-coordinator.yaml b/helm/templates/svc-coordinator.yaml
similarity index 100%
rename from docker/helm/templates/svc-coordinator.yaml
rename to helm/templates/svc-coordinator.yaml
diff --git a/docker/helm/templates/svc-tablet.yaml b/helm/templates/svc-tablet.yaml
similarity index 100%
rename from docker/helm/templates/svc-tablet.yaml
rename to helm/templates/svc-tablet.yaml
diff --git a/docker/helm/values.yaml b/helm/values.yaml
similarity index 100%
rename from docker/helm/values.yaml
rename to helm/values.yaml
(fluss) 17/17: [docs] Add documentation for updating cluster configs and table configs (#1870)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 029a39d2ce5175f1c4a61da349d461723e7af12a
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Sun Nov 2 21:14:39 2025 +0800
[docs] Add documentation for updating cluster configs and table configs (#1870)
---
website/docs/engine-flink/ddl.md | 29 +++++++++++++
website/docs/engine-flink/options.md | 10 +++--
.../maintenance/operations/graceful-shutdown.md | 11 +++--
.../maintenance/operations/updating-configs.md | 48 ++++++++++++++++++++++
.../maintenance/operations/upgrade-notes-0.8.md | 2 +-
website/docs/maintenance/operations/upgrading.md | 2 +-
website/src/css/custom.css | 2 -
7 files changed, 92 insertions(+), 12 deletions(-)
diff --git a/website/docs/engine-flink/ddl.md b/website/docs/engine-flink/ddl.md
index dd1788e4c..a542db143 100644
--- a/website/docs/engine-flink/ddl.md
+++ b/website/docs/engine-flink/ddl.md
@@ -220,6 +220,35 @@ DROP TABLE my_table;
This will entirely remove all the data of the table in the Fluss cluster.
+## Alter Table
+### SET properties
+The SET statement allows users to configure one or more connector options including the [Storage Options](engine-flink/options.md#storage-options) for a specified table. If a particular option is already configured on the table, it will be overridden with the new value.
+
+When using SET to modify [Storage Options](engine-flink/options.md#storage-options), the Fluss cluster will dynamically apply the changes to the table. This can be useful for modifying table behavior without needing to recreate the table.
+
+**Supported Options to modify**
+- All [Read Options](engine-flink/options.md#read-options), [Write Options](engine-flink/options.md#write-options), [Lookup Options](engine-flink/options.md#lookup-options) and [Other Options](engine-flink/options.md#other-options) except `bootstrap.servers`.
+- The following [Storage Options](engine-flink/options.md#storage-options):
+ - `table.datalake.enabled`: Enable or disable lakehouse storage for the table.
+
+```sql title="Flink SQL"
+ALTER TABLE my_table SET ('table.datalake.enabled' = 'true');
+```
+
+**Limits**
+- If lakehouse storage (`table.datalake.enabled`) is already enabled for a table, options with lakehouse format prefixes (e.g., `paimon.*`) cannot be modified again.
+
+
+### RESET properties
+Reset one or more connector option including the [Storage Options](engine-flink/options.md#storage-options) to its default value.
+
+The following example illustrates reset the `table.datalake.enabled` option to its default value `false` on the table.
+
+```sql title="Flink SQL"
+ALTER TABLE my_table RESET ('table.datalake.enabled');
+```
+
+
## Add Partition
Fluss supports manually adding partitions to an existing partitioned table through the Fluss Catalog. If the specified partition
diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md
index 3cf7490e4..caddc0f58 100644
--- a/website/docs/engine-flink/options.md
+++ b/website/docs/engine-flink/options.md
@@ -49,15 +49,16 @@ INSERT INTO pk_table2 /*+ OPTIONS('sink.ignore-delete'='true') */ select * from
```
-### Configure options by altering table
+### Reconfigure options by altering table
-This is not supported yet, but is planned in the near future.
-For example, the following SQL statement alters the Fluss table with the `table.log.ttl` set to 7 days:
+Using `ALTER TABLE ... SET` statement to modify the table options. For example, to enable lakehouse storage for an existing table, you can run the following SQL command:
```sql
-ALTER TABLE log_table SET ('table.log.ttl' = '7d');
+ALTER TABLE log_table SET ('table.datalake.enable' = 'true');
```
+See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) and [ALTER TABLE ... RESET](engine-flink/ddl.md#reset-properties) documentation.
+
## Storage Options
| Option | Type | Default | Description [...]
@@ -85,6 +86,7 @@ ALTER TABLE log_table SET ('table.log.ttl' = '7d');
| table.merge-engine.versioned.ver-column | String | (None) | The column name of the version column for the `versioned` merge engine. If the merge engine is set to `versioned`, the version column must be set. [...]
| table.delete.behavior | Enum | ALLOW | Controls the behavior of delete operations on primary key tables. Three modes are supported: `ALLOW` (default) - allows normal delete operations; `IGNORE` - silently ignores delete requests without errors; `DISABLE` - rejects delete requests and throws explicit errors. This configuration provides system-level guarantees for some downstream pipelines (e.g., Flink Delta Join) that must not receive [...]
+
## Read Options
| Option | Type | Default | Description [...]
diff --git a/website/docs/maintenance/operations/graceful-shutdown.md b/website/docs/maintenance/operations/graceful-shutdown.md
index cb20bcae2..feb4120cb 100644
--- a/website/docs/maintenance/operations/graceful-shutdown.md
+++ b/website/docs/maintenance/operations/graceful-shutdown.md
@@ -1,3 +1,8 @@
+---
+title: Graceful Shutdown
+sidebar_position: 3
+---
+
# Graceful Shutdown
Apache Fluss provides a **comprehensive graceful shutdown mechanism** to ensure data integrity and proper resource cleanup when stopping servers or services.
@@ -78,10 +83,8 @@ The controlled shutdown process can be configured using the following options:
```yaml
# server.yaml
-tablet-server:
- controlled-shutdown:
- max-retries: 5
- retry-interval: 2000ms
+tablet-server.controlled-shutdown.max-retries: 5
+tablet-server.controlled-shutdown.retry-interval: 2000ms
```
## Monitoring Shutdown
diff --git a/website/docs/maintenance/operations/updating-configs.md b/website/docs/maintenance/operations/updating-configs.md
new file mode 100644
index 000000000..0643a46fe
--- /dev/null
+++ b/website/docs/maintenance/operations/updating-configs.md
@@ -0,0 +1,48 @@
+---
+title: Updating Configs
+sidebar_position: 1
+---
+# Updating Configs
+
+## Overview
+
+Fluss allows you to update cluster or table configurations dynamically without requiring a cluster restart or table recreation. This section demonstrates how to modify and apply such configurations.
+
+## Updating Cluster Configs
+
+From Fluss version 0.8 onwards, some of the server configs can be updated without restarting the server.
+
+Currently, the supported dynamically updatable server configurations include:
+- `datalake.format`: Enable lakehouse storage by specifying the lakehouse format, e.g., `paimon`, `iceberg`.
+- Options with prefix `datalake.${datalake.format}`
+
+
+You can update the configuration of a cluster with [Java client](apis/java-client.md).
+Here is a code snippet to demonstrate how to update the cluster configurations using the Java Client:
+
+```java
+// Enable lakehouse storage with Paimon format
+admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.SET)));
+
+// Disable lakehouse storage
+admin.alterClusterConfigs(
+ Collections.singletonList(
+ new AlterConfig(DATALAKE_FORMAT.key(), "paimon", AlterConfigOpType.DELETE)));
+```
+
+The `AlterConfig` class contains three properties:
+* `key`: The configuration key to be modified (e.g., `datalake.format`)
+* `value`: The configuration value to be set (e.g., `paimon`)
+* `opType`: The operation type, either `AlterConfigOpType.SET` or `AlterConfigOpType.DELETE`
+
+
+## Updating Table Configs
+
+The connector options on a table including [Storage Options](engine-flink/options.md#storage-options) can be updated dynamically by [ALTER TABLE ... SET](engine-flink/ddl.md#alter-table) statement. See the example below:
+
+```sql
+-- Enable lakehouse storage for the given table
+ALTER TABLE my_table SET ('table.datalake.enabled' = 'true');
+```
\ No newline at end of file
diff --git a/website/docs/maintenance/operations/upgrade-notes-0.8.md b/website/docs/maintenance/operations/upgrade-notes-0.8.md
index 2f537ac5d..eec15be8d 100644
--- a/website/docs/maintenance/operations/upgrade-notes-0.8.md
+++ b/website/docs/maintenance/operations/upgrade-notes-0.8.md
@@ -1,6 +1,6 @@
---
title: Upgrade Notes
-sidebar_position: 3
+sidebar_position: 4
---
# Upgrade Notes from v0.7 to v0.8
diff --git a/website/docs/maintenance/operations/upgrading.md b/website/docs/maintenance/operations/upgrading.md
index f78d23282..581946d94 100644
--- a/website/docs/maintenance/operations/upgrading.md
+++ b/website/docs/maintenance/operations/upgrading.md
@@ -1,6 +1,6 @@
---
title: Upgrading and Compatibility
-sidebar_position: 1
+sidebar_position: 3
---
As an online storage service, Fluss is typically designed to operate over extended periods, often spanning several years.
diff --git a/website/src/css/custom.css b/website/src/css/custom.css
index 52bebc647..42d4a9a86 100644
--- a/website/src/css/custom.css
+++ b/website/src/css/custom.css
@@ -190,7 +190,6 @@
color: var(--ifm-color-primary);
}
ul {
- margin: 16px 0;
padding-left: 20px;
li {
margin-top: 4px;
@@ -208,7 +207,6 @@
}*/
}
ol {
- margin: 16px 0;
padding-left: 20px;
li {
list-style: decimal;
(fluss) 04/17: [hotfix] fix sidebar naming conversion (#1898)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 5da6cdd3fe03b60b35ad7a5a02e158368110950e
Author: Giannis Polyzos <ip...@gmail.com>
AuthorDate: Thu Oct 30 09:37:33 2025 +0200
[hotfix] fix sidebar naming conversion (#1898)
(cherry picked from commit 3551d1ffec8801070e05f145896eae28e737c055)
---
website/docs/quickstart/_category_.json | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/website/docs/quickstart/_category_.json b/website/docs/quickstart/_category_.json
index 133c30f64..d06bba85d 100644
--- a/website/docs/quickstart/_category_.json
+++ b/website/docs/quickstart/_category_.json
@@ -1,4 +1,4 @@
{
- "label": "Quick Start",
+ "label": "Quickstart",
"position": 2
}
(fluss) 12/17: [client] Add support for inner class and is/has methods for boolean (#1908)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 6f1715fc86133d32152961d431d623a1ea5b63ee
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Fri Oct 31 07:23:49 2025 +0100
[client] Add support for inner class and is/has methods for boolean (#1908)
* [client] Add support for inner class and is/has methods for boolean
* fix
(cherry picked from commit 7af5c0cc94db6afae3bbc21bf635c7c501426815)
---
.../apache/fluss/client/converter/PojoType.java | 110 ++++++++++----
.../fluss/client/converter/PojoTypeTest.java | 163 +++++++++++++++++++++
2 files changed, 244 insertions(+), 29 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java b/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
index 78af49700..7790b9143 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java
@@ -66,8 +66,8 @@ final class PojoType<T> {
Constructor<T> ctor = requirePublicDefaultConstructor(pojoClass);
Map<String, Field> allFields = discoverAllInstanceFields(pojoClass);
- Map<String, Method> getters = discoverGetters(pojoClass);
- Map<String, Method> setters = discoverSetters(pojoClass);
+ Map<String, Method> getters = discoverGetters(pojoClass, allFields);
+ Map<String, Method> setters = discoverSetters(pojoClass, allFields);
Map<String, Property> props = new LinkedHashMap<>();
for (Map.Entry<String, Field> e : allFields.entrySet()) {
@@ -85,10 +85,11 @@ final class PojoType<T> {
if (!publicField) {
// When not a public field, require both getter and setter
if (getter == null || setter == null) {
+ final String capitalizedName = capitalize(name);
throw new IllegalArgumentException(
String.format(
"POJO class %s field '%s' must be public or have both getter and setter (get%s/set%s).",
- pojoClass.getName(), name, capitalize(name), capitalize(name)));
+ pojoClass.getName(), name, capitalizedName, capitalizedName));
}
}
props.put(
@@ -108,22 +109,30 @@ final class PojoType<T> {
}
private static <T> Constructor<T> requirePublicDefaultConstructor(Class<T> pojoClass) {
- try {
- Constructor<T> ctor = pojoClass.getDeclaredConstructor();
- if (!Modifier.isPublic(ctor.getModifiers())) {
- throw new IllegalArgumentException(
- String.format(
- "POJO class %s must have a public default constructor.",
- pojoClass.getName()));
+ Constructor<T>[] ctors = (Constructor<T>[]) pojoClass.getConstructors();
+ for (Constructor<T> c : ctors) {
+ if (!Modifier.isPublic(c.getModifiers())) {
+ continue;
+ }
+ if (c.getParameterCount() == 0) {
+
+ return c;
+ }
+ if (c.getParameterCount() == 1
+ && pojoClass
+ .getName()
+ .equals(
+ c.getParameterTypes()[0].getName()
+ + "$"
+ + pojoClass.getSimpleName())) {
+ return c;
}
- return ctor;
- } catch (NoSuchMethodException e) {
- throw new IllegalArgumentException(
- String.format(
- "POJO class %s must have a public default constructor.",
- pojoClass.getName()),
- e);
}
+
+ throw new IllegalArgumentException(
+ String.format(
+ "POJO class %s must have a public default constructor.",
+ pojoClass.getName()));
}
private static Map<String, Field> discoverAllInstanceFields(Class<?> clazz) {
@@ -135,6 +144,13 @@ final class PojoType<T> {
if (Modifier.isStatic(mod) || Modifier.isTransient(mod)) {
continue;
}
+ // Skip references to enclosing class
+ if (f.getName().startsWith("this$")) {
+ final Class type = f.getType();
+ if ((type.getName() + "$" + clazz.getSimpleName()).equals(clazz.getName())) {
+ continue;
+ }
+ }
f.setAccessible(true);
fields.putIfAbsent(f.getName(), f);
}
@@ -143,32 +159,68 @@ final class PojoType<T> {
return fields;
}
- private static Map<String, Method> discoverGetters(Class<?> clazz) {
- Map<String, Method> getters = new HashMap<>();
+ private static Map<String, Method> discoverGetters(
+ Class<?> clazz, Map<String, Field> fieldMap) {
+ final Map<String, Method> getters = new HashMap<>();
for (Method m : clazz.getMethods()) { // public methods incl. inherited
- if (m.getParameterCount() == 0
- && m.getName().startsWith("get")
- && !m.getReturnType().equals(void.class)) {
- String prop = decapitalize(m.getName().substring(3));
+ final String prop = getGetterProp(m);
+ if (fieldMap.containsKey(prop)) {
getters.put(prop, m);
}
}
return getters;
}
- private static Map<String, Method> discoverSetters(Class<?> clazz) {
- Map<String, Method> setters = new HashMap<>();
+ private static String getGetterProp(Method m) {
+ if (m.getParameterCount() != 0) {
+ return null;
+ }
+ final Class<?> returnType = m.getReturnType();
+ if (void.class.equals(returnType) || Void.class.equals(returnType)) {
+ return null;
+ }
+ final String name = m.getName();
+ if (name.startsWith("get")) {
+ return decapitalize(name.substring(3));
+ }
+ if (returnType.equals(boolean.class) || returnType.equals(Boolean.class)) {
+ if (name.startsWith("is")) {
+ return decapitalize(name.substring(2));
+ }
+ if (name.startsWith("has")) {
+ return decapitalize(name.substring(3));
+ }
+ }
+ return null;
+ }
+
+ private static Map<String, Method> discoverSetters(
+ Class<?> clazz, Map<String, Field> fieldMap) {
+ final Map<String, Method> setters = new HashMap<>();
for (Method m : clazz.getMethods()) { // public methods incl. inherited
- if (m.getParameterCount() == 1
- && m.getName().startsWith("set")
- && m.getReturnType().equals(void.class)) {
- String prop = decapitalize(m.getName().substring(3));
+ final String prop = getSetterProp(m);
+ if (fieldMap.containsKey(prop)) {
setters.put(prop, m);
}
}
return setters;
}
+ private static String getSetterProp(Method m) {
+ if (m.getParameterCount() != 1) {
+ return null;
+ }
+ final Class<?> returnType = m.getReturnType();
+ if (!void.class.equals(returnType) && !Void.class.equals(returnType)) {
+ return null;
+ }
+ final String name = m.getName();
+ if (name.startsWith("set")) {
+ return decapitalize(name.substring(3));
+ }
+ return null;
+ }
+
private static String capitalize(String s) {
if (s == null || s.isEmpty()) {
return s;
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java b/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
new file mode 100644
index 000000000..e2d44f581
--- /dev/null
+++ b/fluss-client/src/test/java/org/apache/fluss/client/converter/PojoTypeTest.java
@@ -0,0 +1,163 @@
+/*
+ * 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.fluss.client.converter;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Basic tests for {@link PojoType}. */
+class PojoTypeTest {
+ @Test
+ void test() {
+ assertThatThrownBy(() -> PojoType.of(ClassWithNoPublicConstructor.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must have a public default constructor.");
+
+ assertThatThrownBy(() -> PojoType.of(ClassWithNonWithNonPublicField.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(" must be public.");
+
+ assertThatThrownBy(() -> PojoType.of(PublicClass.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "Primitive types are not allowed; all fields must be nullable (use wrapper types).");
+
+ assertThatThrownBy(() -> PojoType.of(PublicClass.InnerClass.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(
+ "Primitive types are not allowed; all fields must be nullable (use wrapper types).");
+
+ assertThatThrownBy(() -> PojoType.of(PublicWithNonPrimitive.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be public or have both getter and setter");
+
+ assertThatThrownBy(() -> PojoType.of(PublicWithPublicWithBoolean.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be public or have both getter and setter");
+
+ assertThatThrownBy(() -> PojoType.of(PublicWithPublicWithBooleanWithGetterOnly.class))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("must be public or have both getter and setter");
+
+ PojoType.of(PublicWithPublicWithBooleanWithGetterAndSetter.class);
+ PojoType.of(PublicWithPublicWithBooleanWithIsAndSetter.class);
+ PojoType.of(PublicWithPublicWithBooleanWithHasAndSetter.class);
+ PojoType.of(PublicWithPublicNonPrimitive.class);
+ }
+
+ public class ClassWithNoPublicConstructor {
+ int f;
+ int j;
+
+ private ClassWithNoPublicConstructor() {}
+ }
+
+ class ClassWithNonWithNonPublicField {
+ int f;
+ int j;
+ }
+
+ public class PublicClass {
+ public class InnerClass {
+ final int e;
+
+ public InnerClass() {
+ e = 2;
+ }
+ }
+
+ final int f;
+ final int j;
+
+ public PublicClass() {
+ f = 1;
+ j = 1;
+ }
+ }
+
+ public class PublicWithNonPrimitive {
+ String s;
+
+ public PublicWithNonPrimitive() {}
+ }
+
+ public class PublicWithPublicNonPrimitive {
+ public String s;
+
+ public PublicWithPublicNonPrimitive() {}
+ }
+
+ public class PublicWithPublicWithBoolean {
+ private Boolean b;
+
+ public PublicWithPublicWithBoolean() {}
+ }
+
+ public class PublicWithPublicWithBooleanWithGetterOnly {
+ private Boolean b;
+
+ public PublicWithPublicWithBooleanWithGetterOnly() {}
+
+ public Boolean getB() {
+ return b;
+ }
+ }
+
+ public class PublicWithPublicWithBooleanWithGetterAndSetter {
+ private Boolean b;
+
+ public PublicWithPublicWithBooleanWithGetterAndSetter() {}
+
+ public Boolean getB() {
+ return b;
+ }
+
+ public void setB(boolean b) {
+ this.b = b;
+ }
+ }
+
+ public class PublicWithPublicWithBooleanWithIsAndSetter {
+ private Boolean b;
+
+ public PublicWithPublicWithBooleanWithIsAndSetter() {}
+
+ public Boolean isB() {
+ return b;
+ }
+
+ public void setB(boolean b) {
+ this.b = b;
+ }
+ }
+
+ public class PublicWithPublicWithBooleanWithHasAndSetter {
+ private Boolean b;
+
+ public PublicWithPublicWithBooleanWithHasAndSetter() {}
+
+ public Boolean hasB() {
+ return b;
+ }
+
+ public void setB(boolean b) {
+ this.b = b;
+ }
+ }
+}
(fluss) 06/17: [docs] upgrade 0.7 to 0.8 about sasl authorization and default database. (#1871)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 0bcc928843f3a05bda3e9afd26744804f8a78666
Author: Hongshun Wang <12...@users.noreply.github.com>
AuthorDate: Thu Oct 30 16:25:24 2025 +0800
[docs] upgrade 0.7 to 0.8 about sasl authorization and default database. (#1871)
(cherry picked from commit aa11a268ec790d4e933562b3bb0fd562f4fe024b)
---
website/docs/maintenance/operations/upgrade-notes-0.8.md | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/website/docs/maintenance/operations/upgrade-notes-0.8.md b/website/docs/maintenance/operations/upgrade-notes-0.8.md
index 9f0c07cbd..a2697dd03 100644
--- a/website/docs/maintenance/operations/upgrade-notes-0.8.md
+++ b/website/docs/maintenance/operations/upgrade-notes-0.8.md
@@ -98,3 +98,12 @@ This change prioritizes **tiering service stability and performance**:
- **Disable auto-compaction** for tables where tiering speed is more important than storage optimization
- **Monitor resource usage** when enabling auto-compaction to ensure it doesn't impact tiering performance
- **Consider manual compaction** for large tables during maintenance windows
+
+## SASL Plain Authorization jaas configuration
+Due to Fluss being donated to the Apache Software Foundation, the package namespace has been changed from `org.alibaba.fluss` to `org.apache.fluss`.
+The `security.sasl.plain.jaas.config` configuration has been updated accordingly:
+* Server side: `security.sasl.plain.jaas.config` is changed from `org.alibaba.fluss.security.auth.sasl.plain.PlainLoginModule` to `org.apache.fluss.security.auth.sasl.plain.PlainLoginModule`.
+* Client side: `client.security.sasl.jaas.config` is changed from `org.alibaba.fluss.security.auth.sasl.plain.PlainLoginModule` to `org.apache.fluss.security.auth.sasl.plain.PlainLoginModule`. If you are using client.security.sasl.username and client.security.sasl.password configurations, no changes are required as these remain compatible.
+
+## Flink Catalog required Default database.
+Fluss catalog used to not need defalut database even though a wrong defalut database won't be checked. From 0.8, it's required to specify a default database.
\ No newline at end of file
(fluss) 05/17: [hotfix][client] Use more efficient `EnumMap` (#1882)
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 8775613a9a68257d4326481ed4d7840a4f604d90
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Thu Oct 30 09:24:45 2025 +0100
[hotfix][client] Use more efficient `EnumMap` (#1882)
(cherry picked from commit c2a785f2075e785a07e034523594179421ede77a)
---
.../main/java/org/apache/fluss/client/converter/ConverterCommons.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java b/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
index f7766d36d..4a1e99246 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java
@@ -24,7 +24,7 @@ import org.apache.fluss.types.RowType;
import java.math.BigDecimal;
import java.util.Arrays;
-import java.util.HashMap;
+import java.util.EnumMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@@ -43,7 +43,7 @@ final class ConverterCommons {
static final Map<DataTypeRoot, Set<Class<?>>> SUPPORTED_TYPES = createSupportedTypes();
private static Map<DataTypeRoot, Set<Class<?>>> createSupportedTypes() {
- Map<DataTypeRoot, Set<Class<?>>> map = new HashMap<>();
+ Map<DataTypeRoot, Set<Class<?>>> map = new EnumMap<>(DataTypeRoot.class);
map.put(DataTypeRoot.BOOLEAN, setOf(Boolean.class));
map.put(DataTypeRoot.TINYINT, setOf(Byte.class));
map.put(DataTypeRoot.SMALLINT, setOf(Short.class));