@Override
    public void init(HazelcastInstance instance, Properties properties, String mapName) {
        validateMapStoreConfig(instance, mapName);

        logger = instance.getLoggingService().getLogger(GenericMapLoader.class);

        this.instance = Util.getHazelcastInstanceImpl(instance);
        this.genericMapStoreProperties = new GenericMapStoreProperties(properties, mapName);
        this.sqlService = instance.getSql();
        this.mappingHelper = new MappingHelper(this.sqlService);

        this.mapName = mapName;
        this.mappingName = MAPPING_PREFIX + mapName;

        HazelcastProperties hzProperties = nodeEngine().getProperties();
        this.initTimeoutMillis = hzProperties.getMillis(MAPSTORE_INIT_TIMEOUT);

        ManagedExecutorService asyncExecutor = getMapStoreExecutor();

        // Init can run on partition thread, creating a mapping uses other maps, so it needs to run elsewhere
        asyncExecutor.submit(this::createOrReadMapping);
    }
@Test
    public void givenMapStoreConfigWithOffloadDisabled_thenFail() {
        MapStoreConfig mapStoreConfig = new MapStoreConfig()
                .setClassName(GenericMapLoader.class.getName())
                .setOffload(false);

        MapConfig mapConfig = new MapConfig(mapName);
        mapConfig.setMapStoreConfig(mapStoreConfig);
        instance().getConfig().addMapConfig(mapConfig);

        mapLoader = new GenericMapLoader<>();
        Properties properties = new Properties();
        properties.setProperty(DATA_CONNECTION_REF_PROPERTY, TEST_DATABASE_REF);

        assertThatThrownBy(() -> mapLoader.init(hz, properties, mapName))
                .isInstanceOf(HazelcastException.class)
                .hasMessage("MapStoreConfig for " + mapName + " must have `offload` property set to true");
    }