Configures the cluster state.

@param configuration The cluster configuration.
@return The cluster state.
493112
ClusterState configure(Configuration configuration) {
    Assert.notNull(configuration, "configuration");

    // If the configuration index is less than the currently configured index, ignore it.
    // Configurations can be persisted and applying old configurations can revert newer configurations.
    if (this.configuration != null && configuration.index() <= this.configuration.index())
      return this;

    Instant time = Instant.ofEpochMilli(configuration.time());

    // Iterate through members in the new configuration, add any missing members, and update existing members.
    boolean transition = false;
    for (Member member : configuration.members()) {
      if (member.equals(this.member)) {
        transition = this.member.type().ordinal() < member.type().ordinal();
        this.member.update(member.type(), time).update(member.clientAddress(), time);
        members.add(this.member);
      } else {
        // If the member state doesn't already exist, create it.
        MemberState state = membersMap.get(member.id());
        if (state == null) {
          state = new MemberState(new ServerMember(member.type(), member.serverAddress(), member.clientAddress(), time), this);
          state.resetState(context.getLog());
          this.members.add(state.getMember());
          this.remoteMembers.add(state);
          membersMap.put(member.id(), state);
          addressMap.put(member.address(), state);
          joinListeners.accept(state.getMember());
        }

        // If the member type has changed, update the member type and reset its state.
        state.getMember().update(member.clientAddress(), time);
        if (state.getMember().type() != member.type()) {
          state.getMember().update(member.type(), time);
          state.resetState(context.getLog());
        }

        // If the member status has changed, update the local member status.
        if (state.getMember().status() != member.status()) {
          state.getMember().update(member.status(), time);
        }

        // Update the optimized member collections according to the member type.
        for (List<MemberState> memberType : memberTypes.values()) {
          memberType.remove(state);
        }

        List<MemberState> memberType = memberTypes.get(member.type());
        if (memberType == null) {
          memberType = new CopyOnWriteArrayList<>();
          memberTypes.put(member.type(), memberType);
        }
        memberType.add(state);
      }
    }

    // Transition the local member only if the member is being promoted and not demoted.
    // Configuration changes that demote the local member are only applied to the local server
    // upon commitment. This ensures that e.g. a leader that's removing itself from the quorum
    // can commit the configuration change prior to shutting down.
    if (transition) {
      context.transition(this.member.type());
    }

    // Iterate through configured members and remove any that no longer exist in the configuration.
    int i = 0;
    while (i < this.remoteMembers.size()) {
      MemberState member = this.remoteMembers.get(i);
      if (!configuration.members().contains(member.getMember())) {
        this.members.remove(member.getMember());
        this.remoteMembers.remove(i);
        for (List<MemberState> memberType : memberTypes.values()) {
          memberType.remove(member);
        }
        membersMap.remove(member.getMember().id());
        addressMap.remove(member.getMember().address());
        leaveListeners.accept(member.getMember());
      } else {
        i++;
      }
    }

    // If the local member was removed from the cluster, remove it from the members list.
    if (!configuration.members().contains(member)) {
      members.remove(member);
    }

    this.configuration = configuration;

    // Store the configuration if it's already committed.
    if (context.getCommitIndex() >= configuration.index()) {
      context.getMetaStore().storeConfiguration(configuration);
    }

    // Reassign members based on availability.
    reassign();

    return this;
  }
1298842
public void addCacheConfiguration(String partition, Configuration<String, byte[]> configuration) {
		cacheConfigurations.put(partition, configuration);
	}
1165451
@Override
    public Configuration getConfiguration() {
        try {
            globalLock.readLock().lock();
            return configuration;
        } finally {
            globalLock.readLock().unlock();
        }
    }
291251
protected org.apache.commons.configuration.Configuration getConfiguration() {
        try {
            return ApplicationProperties.get();
        } catch (AtlasException e) {
            throw new RuntimeException("Unable to load configuration: " + ApplicationProperties.APPLICATION_PROPERTIES);
        }
    }
1013284
public RaftClusterContext configure(Configuration configuration) {
    checkNotNull(configuration, "configuration cannot be null");

    // If the configuration index is less than the currently configured index, ignore it.
    // Configurations can be persisted and applying old configurations can revert newer configurations.
    if (this.configuration != null && configuration.index() <= this.configuration.index()) {
      return this;
    }

    Instant time = Instant.ofEpochMilli(configuration.time());

    // Iterate through members in the new configuration, add any missing members, and update existing members.
    boolean transition = false;
    for (RaftMember member : configuration.members()) {
      if (member.equals(this.member)) {
        transition = this.member.getType().ordinal() < member.getType().ordinal();
        this.member.update(member.getType(), time);
        members.add(this.member);
      } else {
        // If the member state doesn't already exist, create it.
        RaftMemberContext state = membersMap.get(member.memberId());
        if (state == null) {
          DefaultRaftMember defaultMember = new DefaultRaftMember(member.memberId(), member.getType(), time);
          state = new RaftMemberContext(defaultMember, this);
          state.resetState(raft.getLog());
          this.members.add(state.getMember());
          this.remoteMembers.add(state);
          membersMap.put(member.memberId(), state);
          listeners.forEach(l -> l.event(new RaftClusterEvent(RaftClusterEvent.Type.JOIN, defaultMember, time.toEpochMilli())));
        }

        // If the member type has changed, update the member type and reset its state.
        if (state.getMember().getType() != member.getType()) {
          state.getMember().update(member.getType(), time);
          state.resetState(raft.getLog());
        }

        // Update the optimized member collections according to the member type.
        for (List<RaftMemberContext> memberType : memberTypes.values()) {
          memberType.remove(state);
        }

        List<RaftMemberContext> memberType = memberTypes.get(member.getType());
        if (memberType == null) {
          memberType = new CopyOnWriteArrayList<>();
          memberTypes.put(member.getType(), memberType);
        }
        memberType.add(state);
      }
    }

    // Transition the local member only if the member is being promoted and not demoted.
    // Configuration changes that demote the local member are only applied to the local server
    // upon commitment. This ensures that e.g. a leader that's removing itself from the quorum
    // can commit the configuration change prior to shutting down.
    if (transition) {
      raft.transition(this.member.getType());
    }

    // Iterate through configured members and remove any that no longer exist in the configuration.
    int i = 0;
    while (i < this.remoteMembers.size()) {
      RaftMemberContext member = this.remoteMembers.get(i);
      if (!configuration.members().contains(member.getMember())) {
        this.members.remove(member.getMember());
        this.remoteMembers.remove(i);
        for (List<RaftMemberContext> memberType : memberTypes.values()) {
          memberType.remove(member);
        }
        membersMap.remove(member.getMember().memberId());
        listeners.forEach(l -> l.event(new RaftClusterEvent(RaftClusterEvent.Type.LEAVE, member.getMember(), time.toEpochMilli())));
      } else {
        i++;
      }
    }

    // If the local member was removed from the cluster, remove it from the members list.
    if (!configuration.members().contains(member)) {
      members.remove(member);
    }

    this.configuration = configuration;

    // Store the configuration if it's already committed.
    if (raft.getCommitIndex() >= configuration.index()) {
      raft.getMetaStore().storeConfiguration(configuration);
    }

    return this;
  }
230397
private static AlluxioConfiguration loadClusterConfiguration(GetConfigurationPResponse response,
      AlluxioConfiguration conf) {
    String clientVersion = conf.get(PropertyKey.VERSION);
    LOG.info("Alluxio client (version {}) is trying to load cluster level configurations",
        clientVersion);
    List<alluxio.grpc.ConfigProperty> clusterConfig = response.getConfigsList();
    Properties clusterProps = loadClientProperties(clusterConfig, (key, value) ->
        String.format("Loading property: %s (%s) -> %s", key, key.getScope(), value));
    // Check version.
    String clusterVersion = clusterProps.get(PropertyKey.VERSION).toString();
    if (!clientVersion.equals(clusterVersion)) {
      LOG.warn("Alluxio client version ({}) does not match Alluxio cluster version ({})",
          clientVersion, clusterVersion);
      clusterProps.remove(PropertyKey.VERSION);
    }
    // Merge conf returned by master as the cluster default into conf object
    AlluxioProperties props = conf.copyProperties();
    props.merge(clusterProps, Source.CLUSTER_DEFAULT);
    // Use the constructor to set cluster defaults as being loaded.
    InstancedConfiguration updatedConf = new InstancedConfiguration(props, true);
    updatedConf.validate();
    LOG.info("Alluxio client has loaded cluster level configurations");
    return updatedConf;
  }
698630
public synchronized void storeConfiguration(Configuration configuration) {
    log.trace("Store configuration {}", configuration);
    byte[] bytes = serializer.encode(configuration);
    configurationBuffer.position(0)
        .writeByte(1)
        .writeInt(bytes.length)
        .write(bytes);
    configurationBuffer.flush();
  }
1041014
public java.util.List<String> getClusterStates() {
        if (clusterStates == null) {
            clusterStates = new com.amazonaws.internal.SdkInternalList<String>();
        }
        return clusterStates;
    }
412917
public void setClusters(java.util.Collection<String> clusters) {
        if (clusters == null) {
            this.clusters = null;
            return;
        }

        this.clusters = new java.util.ArrayList<String>(clusters);
    }
727216
public Configuration fromSystem() {
    MapConfiguration mapConfig =
        setupConfiguration(new MapConfiguration((Properties) System.getProperties().clone()));
    // Disables trimming so system properties that include whitespace (such as line.separator) will
    // be preserved.
    mapConfig.setTrimmingDisabled(true);
    return mapConfig;
  }
914181
public static String getCluster(Configuration jobConf) {
    String jobtracker = jobConf.get(RESOURCE_MANAGER_KEY);
    if (jobtracker == null) {
      jobtracker = jobConf.get(JOBTRACKER_KEY);
    }
    String cluster = null;
    if (jobtracker != null) {
      // strip any port number
      int portIdx = jobtracker.indexOf(':');
      if (portIdx > -1) {
        jobtracker = jobtracker.substring(0, portIdx);
      }
      // An ExceptionInInitializerError may be thrown to indicate that an exception occurred during
      // evaluation of Cluster class' static initialization
      cluster = Cluster.getIdentifier(jobtracker);
    }
    return cluster;
  }
564061
public void configure(Configuration configuration){
		setConfiguration(Objects.requireNonNull(configuration));

		setValueFactory(null);

		resetInputFields();
		resetResultFields();
	}
586075
public static Map<String, String> getClusterNameTags(Configuration conf) {
    ImmutableMap.Builder<String, String> tagMap = ImmutableMap.builder();

    String clusterIdentifierTag = ClustersNames.getInstance().getClusterName(conf);
    if (!Strings.isNullOrEmpty(clusterIdentifierTag)) {
      tagMap.put(CLUSTER_IDENTIFIER_TAG_NAME, clusterIdentifierTag);
    }
    return tagMap.build();
  }
756751
@Override
	public void open(Configuration configuration) {
		producer = getKafkaProducer(this.producerConfig);

		RuntimeContext ctx = getRuntimeContext();

		if (null != flinkKafkaPartitioner) {
			if (flinkKafkaPartitioner instanceof FlinkKafkaDelegatePartitioner) {
				((FlinkKafkaDelegatePartitioner) flinkKafkaPartitioner).setPartitions(
						getPartitionsByTopic(this.defaultTopicId, this.producer));
			}
			flinkKafkaPartitioner.open(ctx.getIndexOfThisSubtask(), ctx.getNumberOfParallelSubtasks());
		}

		LOG.info("Starting FlinkKafkaProducer ({}/{}) to produce into default topic {}",
				ctx.getIndexOfThisSubtask() + 1, ctx.getNumberOfParallelSubtasks(), defaultTopicId);

		// register Kafka metrics to Flink accumulators
		if (!Boolean.parseBoolean(producerConfig.getProperty(KEY_DISABLE_METRICS, "false"))) {
			Map<MetricName, ? extends Metric> metrics = this.producer.metrics();

			if (metrics == null) {
				// MapR's Kafka implementation returns null here.
				LOG.info("Producer implementation does not support metrics");
			} else {
				final MetricGroup kafkaMetricGroup = getRuntimeContext().getMetricGroup().addGroup("KafkaProducer");
				for (Map.Entry<MetricName, ? extends Metric> metric: metrics.entrySet()) {
					kafkaMetricGroup.gauge(metric.getKey().name(), new KafkaMetricWrapper(metric.getValue()));
				}
			}
		}

		if (flushOnCheckpoint && !((StreamingRuntimeContext) this.getRuntimeContext()).isCheckpointingEnabled()) {
			LOG.warn("Flushing on checkpoint is enabled, but checkpointing is not enabled. Disabling flushing.");
			flushOnCheckpoint = false;
		}

		if (logFailuresOnly) {
			callback = new Callback() {
				@Override
				public void onCompletion(RecordMetadata metadata, Exception e) {
					if (e != null) {
						LOG.error("Error while sending record to Kafka: " + e.getMessage(), e);
					}
					acknowledgeMessage();
				}
			};
		}
		else {
			callback = new Callback() {
				@Override
				public void onCompletion(RecordMetadata metadata, Exception exception) {
					if (exception != null && asyncException == null) {
						asyncException = exception;
					}
					acknowledgeMessage();
				}
			};
		}
	}
824384
@com.fasterxml.jackson.annotation.JsonProperty("clusterName")
    public void setClusterName(String clusterName) {
        this.clusterName = clusterName;
    }
570344
public void setConiguration(Configuration configuration) {
        this.configuration = configuration;
        getLoader().setConfiguration(configuration);
        getSaver().setConfiguration(configuration);
    }
626028
public String resolve( Map<String, String> configuration )
    {
        StringBuilder buffer = new StringBuilder();
        append( buffer, configuration, null );
        return buffer.toString();
    }