package com.atlassian.stash.internal.hazelcast;

import com.atlassian.bitbucket.license.LicenseService;
import com.atlassian.bitbucket.server.ApplicationMode;
import com.atlassian.bitbucket.server.ApplicationPropertiesService;
import com.atlassian.bitbucket.util.PropertiesUtils;
import com.atlassian.crowd.model.application.ApplicationImpl;
import com.atlassian.crowd.model.application.DirectoryMapping;
import com.atlassian.crowd.model.application.GroupMapping;
import com.atlassian.crowd.model.directory.DirectoryImpl;
import com.atlassian.crowd.model.group.InternalGroup;
import com.atlassian.crowd.model.property.Property;
import com.atlassian.crowd.model.user.InternalUser;
import com.atlassian.fugue.Option;
import com.atlassian.hazelcast.serialization.OsgiClassLoaderRegistrySynchronizer;
import com.atlassian.hazelcast.serialization.OsgiSafe;
import com.atlassian.hazelcast.serialization.OsgiSafeStreamSerializer;
import com.atlassian.stash.internal.ApplicationConstants;
import com.atlassian.stash.internal.activity.InternalActivity;
import com.atlassian.stash.internal.auth.trusted.InternalCurrentApplication;
import com.atlassian.stash.internal.auth.trusted.InternalTrustedApplication;
import com.atlassian.stash.internal.auth.trusted.InternalTrustedApplicationRestriction;
import com.atlassian.stash.internal.cluster.ClusterNodeNameResolver;
import com.atlassian.stash.internal.comment.InternalComment;
import com.atlassian.stash.internal.comment.InternalDiffCommentAnchor;
import com.atlassian.stash.internal.commit.InternalCommitDiscussion;
import com.atlassian.stash.internal.commit.InternalCommitDiscussionParticipant;
import com.atlassian.stash.internal.integrity.InternalIntegrityEvent;
import com.atlassian.stash.internal.license.LicenseHelper;
import com.atlassian.stash.internal.license.SimpleDualLicense;
import com.atlassian.stash.internal.merge.InternalProjectMergeConfig;
import com.atlassian.stash.internal.merge.InternalRepositoryMergeConfig;
import com.atlassian.stash.internal.merge.InternalScmMergeConfig;
import com.atlassian.stash.internal.plugin.PluginSetting;
import com.atlassian.stash.internal.project.InternalProject;
import com.atlassian.stash.internal.pull.InternalPullRequest;
import com.atlassian.stash.internal.pull.InternalPullRequestParticipant;
import com.atlassian.stash.internal.pull.InternalPullRequestRescopeActivity;
import com.atlassian.stash.internal.repository.InternalRepository;
import com.atlassian.stash.internal.repository.InternalRepositoryHook;
import com.atlassian.stash.internal.repository.InternalRepositoryPushActivity;
import com.atlassian.stash.internal.task.InternalTask;
import com.atlassian.stash.internal.user.InternalApplicationUser;
import com.atlassian.stash.internal.user.InternalPermissionType;
import com.atlassian.stash.internal.user.InternalUserSettings;
import com.atlassian.stash.internal.watcher.InternalWatcher;
import com.google.common.collect.ImmutableList;
import com.hazelcast.config.*;
import com.hazelcast.instance.Capability;
import com.hazelcast.map.merge.LatestUpdateMapMergePolicy;
import com.hazelcast.spring.context.SpringManagedContext;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.util.ReflectionUtils;

import javax.servlet.http.HttpSession;
import java.util.EnumSet;
import java.util.Optional;
import java.util.Properties;
import java.util.UUID;

import static com.atlassian.stash.internal.hazelcast.HazelcastConstants.*;
import static com.google.common.base.Preconditions.checkState;
import static com.hazelcast.instance.GroupProperties.*;
import static java.util.Optional.empty;
import static java.util.Optional.of;
import static java.util.concurrent.TimeUnit.MINUTES;

/**
 * Programmatically assembles a Hazelcast {@code Config} object, allowing Hazelcast to be configured using
 * {@code bitbucket.properties} rather than Hazelcast's XML. The primary motivation here to is keep all
 * configuration in {@code bitbucket.properties}, but a secondary benefit is that we avoid Hazelcast's XML
 * configuration parsing which has proven buggy for Confluence.
 *
 * @since 3.2
 */
public class HazelcastConfigFactoryBean implements FactoryBean<Config> {

    private static final Logger log = LoggerFactory.getLogger(HazelcastConfigFactoryBean.class);

    private final ApplicationContext applicationContext;
    private final LicenseHelper licenseHelper;
    private final OsgiSafeStreamSerializer osgiSafeStreamSerializer;
    private final Properties properties;

    @Value("${hazelcast.jmx.detailed}")
    private boolean jmxDetailed;
    @Value("${jmx.enabled}")
    private boolean jmxEnabled;
    @Value("${hazelcast.group.name}")
    private String groupName;
    @Value("${hazelcast.group.password}")
    private String groupPassword;
    @Value("${hazelcast.port}")
    private int networkPort;
    @Value("${hazelcast.network.aws}")
    private boolean networkAwsEnabled;
    @Value("#{${hazelcast.network.multicast}}")
    private boolean networkMulticastEnabled;
    @Value("${hazelcast.network.tcpip}")
    private boolean networkTcpIpEnabled;
    @Value("${hazelcast.statistics.enabled}")
    private boolean statisticsEnabled;
    @Value("${application.mode}")
    private String applicationMode;

    @Autowired
    public HazelcastConfigFactoryBean(ApplicationContext applicationContext,
                                      LicenseHelper licenseHelper,
                                      OsgiSafeStreamSerializer osgiSafeStreamSerializer,
                                      @Qualifier("applicationProperties") Properties properties) {
        this.applicationContext = applicationContext;
        this.licenseHelper = licenseHelper;
        this.osgiSafeStreamSerializer = osgiSafeStreamSerializer;
        this.properties = properties;
    }

    @Override
    public Config getObject() throws Exception {
        SpringManagedContext managedContext = new SpringManagedContext();
        managedContext.setApplicationContext(applicationContext);

        Config config = new Config(HAZELCAST_INSTANCE_NAME);

        // No capabilities on connecting to the cluster. This will prevent the Master from sending data / tasks to
        // this instance until it's fully initialised. The capabilities are updated on the HazelcastLifecycle.start()
        config.setCapabilities(EnumSet.noneOf(Capability.class));

        config.setManagedContext(managedContext);
        config.setClassLoader(getClass().getClassLoader());

        configureAttributes(config);
        configureGroup(config);
        configureJmx(config);
        configureEnterprise(config);
        configureNetwork(config.getNetworkConfig());
        configureSerialization(config.getSerializationConfig());
        configureCaches(config);
        configureExecutors(config);
        configureMaps(config);
        configureScheduler(config);

        return config;
    }

    private void configureJmx(Config config) {
        config.setProperty("hazelcast.jmx", Boolean.toString(jmxEnabled));
        config.setProperty("hazelcast.jmx.detailed", Boolean.toString(jmxEnabled && jmxDetailed));
    }

    @Override
    public Class<?> getObjectType() {
        return Config.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

    private MapConfig boundedCache(Class<?> entityClass) {
        return boundedCache(entityClass.getName());
    }

    private MapConfig boundedCache(Class<?> entityClass, String field) {
        String name = entityClass.getName();
        if (StringUtils.isNotBlank(field)) {
            // Validate that the field exists
            name += "." + field;
            if (ReflectionUtils.findField(entityClass, field) == null) {
                throw new IllegalStateException(name + " field does not exist. Please update the cache configuration");
            }
        }
        return boundedCache(name);
    }

    private MapConfig boundedCache(String name) {
        return boundedCache(name, true);
    }

    private MapConfig boundedCache(String name, boolean nearCache) {
        int timeToIdleSeconds = getInt("cache." + name + ".tti", 0);
        int timeToLiveSeconds = getInt("cache." + name + ".ttl", 0);

        return boundedCache(name, getMaxOrFail("cache." + name + ".max"), timeToLiveSeconds, timeToIdleSeconds, nearCache);
    }

    private MapConfig boundedCache(String name, int size, int timeToLive, int timeToIdle, boolean nearCache) {
        // Normalize all sizes to ensure direct calls to this method still apply workable maximums
        size = normalizeSize(size);

        MapConfig config = new MapConfig(name)
                .setBackupCount(0)
                .setEvictionPolicy(EvictionPolicy.LFU)
                .setMaxIdleSeconds(Math.max(0, timeToIdle))
                .setMaxSizeConfig(new MaxSizeConfig(size, MaxSizeConfig.MaxSizePolicy.PER_NODE))
                .setMinEvictionCheckMillis(0L)
                .setStatisticsEnabled(statisticsEnabled)
                .setTimeToLiveSeconds(Math.max(0, timeToLive));

        if (nearCache) {
            config.setNearCacheConfig(defaultNearCacheConfig()
                    .setEvictionPolicy(EvictionPolicy.LFU.name())
                    .setMaxSize(size)
                    .setMaxIdleSeconds(Math.max(0, timeToIdle)))
                    .setTimeToLiveSeconds(Math.max(0, timeToLive));
        }
        return config;
    }

    private void configureAttributes(Config config) {
        MemberAttributeConfig attributeConfig = new MemberAttributeConfig();
        // Set the node's ID attribute that remains stable while the node is running. This is useful because
        // Member.uuid is reset when a node rejoins a cluster.
        attributeConfig.setStringAttribute(HazelcastConstants.ATT_NODE_VM_ID, UUID.randomUUID().toString());
        attributeConfig.setStringAttribute(HazelcastConstants.ATT_NODE_NAME, ClusterNodeNameResolver.getNodeName());
        config.setMemberAttributeConfig(attributeConfig);
    }

    private void configureCaches(Config config) {
        // OSGi bundle-to-id map, see atlassian-hazelcast-extras
        OsgiClassLoaderRegistrySynchronizer.configure(config);

        // Hibernate caches

        // Note that the UpdateTimestampsCache must NOT have a shorter expiry time than any other entity, collection,
        // or query cache, or entries in those caches may be allowed to get stale.
        config.addMapConfig(new MapConfig("org.hibernate.cache.spi.UpdateTimestampsCache")
                .setBackupCount(0)
                .setMergePolicy(LatestUpdateMapMergePolicy.class.getName())
                .setNearCacheConfig(defaultNearCacheConfig())
                .setStatisticsEnabled(statisticsEnabled));

        // Hibernate query caches (Honored by Hazelcast 3.3-RC2 and higher)
        config.addMapConfig(boundedCache("query.*", false));
        config.addMapConfig(boundedCache("org.hibernate.cache.internal.StandardQueryCache"));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_CWD_USER_BY_DIRECTORY));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_GROUP_MEMBERSHIPS));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_PERMISSIONS));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_PROJECTS));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_PUBLIC_REPOSITORY_COUNT));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_PULL_REQUEST_COUNT));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_PULL_REQUEST_PARTICIPANT_USERS));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_REPOSITORIES_BY_HIERARCHY));
        config.addMapConfig(boundedCache(ApplicationConstants.QUERY_CACHE_REPOSITORY_BY_SLUG));

        // Hibernate @NaturalId caches
        config.addMapConfig(boundedCache(InternalProjectMergeConfig.class.getName() + "##NaturalId"));
        config.addMapConfig(boundedCache(InternalRepositoryMergeConfig.class.getName() + "##NaturalId"));
        config.addMapConfig(boundedCache(InternalScmMergeConfig.class.getName() + "##NaturalId"));
        config.addMapConfig(boundedCache(PluginSetting.class.getName() + "##NaturalId"));

        // Crowd entity caches
        config.addMapConfig(boundedCache(ApplicationImpl.class));
        config.addMapConfig(boundedCache(ApplicationImpl.class, "attributes"));
        config.addMapConfig(boundedCache(ApplicationImpl.class, "directoryMappings"));
        config.addMapConfig(boundedCache(ApplicationImpl.class, "remoteAddresses"));
        config.addMapConfig(boundedCache(DirectoryImpl.class));
        config.addMapConfig(boundedCache(DirectoryImpl.class, "allowedOperations"));
        config.addMapConfig(boundedCache(DirectoryImpl.class, "attributes"));
        config.addMapConfig(boundedCache(DirectoryMapping.class));
        config.addMapConfig(boundedCache(DirectoryMapping.class, "authorisedGroups"));
        config.addMapConfig(boundedCache(DirectoryMapping.class, "allowedOperations"));
        config.addMapConfig(boundedCache(GroupMapping.class));
        config.addMapConfig(boundedCache(InternalGroup.class));
        config.addMapConfig(boundedCache(InternalUser.class));
        config.addMapConfig(boundedCache(InternalUser.class, "credentialRecords"));
        config.addMapConfig(boundedCache(Property.class));

        // Bitbucket Server entity caches
        // The cache for InternalActivity also controls all of its joined subtypes. Activities do
        // not near-cache effectively, so their near-cache is disabled.
        config.addMapConfig(boundedCache(InternalActivity.class).setNearCacheConfig(null));
        config.addMapConfig(boundedCache(InternalApplicationUser.class));
        config.addMapConfig(boundedCache(InternalComment.class));
        config.addMapConfig(boundedCache(InternalComment.class, "comments"));
        // Commit discussions do not near-cache effectively, so their near-cache is disabled.
        config.addMapConfig(boundedCache(InternalCommitDiscussion.class).setNearCacheConfig(null));
        config.addMapConfig(boundedCache(InternalCommitDiscussion.class, "participants"));
        config.addMapConfig(boundedCache(InternalCommitDiscussion.class, "watchers"));
        config.addMapConfig(boundedCache(InternalCommitDiscussionParticipant.class));
        config.addMapConfig(boundedCache(InternalCurrentApplication.class));
        config.addMapConfig(boundedCache(InternalDiffCommentAnchor.class));
        config.addMapConfig(boundedCache(InternalIntegrityEvent.class));
        config.addMapConfig(boundedCache(InternalPermissionType.class));
        config.addMapConfig(boundedCache(PluginSetting.class));
        config.addMapConfig(boundedCache(InternalProject.class));
        config.addMapConfig(boundedCache(InternalProjectMergeConfig.class));
        config.addMapConfig(boundedCache(InternalProjectMergeConfig.class, "strategyIds"));
        // Pull requests do not near-cache effectively, so their near-cache is disabled.
        config.addMapConfig(boundedCache(InternalPullRequest.class).setNearCacheConfig(null));
        config.addMapConfig(boundedCache(InternalPullRequest.class, "participants"));
        config.addMapConfig(boundedCache(InternalPullRequest.class, "watchers"));
        config.addMapConfig(boundedCache(InternalPullRequestParticipant.class));
        // Activity attributes do not near-cache effectively, so their near-cache is disabled.
        config.addMapConfig(boundedCache(InternalPullRequestRescopeActivity.class, "commits").setNearCacheConfig(null));
        config.addMapConfig(boundedCache(InternalRepository.class));
        // Repository hooks do not near-cache effectively, so their near-cache is disabled
        config.addMapConfig(boundedCache(InternalRepositoryHook.class).setNearCacheConfig(null));
        config.addMapConfig(boundedCache(InternalRepositoryMergeConfig.class));
        config.addMapConfig(boundedCache(InternalRepositoryMergeConfig.class, "strategyIds"));
        // Activity attributes do not near-cache effectively, so their near-cache is disabled.
        config.addMapConfig(boundedCache(InternalRepositoryPushActivity.class, "refChanges").setNearCacheConfig(null));
        config.addMapConfig(boundedCache(InternalScmMergeConfig.class));
        config.addMapConfig(boundedCache(InternalScmMergeConfig.class, "strategyIds"));
        config.addMapConfig(boundedCache(InternalTask.class));
        config.addMapConfig(boundedCache(InternalTrustedApplication.class));
        config.addMapConfig(boundedCache(InternalTrustedApplication.class, "restrictions"));
        config.addMapConfig(boundedCache(InternalTrustedApplicationRestriction.class));
        config.addMapConfig(boundedCache(InternalUserSettings.class));
        config.addMapConfig(boundedCache(InternalWatcher.class));

        // Atlassian Cache caches
        // Note: In general the configuration for these caches is set when they are created. However, for
        //       some caches, we want to enable near-caching for increased performance. Those caches are
        //       configured here. Note that size, TTI and TTL values set here will be overridden when the
        //       associated Atlassian Cache is created
        config.addMapConfig(new MapConfig("atlassian-cache.*")
                .setBackupCount(0)
                .setNearCacheConfig(defaultNearCacheConfig())
                .setStatisticsEnabled(statisticsEnabled));

        // Average entry size is 0.4KB; will only contain ~20 entries (ApplicationProperty.Key possible values)
        config.addMapConfig(boundedCache("atlassian-cache.Cache." + ApplicationPropertiesService.class.getName(), 0, 0, 0, true));
        // Average entry size is 0.7KB; will only contain a single entry
        config.addMapConfig(boundedCache("atlassian-cache.CacheReference." + LicenseService.class.getName() + ".BitbucketServerLicense", 0, 0, 0, true));

        // Version trackers
        config.addMapConfig(new MapConfig(MAP_VERSION_TRACKER_PREFIX + "*")
                .setBackupCount(1)
                .setMergePolicy(HighestPlusOneMergePolicy.class.getName())
                .setStatisticsEnabled(statisticsEnabled));
    }

    private void configureEnterprise(Config config) {
        String license = properties.getProperty("hazelcast.enterprise.license");
        if (StringUtils.isNotBlank(license)) {
            config.setLicenseKey(license);
        }

        String managementCenterUrl = properties.getProperty("hazelcast.managementcenter.url");
        if (StringUtils.isNotBlank(managementCenterUrl)) {
            config.setManagementCenterConfig(new ManagementCenterConfig().setEnabled(true).setUrl(managementCenterUrl));
        }
    }

    private void configureExecutors(Config config) {
        int poolSize = PropertiesUtils.parseExpression(properties.getProperty(EXECUTOR_MAX_THREADS),
                EXECUTOR_MAX_THREADS_DEFAULT);
        int queueCapacity = PropertiesUtils.parseExpression(properties.getProperty(EXECUTOR_QUEUE_SIZE),
                EXECUTOR_QUEUE_SIZE_DEFAULT);
        ExecutorConfig executorConfig = new ExecutorConfig(EXECUTOR_BITBUCKET_CORE, poolSize)
                .setQueueCapacity(queueCapacity)
                .setStatisticsEnabled(statisticsEnabled);
        config.addExecutorConfig(executorConfig);
    }

    private void configureGroup(Config config) {
        config.setGroupConfig(new GroupConfig(groupName, groupPassword));

        config.setProperty(PROP_GRACEFUL_SHUTDOWN_MAX_WAIT, properties.getProperty(GRACEFUL_SHUTDOWN_TIMEOUT));
        config.setProperty(PROP_HEALTH_MONITORING_LEVEL, properties.getProperty(HEALTH_MONITORING_LEVEL));
        config.setProperty(PROP_HEALTH_MONITORING_DELAY_SECONDS, properties.getProperty(HEALTH_MONITORING_DELAY));
        config.setProperty(PROP_MAX_NO_HEARTBEAT_SECONDS, properties.getProperty(NODE_HEARTBEAT_TIMEOUT));
        config.setProperty(PROP_OPERATION_CALL_TIMEOUT_MILLIS, properties.getProperty(OPERATION_CALL_TIMEOUT));
        config.setProperty(PROP_PARTITION_BACKUP_SYNC_INTERVAL, Long.toString(MINUTES.toSeconds(getIntOrFail(BACKUP_SYNC_INTERVAL))));
        config.setProperty(PROP_VERSION_CHECK_ENABLED, properties.getProperty(NEW_VERSION_CHECK_ENABLED));
    }

    private void configureMaps(Config config) {
        config.addMapConfig(new MapConfig("default")
                .setBackupCount(1)
                .setStatisticsEnabled(statisticsEnabled));

        config.addMapConfig(new MapConfig(HttpSession.class.getName())
                .setAsyncBackupCount(1) // Async for speed, still using a backup to try to avoid losing sessions
                .setBackupCount(0)
                .setNearCacheConfig(defaultNearCacheConfig())
                .setStatisticsEnabled(statisticsEnabled));

        config.addMapConfig(new MapConfig(HazelcastConstants.MAP_BUCKETED_EXECUTOR)
                .setAsyncBackupCount(1)
                .setBackupCount(0)
                .setStatisticsEnabled(statisticsEnabled)
                .setInMemoryFormat(InMemoryFormat.OBJECT));

        config.addMapConfig(new MapConfig("com.atlassian.stash.internal.pull.PendingUpdatesCount")
                .setAsyncBackupCount(1)
                .setBackupCount(0)
                .setStatisticsEnabled(statisticsEnabled));
    }

    private void configureNetwork(NetworkConfig config) {
        if (networkMulticastEnabled && networkTcpIpEnabled) {
            // Enabling both Multicast and TCP/IP causes Hazelcast to go into an infinite loop when trying to initialize
            // This was tested on Hazelcast v3.3-RC2
            log.warn("Both TCP/IP and Multicast have been enabled for Hazelcast node discovery. Setting to default TCP/IP");
            networkMulticastEnabled = false;
        }

        JoinConfig joinConfig = config.getJoin();

        AwsConfig awsConfig = joinConfig.getAwsConfig();
        awsConfig.setEnabled(networkAwsEnabled);
        if (awsConfig.isEnabled()) {
            String awsAccessKey = properties.getProperty("hazelcast.network.aws.access.key");
            String awsSecretKey = properties.getProperty("hazelcast.network.aws.secret.key");
            String awsIamRole = properties.getProperty("hazelcast.network.aws.iam.role");
            String awsRegion = properties.getProperty("hazelcast.network.aws.region");
            String awsHostHeader = properties.getProperty("hazelcast.network.aws.host.header");
            String awsSecurityGroupName = properties.getProperty("hazelcast.network.aws.security.group.name");
            String awsTagKey = properties.getProperty("hazelcast.network.aws.tag.key");
            String awsTagValue = properties.getProperty("hazelcast.network.aws.tag.value");
            Optional<Integer> awsTimeout = getTimeout("hazelcast.network.aws.connection.timeout.seconds");

            if (StringUtils.isNotEmpty(awsAccessKey) && StringUtils.isNotEmpty(awsSecretKey)) {
                awsConfig.setAccessKey(awsAccessKey);
                awsConfig.setSecretKey(awsSecretKey);
            }
            if (StringUtils.isNotEmpty(awsIamRole)) {
                if (StringUtils.isNotEmpty(awsAccessKey) || StringUtils.isNotEmpty(awsSecretKey)) {
                    log.warn("hazelcast.network.aws.iam.role property will take precedence over " +
                            "hazelcast.network.aws.access.key and hazelcast.network.aws.secret.key");
                }
                awsConfig.setIamRole(awsIamRole);
            }
            awsTimeout.ifPresent(awsConfig::setConnectionTimeoutSeconds);
            if (StringUtils.isNotEmpty(awsRegion)) {
                awsConfig.setRegion(awsRegion);
            }
            if (StringUtils.isNotEmpty(awsHostHeader)) {
                awsConfig.setHostHeader(awsHostHeader);
            }
            if (StringUtils.isNotEmpty(awsSecurityGroupName)) {
                awsConfig.setSecurityGroupName(awsSecurityGroupName);
            }
            if (StringUtils.isNotEmpty(awsTagKey)) {
                awsConfig.setTagKey(awsTagKey);
                awsConfig.setTagValue(awsTagValue);
            }
        }

        MulticastConfig multicastConfig = joinConfig.getMulticastConfig();
        multicastConfig.setEnabled(networkMulticastEnabled);
        if (multicastConfig.isEnabled()) {
            String multicastAddress = properties.getProperty("hazelcast.network.multicast.address");
            int multicastPort = getInt("hazelcast.network.multicast.port", MulticastConfig.DEFAULT_MULTICAST_PORT);
            Optional<Integer> multicastTimeout = getTimeout("hazelcast.network.multicast.timeout.seconds");
            Optional<Integer> multicastTtl = getTtl("hazelcast.network.multicast.ttl");

            // application-default.properties sets multicastPort to the same value as networkPort (they bind on
            // different protocols, so this works) for backwards compatibility with previous releases. If a blank
            // entry for multicast.port is added, Hazelcast's default will be used. Otherwise, if an explicit
            // port is set, that will be used.
            multicastConfig.setMulticastPort(multicastPort);
            multicastTimeout.ifPresent(multicastConfig::setMulticastTimeoutSeconds);
            multicastTtl.ifPresent(multicastConfig::setMulticastTimeToLive);
            if (StringUtils.isNotBlank(multicastAddress)) {
                multicastConfig.setMulticastGroup(multicastAddress);
            }
        }

        TcpIpConfig tcpIpConfig = joinConfig.getTcpIpConfig();
        tcpIpConfig.setEnabled(networkTcpIpEnabled);
        if (tcpIpConfig.isEnabled()) {
            String tcpIpMembers = properties.getProperty("hazelcast.network.tcpip.members");
            Optional<Integer> tcpIpTimeout = getTimeout("hazelcast.network.tcpip.connection.timeout.seconds");

            tcpIpTimeout.ifPresent(tcpIpConfig::setConnectionTimeoutSeconds);
            if (StringUtils.isNotBlank(tcpIpMembers)) {
                tcpIpConfig.addMember(tcpIpMembers); // addMember takes a comma separated list
            }
        }

        config.setPort(networkPort);

        SocketInterceptorConfig interceptorConfig = new SocketInterceptorConfig();
        interceptorConfig.setClassName(ClusterJoinSocketInterceptor.class.getName());
        interceptorConfig.setEnabled(true);
        config.setSocketInterceptorConfig(interceptorConfig);
    }

    private void configureScheduler(Config config) {
        config.addMapConfig(new MapConfig(MAP_QUARTZ_JOBSTORE_CALENDARS)
                .setStatisticsEnabled(statisticsEnabled));
        config.addMapConfig(new MapConfig(MAP_QUARTZ_JOBSTORE_JOBS)
                .setStatisticsEnabled(statisticsEnabled));
        config.addMapConfig(new MapConfig(MAP_QUARTZ_JOBSTORE_TRIGGERS)
                .setStatisticsEnabled(statisticsEnabled)
                .setMapIndexConfigs(ImmutableList.of(
                        new MapIndexConfig("jobGroup", false),
                        new MapIndexConfig("jobName", false),
                        new MapIndexConfig("nextFireTime", false),
                        new MapIndexConfig("state", false))));
    }

    private void configureSerialization(SerializationConfig config) {
        config.addSerializerConfig(new SerializerConfig()
                .setImplementation(osgiSafeStreamSerializer)
                .setTypeClass(OsgiSafe.class));
        config.addSerializerConfig(new SerializerConfig()
                .setImplementation(new LicenseStreamSerializer(licenseHelper))
                .setTypeClass(SimpleDualLicense.class));
        config.addSerializerConfig(new SerializerConfig()
                .setImplementation(new OptionStreamSerializer())
                .setTypeClass(Option.class));
    }

    private NearCacheConfig defaultNearCacheConfig() {
        return new NearCacheConfig()
                .setInMemoryFormat(InMemoryFormat.OBJECT) // Always near-cache in Object mode to avoid deserialization costs
                .setCacheLocalEntries(true); // Near-cache on the local node as well to avoid serialization costs
    }

    private int getInt(String propertyName, int defaultValue) {
        String maxSize = properties.getProperty(propertyName);
        if (StringUtils.isBlank(maxSize)) {
            return defaultValue;
        }

        return Integer.parseInt(maxSize);
    }

    private int getIntOrFail(String propertyName) {
        String maxSize = properties.getProperty(propertyName);
        checkState(!StringUtils.isBlank(maxSize), "Property '{}' is undefined", propertyName);

        return Integer.parseInt(maxSize);
    }

    /**
     * Retrieves the configured maximum value and {@link #normalizeSize(int) normalizes values}.
     *
     * @param propertyName the {@code .max} property to retrieve a value for
     * @return the configured maximum, safely normalized to at least 500
     */
    private int getMaxOrFail(String propertyName) {
        int configured = getIntOrFail(propertyName);
        int normalized = normalizeSize(configured);
        if (configured != 0 && configured != normalized) {
            // If the normalized value doesn't match the configured value log a warning about the invalid max size
            log.warn("Ignoring \"{}={}\"; values less than 500, except 0, are not valid", propertyName, configured);
        }
        return normalized;
    }

    /**
     * Retrieves the configured timeout and, if set, parses it. If the resulting value is less than zero,
     * {@code empty()} is returned. Otherwise, the parsed timeout is returned.
     *
     * @param propertyName the property to use to retrieve the timeout value
     * @return the parsed timeout, or {@code empty()} if no timeout was configured or the timeout was
     *         negative or invalid
     */
    private Optional<Integer> getTimeout(String propertyName) {
        String value = properties.getProperty(propertyName, "-1");
        try {
            int timeout = Integer.parseInt(value);
            if (timeout > -1) {
                return of(timeout);
            }
        } catch (NumberFormatException e) {
            log.warn("Ignoring \"{}={}\"; timeouts must be numbers", propertyName, value);
        }

        return empty();
    }

    /**
     * Retrieves the configured TTL and, if set, parses its value as a time-to-live. Per the specification, TTL
     * values must be between 0 and 255. If the configured value is not a number, or is not in the acceptable
     * range, {@code empty()} is returned. Otherwise, the parsed TTL value is returned.
     *
     * @param propertyName the property to use to retrieve the TTL value
     * @return the parsed TTL, or {@code empty()} if no TTL was configured or the TTL was not valid
     */
    private Optional<Integer> getTtl(String propertyName) {
        String value = properties.getProperty(propertyName);
        if (StringUtils.isBlank(value)) {
            return empty();
        }

        try {
            int ttl = Integer.parseInt(value.trim());
            if (ttl >= 0 || ttl <= 255) {
                return of(ttl);
            }
        } catch (NumberFormatException ignored) {
            // Fall-through intentional
        }

        // If we get here, either the number couldn't be parsed _or_ it wasn't in the required range
        log.warn("Ignoring \"{}={}\"; TTLs must be a number between 0 and 255", propertyName, value);
        return empty();
    }

    /**
     * Normalizes size values to ensure they will work as expected when applied to a Hazelcast {@code IMap}. The rules
     * applied are:
     * <ul>
     *     <li>Values less than 0 are normalized to Integer.MAX_VALUE (No limit)</li>
     *     <li>0 values are returned unchanged</li>
     *     <li>Values from 1 to 499 are normalized to 500</li>
     *     <li>Values of 500 or higher are returned unchanged</li>
     * </ul>
     * <p>
     * By default Hazelcast uses 271 partitions on each node to store data, so max size values are divided by 271 to
     * determine the per-partition limit. That means if the max size is less than 271, the per-partition limit is 0
     * and all {@code IMap} data is <i>continuously evicted</i>.
     *
     * @param size the size to normalize
     * @return the normalized size
     */
    private int normalizeSize(int size) {
        if (size <= 0) {
            // 0 or smaller means unlimited
            return Integer.MAX_VALUE;
        }

        return Math.max(size, 500);
    }
}
