Unlike the OSS / S3 plugins, JuiceFS does not require Fluss to obtain or distribute a + * delegation token: the JuiceFS client itself authenticates against the meta server using + * locally-configured credentials (e.g. {@code juicefs.access-key} / {@code juicefs.secret-key}, or + * implicit IAM). We therefore return an empty placeholder token from {@link + * #obtainSecurityToken()}. + */ +class JuiceFsFileSystem extends HadoopFileSystem { + + private static final ObtainedSecurityToken EMPTY_TOKEN = + new ObtainedSecurityToken( + JuiceFsPlugin.SCHEME, new byte[0], null, Collections.emptyMap()); + + JuiceFsFileSystem(FileSystem hadoopFileSystem) { + super(hadoopFileSystem); + } + + @Override + public ObtainedSecurityToken obtainSecurityToken() { + return EMPTY_TOKEN; + } +} diff --git a/fluss-filesystems/fluss-fs-juicefs/src/main/java/org/apache/fluss/fs/juicefs/JuiceFsPlugin.java b/fluss-filesystems/fluss-fs-juicefs/src/main/java/org/apache/fluss/fs/juicefs/JuiceFsPlugin.java new file mode 100644 index 00000000000..3aedb793d7d --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/main/java/org/apache/fluss/fs/juicefs/JuiceFsPlugin.java @@ -0,0 +1,147 @@ +/* + * 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.fs.juicefs; + +import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.config.ConfigBuilder; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.fs.FileSystem; +import org.apache.fluss.fs.FileSystemPlugin; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.URI; + +/** + * Simple factory for the JuiceFS file system. + * + *
This plugin registers the {@code jfs} scheme and bridges Fluss to the JuiceFS Hadoop SDK + * ({@code io.juicefs.JuiceFileSystem}). The JuiceFS client itself manages authentication via its + * meta server / access keys, so no Fluss-side delegation token is required. + * + *
Configuration is propagated from Fluss to the underlying Hadoop {@link + * org.apache.hadoop.conf.Configuration} for any key starting with {@code fs.jfs.} or {@code + * juicefs.}. Two defaults are injected when not provided by the user: + * + *
This is package-private to allow direct testing without a live JuiceFS meta server. + */ + @VisibleForTesting + static void applyJuiceFsDefaults(org.apache.hadoop.conf.Configuration hadoopConfig) { + if (hadoopConfig.get(FS_JFS_IMPL_KEY) == null) { + hadoopConfig.set(FS_JFS_IMPL_KEY, JUICEFS_HADOOP_FS_IMPL); + } + if (hadoopConfig.get(FS_JFS_IMPL_DISABLE_CACHE_KEY) == null) { + hadoopConfig.set(FS_JFS_IMPL_DISABLE_CACHE_KEY, "false"); + } + } + + @VisibleForTesting + org.apache.hadoop.conf.Configuration getHadoopConfiguration(Configuration flussConfig) { + org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration(); + if (flussConfig == null) { + return conf; + } + + // read all configuration entries with a prefix in 'FLUSS_CONFIG_PREFIXES' + for (String key : flussConfig.keySet()) { + for (String prefix : FLUSS_CONFIG_PREFIXES) { + if (key.startsWith(prefix)) { + String value = + flussConfig.getString( + ConfigBuilder.key(key).stringType().noDefaultValue(), null); + conf.set(key, value); + LOG.debug("Adding Fluss config entry {} to Hadoop config", key); + break; + } + } + } + return conf; + } +} diff --git a/fluss-filesystems/fluss-fs-juicefs/src/main/java/org/apache/fluss/fs/juicefs/JuiceFsSecurityTokenReceiver.java b/fluss-filesystems/fluss-fs-juicefs/src/main/java/org/apache/fluss/fs/juicefs/JuiceFsSecurityTokenReceiver.java new file mode 100644 index 00000000000..79353bf9a46 --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/main/java/org/apache/fluss/fs/juicefs/JuiceFsSecurityTokenReceiver.java @@ -0,0 +1,53 @@ +/* + * 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.fs.juicefs; + +import org.apache.fluss.fs.token.ObtainedSecurityToken; +import org.apache.fluss.fs.token.SecurityTokenReceiver; + +/** + * Security token receiver for JuiceFS filesystems. + * + *
Unlike the OSS / S3 / COS plugins, Fluss does not perform any STS or delegation-token exchange + * for JuiceFS: the JuiceFS client itself authenticates locally against the metadata engine and the + * backing object storage. Consequently {@link + * org.apache.fluss.fs.juicefs.JuiceFsFileSystem#obtainSecurityToken()} returns an empty placeholder + * token with scheme {@code "jfs"}. + * + *
This receiver exists solely to satisfy the contract of {@code SecurityTokenReceiverRepository} + * on the client side. Without it, every placeholder token arriving with scheme {@code "jfs"} would + * be reported as {@code "Token arrived for service but no receiver found for it: jfs"}, which is + * caught inside {@code DefaultSecurityTokenManager} and translated into a periodic re-schedule of + * the token renewal task after {@code client.filesystem.security.token.renewal.backoff} — a + * persistent, unnecessary retry / log-noise loop on every JuiceFS-enabled client. + * + *
The implementation mirrors {@link org.apache.fluss.fs.hdfs.HdfsSecurityTokenReceiver} and is + * intentionally a no-op. + */ +public class JuiceFsSecurityTokenReceiver implements SecurityTokenReceiver { + + @Override + public String scheme() { + return JuiceFsPlugin.SCHEME; + } + + @Override + public void onNewTokensObtained(ObtainedSecurityToken token) { + // no-op: JuiceFS authenticates locally on each node, so there is nothing to install. + } +} diff --git a/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/NOTICE b/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/NOTICE new file mode 100644 index 00000000000..0e5dc95ca97 --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/NOTICE @@ -0,0 +1,59 @@ +fluss-fs-juicefs +Copyright 2025-2026 The Apache Software Foundation + +This project includes software developed at +The Apache Software Foundation (http://www.apache.org/). + +This project bundles the following dependencies under the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt) + +- com.fasterxml.jackson.core:jackson-annotations:2.15.3 +- com.fasterxml.jackson.core:jackson-core:2.15.3 +- com.fasterxml.jackson.core:jackson-databind:2.15.3 +- com.fasterxml.woodstox:woodstox-core:5.4.0 +- com.google.guava:failureaccess:1.0 +- com.google.guava:guava:27.0-jre +- com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava +- com.google.j2objc:j2objc-annotations:1.1 +- com.google.re2j:re2j:1.1 +- commons-beanutils:commons-beanutils:1.9.4 +- commons-collections:commons-collections:3.2.2 +- commons-io:commons-io:2.14.0 +- commons-logging:commons-logging:1.2 +- dnsjava:dnsjava:3.4.0 +- io.dropwizard.metrics:metrics-core:3.2.4 +- io.juicefs:juicefs-hadoop:1.4.0 +- io.netty:netty-buffer:4.1.100.Final +- io.netty:netty-codec:4.1.100.Final +- io.netty:netty-common:4.1.100.Final +- io.netty:netty-handler:4.1.100.Final +- io.netty:netty-resolver:4.1.100.Final +- io.netty:netty-transport-classes-epoll:4.1.100.Final +- io.netty:netty-transport-native-epoll:4.1.100.Final +- io.netty:netty-transport-native-unix-common:4.1.100.Final +- io.netty:netty-transport:4.1.100.Final +- jakarta.activation:jakarta.activation-api:1.2.1 +- org.apache.commons:commons-compress:1.24.0 +- org.apache.commons:commons-configuration2:2.8.0 +- org.apache.commons:commons-lang3:3.18.0 +- org.apache.commons:commons-text:1.10.0 +- org.apache.hadoop:hadoop-annotations:3.4.0 +- org.apache.hadoop:hadoop-auth:3.4.0 +- org.apache.hadoop:hadoop-common:3.4.0 +- org.apache.hadoop.thirdparty:hadoop-shaded-guava:1.2.0 +- org.apache.hadoop.thirdparty:hadoop-shaded-protobuf_3_21:1.2.0 +- org.apache.kerby:kerb-core:2.0.3 +- org.apache.kerby:kerby-asn1:2.0.3 +- org.apache.kerby:kerby-pkix:2.0.3 +- org.apache.kerby:kerby-util:2.0.3 +- org.codehaus.jettison:jettison:1.5.4 +- org.codehaus.mojo:animal-sniffer-annotations:1.17 +- org.codehaus.woodstox:stax2-api:4.2.1 +- org.xerial.snappy:snappy-java:1.1.10.4 + +This project bundles the following dependencies under the Bouncy Castle License (https://www.bouncycastle.org/licence.html) + +- org.bouncycastle:bcprov-jdk15on:1.70 + +This project bundles the following dependencies under the MIT License (https://opensource.org/licenses/MIT) + +- org.checkerframework:checker-qual:2.5.2 diff --git a/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin b/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin new file mode 100644 index 00000000000..96de7aefe2a --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/services/org.apache.fluss.fs.FileSystemPlugin @@ -0,0 +1,17 @@ +# 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. + +org.apache.fluss.fs.juicefs.JuiceFsPlugin diff --git a/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver b/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver new file mode 100644 index 00000000000..5234992e28f --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/main/resources/META-INF/services/org.apache.fluss.fs.token.SecurityTokenReceiver @@ -0,0 +1,17 @@ +# 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. + +org.apache.fluss.fs.juicefs.JuiceFsSecurityTokenReceiver diff --git a/fluss-filesystems/fluss-fs-juicefs/src/test/java/org/apache/fluss/fs/juicefs/JuiceFsPluginTest.java b/fluss-filesystems/fluss-fs-juicefs/src/test/java/org/apache/fluss/fs/juicefs/JuiceFsPluginTest.java new file mode 100644 index 00000000000..04e0de4af20 --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/test/java/org/apache/fluss/fs/juicefs/JuiceFsPluginTest.java @@ -0,0 +1,83 @@ +/* + * 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.fs.juicefs; + +import org.apache.fluss.config.Configuration; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests that validate the behavior of the JuiceFS File System Plugin. */ +class JuiceFsPluginTest { + + @Test + void testScheme() { + assertThat(new JuiceFsPlugin().getScheme()).isEqualTo("jfs"); + } + + @Test + void testHadoopConfigPropagation() { + Configuration flussConfig = new Configuration(); + flussConfig.setString("fs.jfs.access-key", "ak-value"); + flussConfig.setString("juicefs.meta", "redis://meta-host:6379/1"); + flussConfig.setString("juicefs.cache-dir", "/var/jfsCache"); + // unrelated key — should NOT be forwarded + flussConfig.setString("unrelated.key", "x"); + + org.apache.hadoop.conf.Configuration hadoopConfig = + new JuiceFsPlugin().getHadoopConfiguration(flussConfig); + + assertThat(hadoopConfig.get("fs.jfs.access-key")).isEqualTo("ak-value"); + assertThat(hadoopConfig.get("juicefs.meta")).isEqualTo("redis://meta-host:6379/1"); + assertThat(hadoopConfig.get("juicefs.cache-dir")).isEqualTo("/var/jfsCache"); + assertThat(hadoopConfig.get("unrelated.key")).isNull(); + } + + @Test + void testApplyJuiceFsDefaultsInjectsImpl() { + org.apache.hadoop.conf.Configuration hadoopConfig = + new org.apache.hadoop.conf.Configuration(false); + JuiceFsPlugin.applyJuiceFsDefaults(hadoopConfig); + + assertThat(hadoopConfig.get(JuiceFsPlugin.FS_JFS_IMPL_KEY)) + .isEqualTo(JuiceFsPlugin.JUICEFS_HADOOP_FS_IMPL); + assertThat(hadoopConfig.get(JuiceFsPlugin.FS_JFS_IMPL_DISABLE_CACHE_KEY)) + .isEqualTo("false"); + } + + @Test + void testApplyJuiceFsDefaultsDoesNotOverrideUserValues() { + org.apache.hadoop.conf.Configuration hadoopConfig = + new org.apache.hadoop.conf.Configuration(false); + hadoopConfig.set(JuiceFsPlugin.FS_JFS_IMPL_KEY, "com.foo.MyJuiceFs"); + hadoopConfig.set(JuiceFsPlugin.FS_JFS_IMPL_DISABLE_CACHE_KEY, "true"); + + JuiceFsPlugin.applyJuiceFsDefaults(hadoopConfig); + + assertThat(hadoopConfig.get(JuiceFsPlugin.FS_JFS_IMPL_KEY)).isEqualTo("com.foo.MyJuiceFs"); + assertThat(hadoopConfig.get(JuiceFsPlugin.FS_JFS_IMPL_DISABLE_CACHE_KEY)).isEqualTo("true"); + } + + @Test + void testGetHadoopConfigurationWithNullFlussConfig() { + org.apache.hadoop.conf.Configuration hadoopConfig = + new JuiceFsPlugin().getHadoopConfiguration(null); + assertThat(hadoopConfig).isNotNull(); + } +} diff --git a/fluss-filesystems/fluss-fs-juicefs/src/test/java/org/apache/fluss/fs/juicefs/JuiceFsSecurityTokenReceiverTest.java b/fluss-filesystems/fluss-fs-juicefs/src/test/java/org/apache/fluss/fs/juicefs/JuiceFsSecurityTokenReceiverTest.java new file mode 100644 index 00000000000..d84d886be12 --- /dev/null +++ b/fluss-filesystems/fluss-fs-juicefs/src/test/java/org/apache/fluss/fs/juicefs/JuiceFsSecurityTokenReceiverTest.java @@ -0,0 +1,72 @@ +/* + * 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.fs.juicefs; + +import org.apache.fluss.fs.token.ObtainedSecurityToken; +import org.apache.fluss.fs.token.SecurityTokenReceiver; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.ServiceLoader; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +/** + * Tests for {@link JuiceFsSecurityTokenReceiver}. + * + *
These tests protect the client-side flow against the regression described in {@link
+ * JuiceFsSecurityTokenReceiver}: without a receiver registered for the {@code "jfs"} scheme, the
+ * placeholder token returned by {@link JuiceFsFileSystem#obtainSecurityToken()} triggers an {@code
+ * IllegalStateException} inside {@code SecurityTokenReceiverRepository}, which {@code
+ * DefaultSecurityTokenManager} translates into a periodic retry loop.
+ */
+class JuiceFsSecurityTokenReceiverTest {
+
+ @Test
+ void schemeShouldBeJfs() {
+ assertThat(new JuiceFsSecurityTokenReceiver().scheme()).isEqualTo(JuiceFsPlugin.SCHEME);
+ }
+
+ @Test
+ void onNewTokensObtainedShouldBeANoOpForPlaceholderToken() {
+ JuiceFsSecurityTokenReceiver receiver = new JuiceFsSecurityTokenReceiver();
+ // Same shape as the token produced by JuiceFsFileSystem#obtainSecurityToken().
+ ObtainedSecurityToken placeholder =
+ new ObtainedSecurityToken(
+ JuiceFsPlugin.SCHEME, new byte[0], null, Collections.emptyMap());
+
+ assertThatCode(() -> receiver.onNewTokensObtained(placeholder)).doesNotThrowAnyException();
+ }
+
+ @Test
+ void serviceLoaderShouldDiscoverJfsReceiver() {
+ // Mirrors the discovery contract used by SecurityTokenReceiverRepository#loadReceivers().
+ List