This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 1b6811a239 [Fix][Zeta] Fix IMap storage SPI registration lost during
shade packaging (#10884)
1b6811a239 is described below
commit 1b6811a2396a09ab6d7f232a1dc9ad07a7ed25eb
Author: zhiwei.niu <[email protected]>
AuthorDate: Thu May 14 23:10:28 2026 +0800
[Fix][Zeta] Fix IMap storage SPI registration lost during shade packaging
(#10884)
---
seatunnel-core/seatunnel-starter/pom.xml | 3 +
.../engine/server/persistence/FileMapStore.java | 26 +++--
.../server/persistence/FileMapStoreTest.java | 116 +++++++++++++++++++++
.../storage/file/IMapStorageFactorySpiTest.java | 68 ++++++++++++
4 files changed, 206 insertions(+), 7 deletions(-)
diff --git a/seatunnel-core/seatunnel-starter/pom.xml
b/seatunnel-core/seatunnel-starter/pom.xml
index 0f67174c14..8d59312727 100644
--- a/seatunnel-core/seatunnel-starter/pom.xml
+++ b/seatunnel-core/seatunnel-starter/pom.xml
@@ -137,6 +137,9 @@
<exclude>org.apache.seatunnel:seatunnel-hadoop3-3.1.4-uber</exclude>
</excludes>
</artifactSet>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"
/>
+ </transformers>
</configuration>
</plugin>
</plugins>
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
index 60cc514c17..2900430440 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/persistence/FileMapStore.java
@@ -27,6 +27,7 @@ import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.MapLoaderLifecycleSupport;
import com.hazelcast.map.MapStore;
import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
import java.util.Collection;
import java.util.Collections;
@@ -34,20 +35,31 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+@Slf4j
public class FileMapStore implements MapStore<Object, Object>,
MapLoaderLifecycleSupport {
private IMapStorage mapStorage;
@Override
public void init(HazelcastInstance hazelcastInstance, Properties
properties, String mapName) {
-
Map<String, Object> initMap = new
HashMap<>(Maps.fromProperties(properties));
- this.mapStorage =
- FactoryUtil.discoverFactory(
- Thread.currentThread().getContextClassLoader(),
- IMapStorageFactory.class,
- (String) initMap.get("type"))
- .create(initMap);
+ String storageType = (String) initMap.get("type");
+ try {
+ this.mapStorage =
+ FactoryUtil.discoverFactory(
+
Thread.currentThread().getContextClassLoader(),
+ IMapStorageFactory.class,
+ storageType)
+ .create(initMap);
+ } catch (RuntimeException e) {
+ log.error(
+ "Failed to initialize IMap storage for map '{}',
type='{}'. "
+ + "Cluster state will NOT be persisted.",
+ mapName,
+ storageType,
+ e);
+ throw e;
+ }
}
@Override
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
new file mode 100644
index 0000000000..3c6d9e1c61
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/persistence/FileMapStoreTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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.seatunnel.engine.server.persistence;
+
+import org.apache.seatunnel.engine.imap.storage.file.common.FileConstants;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.io.TempDir;
+
+import com.hazelcast.core.HazelcastInstance;
+
+import java.nio.file.Path;
+import java.util.Properties;
+
+import static org.mockito.Mockito.mock;
+
+/**
+ * Unit tests for FileMapStore.
+ *
+ * <p>Covers two scenarios introduced/improved by the fix for
+ * https://github.com/apache/seatunnel/issues/10883:
+ *
+ * <ol>
+ * <li>init() with a valid local-fs config succeeds end-to-end (SPI
discovery + storage init).
+ * <li>init() with an unknown storage type throws immediately instead of
silently failing.
+ * </ol>
+ *
+ * <p>Note: HDFS/S3/OSS backends are NOT tested here because they require
Hadoop uber jars and
+ * remote infrastructure. Those are covered by IMapFileStorageTest. What we
verify here is that the
+ * SPI discovery path (FactoryUtil.discoverFactory) is wired correctly through
FileMapStore, and
+ * that failures surface as exceptions rather than silent no-ops.
+ */
+@EnabledOnOs({OS.LINUX, OS.MAC})
+public class FileMapStoreTest {
+
+ @TempDir Path tempDir;
+
+ /**
+ * Verifies that FileMapStore.init() can successfully resolve the "hdfs"
factory via SPI and
+ * initialize local-filesystem storage without errors.
+ *
+ * <p>This test fails if ServicesResourceTransformer is absent from the
shade config, because
+ * ServiceLoader would return no IMapStorageFactory implementations.
+ */
+ @Test
+ public void testInitSucceedsWithLocalFileSystem() {
+ FileMapStore store = new FileMapStore();
+ Properties props = buildLocalFsProperties(tempDir.toString());
+
+ Assertions.assertDoesNotThrow(
+ () -> store.init(mock(HazelcastInstance.class), props,
"test-map"),
+ "FileMapStore.init() should succeed with local-fs config. "
+ + "If this fails with 'Could not find any factories', "
+ + "ServicesResourceTransformer is likely missing from
seatunnel-starter shade config.");
+
+ store.destroy();
+ }
+
+ /**
+ * Verifies that FileMapStore.init() throws (rather than silently
swallowing) when the storage
+ * type is unknown.
+ *
+ * <p>Before the fix, Hazelcast's MapLoader lifecycle could absorb the
exception and leave
+ * mapStorage null, causing NullPointerException on the first store/load
call with no indication
+ * of the real root cause.
+ */
+ @Test
+ public void testInitThrowsOnUnknownStorageType() {
+ FileMapStore store = new FileMapStore();
+ Properties props = new Properties();
+ props.setProperty("type", "non-existent-storage-type");
+
+ RuntimeException ex =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> store.init(mock(HazelcastInstance.class), props,
"test-map"),
+ "FileMapStore.init() must throw on unknown storage
type instead of silently failing.");
+
+ String msg = ex.getMessage() != null ? ex.getMessage() : "";
+ Assertions.assertTrue(
+ msg.contains("non-existent-storage-type")
+ || msg.contains("Could not find any factories"),
+ "Exception message should identify the unknown type. Got: " +
msg);
+ }
+
+ private Properties buildLocalFsProperties(String namespace) {
+ Properties props = new Properties();
+ // "hdfs" is the factoryIdentifier of IMapFileStorageFactory — covers
local fs via
+ // fs.defaultFS=file:///
+ props.setProperty("type", "hdfs");
+ props.setProperty("fs.defaultFS", "file:///");
+ props.setProperty("fs.file.impl",
"org.apache.hadoop.fs.LocalFileSystem");
+ props.setProperty(FileConstants.FileInitProperties.BUSINESS_KEY,
"test");
+ props.setProperty(FileConstants.FileInitProperties.NAMESPACE_KEY,
namespace);
+ props.setProperty(FileConstants.FileInitProperties.CLUSTER_NAME,
"test-cluster");
+ return props;
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapStorageFactorySpiTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapStorageFactorySpiTest.java
new file mode 100644
index 0000000000..d77faddcc9
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/IMapStorageFactorySpiTest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.seatunnel.engine.imap.storage.file;
+
+import org.apache.seatunnel.engine.imap.storage.api.IMapStorageFactory;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.ServiceLoader;
+import java.util.stream.Collectors;
+
+public class IMapStorageFactorySpiTest {
+
+ @Test
+ public void testHdfsFactoryIsDiscoverableViaSpi() {
+ List<IMapStorageFactory> factories = new ArrayList<>();
+ ServiceLoader.load(IMapStorageFactory.class,
Thread.currentThread().getContextClassLoader())
+ .forEach(factories::add);
+
+ Assertions.assertFalse(
+ factories.isEmpty(),
+ "ServiceLoader found no IMapStorageFactory implementations. "
+ + "This likely means META-INF/services was lost during
shade packaging. "
+ + "Ensure ServicesResourceTransformer is configured in
seatunnel-starter/pom.xml.");
+
+ List<String> identifiers =
+ factories.stream()
+ .map(IMapStorageFactory::factoryIdentifier)
+ .collect(Collectors.toList());
+
+ Assertions.assertTrue(
+ identifiers.contains("hdfs"),
+ "No IMapStorageFactory with identifier 'hdfs' found.
Available: " + identifiers);
+ }
+
+ @Test
+ public void testNoAmbiguousFactoryIdentifiers() {
+ List<String> identifiers = new ArrayList<>();
+ ServiceLoader.load(IMapStorageFactory.class,
Thread.currentThread().getContextClassLoader())
+ .forEach(f -> identifiers.add(f.factoryIdentifier()));
+
+ long distinctCount = identifiers.stream().distinct().count();
+ Assertions.assertEquals(
+ identifiers.size(),
+ distinctCount,
+ "Duplicate IMapStorageFactory identifiers detected after SPI
merge: "
+ + identifiers
+ + ". ServicesResourceTransformer may have merged
conflicting registrations.");
+ }
+}
