class SimpleDistributedProperty<T extends Serializable> implements DistributedChangeableProperty<T> {
    /** Name of property. */
    private final String name;

    /** Description of property. */
    private final String description;

    /** Property value. */
    protected volatile T val;

    /** Sign of attachment to the processor. */
    private volatile boolean attached = false;

    /** Listeners of property update. */
    private final ConcurrentLinkedQueue<DistributePropertyListener<? super T>> updateListeners = new ConcurrentLinkedQueue<>();

    /**
     * Specific consumer for update value in cluster. It is null when property doesn't ready to update value on cluster
     * wide.
     */
    @GridToStringExclude
    private volatile PropertyUpdateClosure clusterWideUpdater;

    /** Property value parser. */
    @GridToStringExclude
    private final Function<String, T> parser;

    /**
     * @param name Name of property.
     * @param parser Property value parser.
     * @param description Description of property.
     */
    public SimpleDistributedProperty(String name, Function<String, T> parser, String description) {
        this.name = name;
        this.parser = parser;
        this.description = description;
    }

    /** {@inheritDoc} */
    @Override public boolean propagate(T newVal) throws IgniteCheckedException {
        ensureClusterWideUpdateIsReady();

        clusterWideUpdater.update(name, newVal).get();

        return true;
    }

    /**
     * @throws DetachedPropertyException If this property have not been attached to processor yet, please call {@link
     * DistributedConfigurationProcessor#registerProperty(DistributedChangeableProperty)} before this method.
     * @throws NotWritablePropertyException If this property don't ready to cluster wide update yet, perhaps cluster is
     * not active yet.
     */
    private void ensureClusterWideUpdateIsReady() throws DetachedPropertyException, NotWritablePropertyException {
        if (!attached)
            throw new DetachedPropertyException(name);

        if (clusterWideUpdater == null)
            throw new NotWritablePropertyException(name);
    }

    /** {@inheritDoc} */
    @Override public GridFutureAdapter<?> propagateAsync(T newVal) throws IgniteCheckedException {
        ensureClusterWideUpdateIsReady();

        return clusterWideUpdater.update(name, newVal);
    }

    /** {@inheritDoc} */
    @Override public GridFutureAdapter<?> propagateAsync(T expectedVal, T newVal) throws IgniteCheckedException {
        ensureClusterWideUpdateIsReady();

        return clusterWideUpdater.casUpdate(name, expectedVal, newVal);
    }

    /** {@inheritDoc} */
    @Override public T get() {
        return val;
    }

    /** {@inheritDoc} */
    @Override public T getOrDefault(T dfltVal) {
        return val == null ? dfltVal : val;
    }

    /** {@inheritDoc} */
    @Override public String getName() {
        return name;
    }

    /** {@inheritDoc} */
    @Override public String description() {
        return description;
    }

    /** {@inheritDoc} */
    @Override public void addListener(DistributePropertyListener<? super T> listener) {
        updateListeners.add(listener);
    }

    /** {@inheritDoc} */
    @Override public void onAttached() {
        attached = true;
    }

    /** {@inheritDoc} */
    @Override public void onReadyForUpdate(@NotNull PropertyUpdateClosure updater) {
        this.clusterWideUpdater = updater;
    }

    /** {@inheritDoc} */
    @Override public void localUpdate(Serializable newVal) {
        T oldVal = val;

        val = (T)newVal;

        updateListeners.forEach(listener -> listener.onUpdate(name, oldVal, val));
    }

    /** {@inheritDoc} */
    @Override public T parse(String str) {
        if (parser == null)
            throw new IgniteException("The parser is not specified for property [name=" + name + ']');

        return parser.apply(str);
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return S.toString(SimpleDistributedProperty.class, this);
    }

    /**
     * @param val String to parse.
     * @return Integer value.
     */
    public static Integer parseNonNegativeInteger(String val) {<FILL_FUNCTION_BODY>}

    /**
     * @param val String to parse.
     * @return Long value.
     */
    public static Long parseNonNegativeLong(String val) {
        if (val == null || val.trim().isEmpty())
            return null;

        long intVal = Long.parseLong(val);

        if (intVal < 0)
            throw new IllegalArgumentException("The value must not be negative");

        return intVal;
    }

    /**
     * @param val String value.
     * @return String set.
     */
    public static HashSet<String> parseStringSet(String val) {
        HashSet<String> set = new HashSet<>();

        if (val == null || val.trim().isEmpty())
            return set;

        String[] vals = val.split("\\W+");

        set.addAll(Arrays.asList(vals));

        return set;
    }
}

        if (val == null || val.trim().isEmpty())
            return null;

        int intVal = Integer.parseInt(val);

        if (intVal < 0)
            throw new IllegalArgumentException("The value must not be negative");

        return intVal;
    
<no_super_class>