How to inject spring resources into Apache Ignite classes?

Question

I have a problem with injection spring beans into some ignite's classes. I'm trying to create this: Client -> Apache Ignite -> Spring-Data -> DataBase Maybe it is wrong, i'm not sure.

So, at this moment my classes look like: AppConfiguration

@Configuration
@ComponentScan(basePackages = arrayOf("com.ignite.cache"))
open class AppConfiguration : Serializable {
    private val logger: Logger = Logger.getLogger(AppConfiguration::class.java)

    @Bean
    open fun igniteInstance(): Ignite {
        val cfg = IgniteConfiguration()

        cfg.igniteInstanceName = "springDataNode"

        cfg.isPeerClassLoadingEnabled = true


        var clientCache: CacheConfiguration<Long, Client> = CacheConfiguration("ClientCache")
        clientCache.apply {
            setIndexedTypes(Long::class.java, Client::class.java)
            setCacheStoreFactory(FactoryBuilder.factoryOf(ClientStore::class.java))
            isReadThrough = true
            isWriteThrough = true
        }

        cfg.setCacheConfiguration(clientCache)

        return Ignition.start(cfg)
    }
}

DataSourceConfiguration:

@Configuration
@EnableTransactionManagement
@EnableJpaRepositories(basePackages = arrayOf("com.ignite.cache.model.repositories.springdatarepository"))
@EnableIgniteRepositories(basePackages = arrayOf("com.ignite.cache.model.repositories.igniterepository"))
@ComponentScan(basePackages = arrayOf("com.ignite.cache.model"))
open class DataSourceConfiguration : Serializable {

    @Bean
    open fun entityManagerFactory(): LocalContainerEntityManagerFactoryBean {
        var entityManagerFactory: LocalContainerEntityManagerFactoryBean = LocalContainerEntityManagerFactoryBean()

        entityManagerFactory.apply {
            dataSource = dataSource()
            setPackagesToScan("com.ignite.cache.model")

            var vendorAdapter: HibernateJpaVendorAdapter = HibernateJpaVendorAdapter()

            vendorAdapter.apply {
                setGenerateDdl(true)
                setShowSql(true)
            }

            var properties: Properties = Properties()

            properties.apply {
                put("database.dialet", "org.hibernate.dialect.PostgreSQL95Dialect")
                put("database.globally_quoted_identifiers", "false")
                put("database.enable_lazy_load_no_trans", "true")
                put("database.show_sql", "true")
            }

            jpaVendorAdapter = vendorAdapter
            setJpaProperties(properties)
        }

        return entityManagerFactory
    }

    @Bean
    open fun dataSource(): DataSource {
        var source: ComboPooledDataSource = ComboPooledDataSource()

        source.apply {
            driverClass = "org.postgresql.Driver"
            jdbcUrl = "jdbc:postgresql://localhost:5432/ignite"
            user = "postgres"
            password = "1111"
            acquireIncrement = 5
            idleConnectionTestPeriod = 60
            maxPoolSize = 20
            minPoolSize = 10
            initialPoolSize = 10
        }

        return source
    }

    @Bean
    open fun transactionManager() : PlatformTransactionManager {
        var manager: JpaTransactionManager = JpaTransactionManager()

        manager.apply {
            entityManagerFactory = entityManagerFactory().nativeEntityManagerFactory
        }

        return manager
    }

    @Bean
    open fun exceptionTranslator(): PersistenceExceptionTranslationPostProcessor = PersistenceExceptionTranslationPostProcessor()
}

Entity:

@Entity
@Table(name = "client")
data class Client
(
        @Id
        @Column(name = "id")
        @GeneratedValue(generator = "increment")
        @GenericGenerator(name = "increment", strategy = "increment")
        var id: Long = 0,
        @Column(name = "email", nullable = false, unique = true)
        var email: String = "",
        @Column(name = "login", nullable = false, unique = true)
        var login: String = ""
) : Serializable

Repositories:

@RepositoryConfig(cacheName = "ClientCache")
interface IgniteClientRepository : IgniteRepository<Client, Long> {
}
@Repository
interface ClientRepository : CrudRepository<Client, Long>

Standart implemenation for service classes and CacheStore for ignite:

public class ClientStore implements CacheStore<Long, Client>, Serializable {
    private Logger logger = Logger.getLogger(ClientStore.class);

    @SpringResource
    private IClientService clientRepository; // <- error is here (NPE)

    @Override
    public void loadCache(IgniteBiInClosure<Long, Client> igniteBiInClosure, @Nullable Object... objects) throws CacheLoaderException {
        Iterable<Client> clients = clientRepository.findAll();

        for(Client client : clients) {
            igniteBiInClosure.apply(client.getId(), client);
        }
    }
...
}

Main:

public class Main {
    private static Logger logger = Logger.getLogger(Main.class);

    public static void main(String[] args) {
        ApplicationContext context = new AnnotationConfigApplicationContext(AppConfiguration.class, DataSourceConfiguration.class);
        Ignite ignite = context.getBean(Ignite.class);

        (ignite.getOrCreateCache("ClientCache")).loadCache(null);

        IgniteClientRepository clientRepository = context.getBean(IgniteClientRepository.class);

        Iterable<Client> clients = clientRepository.findAll();
        for(Client client : clients) {
            logger.info(client);
        }

        logger.info("Finish");
    }
}

But when i try to load some data from database into cache i get error NPE:

Exception in thread "main" javax.cache.integration.CacheLoaderException: java.lang.NullPointerException
    at org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter.loadCache(GridCacheStoreManagerAdapter.java:528)
    at org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter.localLoadCache(GridDhtCacheAdapter.java:486)
    at org.apache.ignite.internal.processors.cache.GridCacheProxyImpl.localLoadCache(GridCacheProxyImpl.java:217)
    at org.apache.ignite.internal.processors.cache.GridCacheAdapter$LoadCacheJob.localExecute(GridCacheAdapter.java:5439)
    at org.apache.ignite.internal.processors.cache.GridCacheAdapter$LoadCacheJobV2.localExecute(GridCacheAdapter.java:5488)
    at org.apache.ignite.internal.processors.cache.GridCacheAdapter$TopologyVersionAwareJob.execute(GridCacheAdapter.java:6103)
    at org.apache.ignite.compute.ComputeJobAdapter.call(ComputeJobAdapter.java:132)
    at org.apache.ignite.internal.processors.closure.GridClosureProcessor$C2.execute(GridClosureProcessor.java:1842)
    at org.apache.ignite.internal.processors.job.GridJobWorker$2.call(GridJobWorker.java:566)
    at org.apache.ignite.internal.util.IgniteUtils.wrapThreadLoader(IgniteUtils.java:6621)
    at org.apache.ignite.internal.processors.job.GridJobWorker.execute0(GridJobWorker.java:560)
    at org.apache.ignite.internal.processors.job.GridJobWorker.body(GridJobWorker.java:489)
    at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:110)
    at org.apache.ignite.internal.processors.job.GridJobProcessor.processJobExecuteRequest(GridJobProcessor.java:1114)
    at org.apache.ignite.internal.processors.task.GridTaskWorker.sendRequest(GridTaskWorker.java:1379)
    at org.apache.ignite.internal.processors.task.GridTaskWorker.processMappedJobs(GridTaskWorker.java:640)
    at org.apache.ignite.internal.processors.task.GridTaskWorker.body(GridTaskWorker.java:532)
    at org.apache.ignite.internal.util.worker.GridWorker.run(GridWorker.java:110)
    at org.apache.ignite.internal.processors.task.GridTaskProcessor.startTask(GridTaskProcessor.java:743)
    at org.apache.ignite.internal.processors.task.GridTaskProcessor.execute(GridTaskProcessor.java:443)
    at org.apache.ignite.internal.processors.closure.GridClosureProcessor.callAsync(GridClosureProcessor.java:447)
    at org.apache.ignite.internal.processors.closure.GridClosureProcessor.callAsync(GridClosureProcessor.java:418)
    at org.apache.ignite.internal.processors.closure.GridClosureProcessor.callAsync(GridClosureProcessor.java:402)
    at org.apache.ignite.internal.processors.cache.GridCacheAdapter.globalLoadCacheAsync(GridCacheAdapter.java:3681)
    at org.apache.ignite.internal.processors.cache.GridCacheAdapter.globalLoadCache(GridCacheAdapter.java:3657)
    at org.apache.ignite.internal.processors.cache.IgniteCacheProxy.loadCache(IgniteCacheProxy.java:387)
    at com.ignite.cache.Main.main(Main.java:22)
Caused by: java.lang.NullPointerException
    at com.ignite.cache.model.service.ClientStore.loadCache(ClientStore.java:30)
    at org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter.loadCache(GridCacheStoreManagerAdapter.java:502)
    ... 26 more

I still can't figure out why my client service doesn't inject into CacheStore class. Maybe i should use xml config instead of java-class config for ignite?

Answer 1

When you start Ignite using Ignition#start with IgniteConfiguration object, it's not aware of Spring context at all. You need to use IgniteSpring#start methods instead to provide context explicitly. Another option is to utilize IgniteSpringBean that already implements ApplicationContextAware and starts Ignite instance properly.

Also note that you will need to provide either bean name or bean class as a parameter to @SpringResource annotation.

Answer 2

I don't know Ignite and I don't know the programming language you are using but it looks like Ignite and not Spring is creating your ClientStore instances. Therefore you need to inject the dependencies manually.

I'm not really familiar with the FactoryBuilder used there, so there might be nice solutions if you can use a constructor with arguments or even a lambda, but if that doesn' work, you can store a reference to the repository in static field and get it from there in the constructor of the ClientStore.