使用 JNDI 进行休眠搜索

Question

我正在使用多租户的 Hibernate Search 6（请参阅 Hibernate Search 6 with multitenancy issue, HSEARCH000520, HSEARCH600029）。我的环境：Hibernate ORM 5.4.28、Hibernate Search 6.0.2、Payara server 2021.1 和 MariaDb。我配置了 2 个数据源（2 个数据库）——myDS 和 my2ndDS。我可以使用下面的多租户解析器方法通过成功引用租户 ID 来查找/合并实体。我也将这种方法应用于搜索（参见下面的编码）。现在的问题是当我搜索某些内容时会在下面显示错误。

@PersistenceUnit
private EntityManagerFactory emf;

public EntityManager getEM(final String tenantId) {
    final SessionFactoryImplementor sf = emf.unwrap(SessionFactoryImplementor.class);

    final MultitenancyResolver tenantResolver = (MultitenancyResolver) sf.getCurrentTenantIdentifierResolver();
    tenantResolver.setTenantIdentifier(tenantId);
    return emf.createEntityManager();
}

休眠搜索类/方法：

    @Stateless
    public class SearchAnnouncementMessage {
      ...
    private EntityManager getEM(final String tenantId) {
     ...
    } 
    public ResultSearchObject searchAnnouncementMsgs(final String tenantId,
            final boolean reindexWithHibernateSearch, final String searchWord,
            final int[] range) {
     ....
          final SearchSession searchSession = Search.session(getEM(tenantId));
            if (reindexWithHibernateSearch) {
                logger.info("Reindex with HibernateSearch");
                try {
                    searchSession.massIndexer()
                            .idFetchSize(150)
                            .batchSizeToLoadObjects(25)
                            .threadsToLoadObjects(THREADS_LOAD_OBJ)
                            .transactionTimeout(SEARCH_TIMEOUT)
                            .startAndWait();
                } catch (final InterruptedException e) {
                    logger.info("#1 can't search at this time; error: {}", () -> e.getMessage());
                    return null;
                }
            }
            try {
    
                logger.info("search in AnnouncementMsgs");
                final SearchQuery<AnnouncementMsgs> result = searchSession.search(AnnouncementMsgs.class).extension(LuceneExtension.get())
                        .where(f -> f.bool(b -> { 
       ...
    }
}

错误信息（显示在 Search.session(getEM(tenantId));:

#
HSEARCH000058: Exception occurred javax.persistence.PersistenceException: org.hibernate.HibernateException: Error trying to get datasource ['java:app/jdbc/my2ndDS']
Failing operation:
Fetching identifiers of entities to index for entity 'Users' during mass indexing
javax.persistence.PersistenceException: org.hibernate.HibernateException: Error trying to get datasource ['java:app/jdbc/my2ndDS']
    at org.hibernate.internal.ExceptionConverterImpl.convert(ExceptionConverterImpl.java:154)
    at org.hibernate.query.internal.AbstractProducedQuery.list(AbstractProducedQuery.java:1602)
    at org.hibernate.query.internal.AbstractProducedQuery.uniqueResult(AbstractProducedQuery.java:1635)
    at org.hibernate.query.criteria.internal.compile.CriteriaQueryTypeQueryAdapter.uniqueResult(CriteriaQueryTypeQueryAdapter.java:81)
    at org.hibernate.search.mapper.orm.massindexing.impl.IdentifierProducer.loadAllIdentifiers(IdentifierProducer.java:144)
    at org.hibernate.search.mapper.orm.massindexing.impl.IdentifierProducer.inTransactionWrapper(IdentifierProducer.java:124)
    at org.hibernate.search.mapper.orm.massindexing.impl.IdentifierProducer.run(IdentifierProducer.java:96)
    at org.hibernate.search.mapper.orm.massindexing.impl.OptionallyWrapInJTATransaction.runWithFailureHandler(OptionallyWrapInJTATransaction.java:68)
    at org.hibernate.search.mapper.orm.massindexing.impl.FailureHandledRunnable.run(FailureHandledRunnable.java:33)
    at org.hibernate.search.util.common.impl.CancellableExecutionCompletableFuture$CompletingRunnable.run(CancellableExecutionCompletableFuture.java:49)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: org.hibernate.HibernateException: Error trying to get datasource ['java:app/jdbc/my2ndDS']
    at com.dao.multitenancy.DatabaseMultiTenantProvider.getConnection(DatabaseMultiTenantProvider.java:96)
    at org.hibernate.internal.ContextualJdbcConnectionAccess.obtainConnection(ContextualJdbcConnectionAccess.java:43)
    at org.hibernate.resource.jdbc.internal.LogicalConnectionManagedImpl.acquireConnectionIfNeeded(LogicalConnectionManagedImpl.java:108)
    at org.hibernate.resource.jdbc.internal.LogicalConnectionManagedImpl.getPhysicalConnection(LogicalConnectionManagedImpl.java:138)
    at org.hibernate.engine.jdbc.internal.StatementPreparerImpl.connection(StatementPreparerImpl.java:50)
    at org.hibernate.engine.jdbc.internal.StatementPreparerImpl$5.doPrepare(StatementPreparerImpl.java:149)
    at org.hibernate.engine.jdbc.internal.StatementPreparerImpl$StatementPreparationTemplate.prepareStatement(StatementPreparerImpl.java:176)
    at org.hibernate.engine.jdbc.internal.StatementPreparerImpl.prepareQueryStatement(StatementPreparerImpl.java:151)
    at org.hibernate.loader.Loader.prepareQueryStatement(Loader.java:2103)
    at org.hibernate.loader.Loader.executeQueryStatement(Loader.java:2040)
    at org.hibernate.loader.Loader.executeQueryStatement(Loader.java:2018)
    at org.hibernate.loader.Loader.doQuery(Loader.java:948)
    at org.hibernate.loader.Loader.doQueryAndInitializeNonLazyCollections(Loader.java:349)
    at org.hibernate.loader.Loader.doList(Loader.java:2849)
    at org.hibernate.loader.Loader.doList(Loader.java:2831)
    at org.hibernate.loader.Loader.listIgnoreQueryCache(Loader.java:2663)
    at org.hibernate.loader.Loader.list(Loader.java:2658)
    at org.hibernate.loader.hql.QueryLoader.list(QueryLoader.java:506)
    at org.hibernate.hql.internal.ast.QueryTranslatorImpl.list(QueryTranslatorImpl.java:400)
    at org.hibernate.engine.query.spi.HQLQueryPlan.performList(HQLQueryPlan.java:219)
    at org.hibernate.internal.StatelessSessionImpl.list(StatelessSessionImpl.java:564)
    at org.hibernate.query.internal.AbstractProducedQuery.doList(AbstractProducedQuery.java:1625)
    at org.hibernate.query.internal.AbstractProducedQuery.list(AbstractProducedQuery.java:1593)
    ... 13 more
Caused by: javax.naming.NamingException: Lookup failed for 'java:app/jdbc/my2ndDS' in SerialContext[myEnv={java.naming.factory.initial=com.sun.enterprise.naming.impl.SerialInitContextFactory, java.naming.factory.state=com.sun.corba.ee.impl.presentation.rmi.JNDIStateFactoryImpl, java.naming.factory.url.pkgs=com.sun.enterprise.naming} [Root exception is javax.naming.NamingException: Invocation exception: Got null ComponentInvocation ]
    at com.sun.enterprise.naming.impl.SerialContext.lookup(SerialContext.java:496)
    at com.sun.enterprise.naming.impl.SerialContext.lookup(SerialContext.java:442)
    at javax.naming.InitialContext.lookup(InitialContext.java:417)
    at javax.naming.InitialContext.lookup(InitialContext.java:417)
    at com.dao.multitenancy.DatabaseMultiTenantProvider.getConnection(DatabaseMultiTenantProvider.java:94)
    ... 35 more
Caused by: javax.naming.NamingException: Invocation exception: Got null ComponentInvocation 
    at com.sun.enterprise.naming.impl.GlassfishNamingManagerImpl.getComponentId(GlassfishNamingManagerImpl.java:870)
    at com.sun.enterprise.naming.impl.GlassfishNamingManagerImpl.lookup(GlassfishNamingManagerImpl.java:737)
    at com.sun.enterprise.naming.impl.JavaURLContext.lookup(JavaURLContext.java:167)
    at com.sun.enterprise.naming.impl.SerialContext.lookup(SerialContext.java:476)
    ... 39 more
|#] 

在公共类 DatabaseMultiTenantProvider（见第 94、96 行；靠近底部）：

import java.sql.Connection;
import org.hibernate.HibernateException;
import org.hibernate.engine.config.spi.ConfigurationService;
import org.hibernate.engine.jdbc.connections.spi.MultiTenantConnectionProvider;
import org.hibernate.service.spi.ServiceRegistryAwareService;
import org.hibernate.service.spi.ServiceRegistryImplementor;

import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import javax.sql.DataSource; 
import java.sql.SQLException; 
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class DatabaseMultiTenantProvider implements MultiTenantConnectionProvider, ServiceRegistryAwareService {

    private static final long serialVersionUID = 1L;
    private static final String TENANT_SUPPORTED = "DATABASE";
    private DataSource dataSource;
    private String typeTenancy;
    private static final Logger logger = LogManager.getLogger();

    @Override
    public boolean supportsAggressiveRelease() {
        return false;
    }

    @Override
    public void injectServices(ServiceRegistryImplementor serviceRegistry) {
        logger.debug("injectService for DatabaseMultiTenantProvider");
        typeTenancy = (String) serviceRegistry
                .getService(ConfigurationService.class)
                .getSettings().get("hibernate.multiTenancy");
        logger.debug("datasouce casting result: {}", () -> serviceRegistry.getService(ConfigurationService.class).getSettings().get("hibernate.connection.datasource"));
        if (serviceRegistry
                .getService(ConfigurationService.class)
                .getSettings().get("hibernate.connection.datasource") instanceof DataSource) {
            logger.debug("can cast to DataSource");
            dataSource = (DataSource) serviceRegistry
                    .getService(ConfigurationService.class)
                    .getSettings().get("hibernate.connection.datasource");
        } else {
            logger.debug("can't cast to DataSource; have to use JNDI lookup");
            try {
                final Context init = new InitialContext();
                dataSource = (DataSource) init.lookup((String) serviceRegistry
                        .getService(ConfigurationService.class)
                        .getSettings().get("hibernate.connection.datasource"));
            } catch (final NamingException e) {
                logger.error("error in init lookup: {}", ()->e.getMessage());
                throw new RuntimeException(e);
            }
        }
    }

    @SuppressWarnings("rawtypes")
    @Override
    public boolean isUnwrappableAs(Class clazz) {
        return false;
    }

    @Override
    public <T> T unwrap(Class<T> clazz) {
        return null;
    }

    @Override
    public Connection getAnyConnection() throws SQLException {
        final Connection connection = dataSource.getConnection();
        return connection;

    }

    @Override
    public Connection getConnection(String tenantIdentifier) throws SQLException {

        
        //Just use the multitenancy if the hibernate.multiTenancy == DATABASE
        logger.debug("connecting to tenent: {}", () -> tenantIdentifier);
        if (TENANT_SUPPORTED.equals(typeTenancy)) {
            try {
 
               final Context  init = new InitialContext();
                logger.debug("use tenant datasource: {}", () -> tenantIdentifier);
                final String ds = "java:app/jdbc/"+tenantIdentifier;
                logger.debug("getConnection for: {}", ()->ds);
                dataSource = (DataSource) init.lookup(ds); //line 94
             } catch (NamingException e) {
                throw new HibernateException("Error trying to get datasource ['java:app/jdbc/" + tenantIdentifier + "']", e);//line 96 
            }
        }
        return dataSource.getConnection();

    }

    @Override
    public void releaseAnyConnection(Connection connection) throws SQLException {
        logger.debug("release any connection");
        connection.close();
    }

    @Override
    public void releaseConnection(String tenantIdentifier, Connection connection) throws SQLException {
        logger.debug("release a connection for tenentId: {}", () -> tenantIdentifier);
        releaseAnyConnection(connection);
    }
}

我认为问题应该来自带有 Hibernate Search 的 JNDI。非常感谢任何想法或提示。

更新(1)：我也用新的 ThreadLocal 进行了测试（见下文），仍然显示相同的错误。

使用新的 ThreadLocal：

public  class CallEntityManager {

    private static ThreadLocal<EntityManager> threadLocal = new ThreadLocal<>();
    private static EntityManagerFactory emf;

    public static EntityManager getEM(final String tenantId) {
        if (emf == null) { 
            emf = Persistence.createEntityManagerFactory("jakartaEEPU");
        }  
        final SessionFactoryImplementor sf = emf.unwrap(SessionFactoryImplementor.class);
        final MultitenancyResolver tenantResolver = (MultitenancyResolver) sf.getCurrentTenantIdentifierResolver();
        tenantResolver.setTenantIdentifier(tenantId);
        EntityManager em = threadLocal.get();
        if (em == null) {
            logger.debug("em is null; will create EM now");
            em = emf.createEntityManager();
            threadLocal.set(em);
        }
        return em;
    }

    public static void closeEntityManager() {
        final EntityManager em = threadLocal.get();
        if (em != null) {
            em.close();
            threadLocal.set(null); 
        }
    }
}

Answer 1

Hibernate Search 的海量索引器需要自己创建实体管理器，因为它可以并行化索引，并且您不能在多个线程中同时使用一个实体管理器。但是，它应该自动使用创建它的原始会话的租户 ID。而且，从错误消息来看，在您的情况下确实如此：那里有对my2ndDS 的引用。

据我所知，问题在于检索数据源，而不是处理多租户。并不是说 Hibernate Search 在海量索引器中创建自己的线程。您的数据源检索是否依赖于可能未在这些新线程中初始化的线程本地上下文？

一种快速而肮脏的测试方法是手动创建一个线程 (new Thread())，并尝试从该线程调用 getEM()。如果你得到同样的错误，问题可能是数据源解析依赖于一些未初始化的线程本地上下文。然后，您应该调查未显示的堆栈跟踪部分，位于“查找失败的 'java:app/jdbc/my2nDS'”下方。

顺便说一句，Got null ComponentInvocation 似乎是 Payara 的特征。这是我在网上搜索时得到的第一个结果：https://github.com/payara/Payara/issues/2430 如果我是你，我会调查 JNDI 决议中出了什么问题。

Answer 2

解决方案：

1. 在 getConnection 方法中更新类 DatabaseMultiTenantProvider（见下文），

2. 可以像开始一样使用 PersistanceContext（无需使用 ThreadLocal）。

   @Override
   public Connection getConnection(String tenantIdentifier) throws SQLException {
      //Just use the multitenancy if the hibernate.multiTenancy == DATABASE
      logger.debug("connecting to tenent DS: {}", () -> tenantIdentifier);
       if (TENANT_SUPPORTED.equals(typeTenancy)) {
        try {
   
                logger.debug("use tenant dataSource name: {}", () -> tenantIdentifier);
                //final String tenant[] = tenantIdentifier.split(":{2}");
                //if (tenant != null && tenant.length > 1) { 
                final MariaDbDataSource mds = new MariaDbDataSource();
                if (tenantIdentifier.equals("myDS")) {
                    logger.debug("using 1st dataSource: {}", () -> tenantIdentifier);
                    mds.setUrl("jdbc:mariadb://localhost:3306/mtDb");
                    mds.setUser("mtII");
                    mds.setPassword("mt123");
                    mds.setServerName("localhost");
                    mds.setPort(3306);
                    mds.setDatabaseName("mtDb");
                } else if (tenantIdentifier.equals("my2ndDS")) {
                    logger.debug("using 2nd database: {}", () -> tenantIdentifier);
                    mds.setUrl("jdbc:mariadb://localhost:3306/mt2Db");
                    mds.setUser("mtII");
                    mds.setPassword("mt123");
                    mds.setServerName("localhost");
                    mds.setPort(3306);
                    mds.setDatabaseName("mt2Db");
                }
                dataSource = mds;
                return dataSource.getConnection();
                /* } else {
                        logger.debug("normal way in connecting to dataSource");
                        final String dsURL = "java:app/jdbc/" + tenantIdentifier;
                        logger.debug("getConnection for: {}", () -> dsURL);
                        final Context init = new InitialContext();
                        dataSource = (DataSource) init.lookup(dsURL);
                        return dataSource.getConnection();
                    }*/
            } catch (Exception e) {
                //e.printStackTrace();
                throw new HibernateException("Error trying to get dataSource ['java:app/jdbc/" + tenantIdentifier + "']", e);
            }
        }
        return null;
   

}