private Configuration getConfiguration() throws Exception {
	// Flink configuration
	final Configuration config = new Configuration();

	config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
	config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, DEFAULT_PARALLELISM);

	UUID id = UUID.randomUUID();
	final File checkpointDir = TEMP_FOLDER.newFolder("checkpoints_" + id).getAbsoluteFile();
	final File savepointDir = TEMP_FOLDER.newFolder("savepoints_" + id).getAbsoluteFile();

	if (!checkpointDir.exists() || !savepointDir.exists()) {
		throw new Exception("Test setup failed: failed to create (temporary) directories.");
	}

	LOG.info("Created temporary checkpoint directory: " + checkpointDir + ".");
	LOG.info("Created savepoint directory: " + savepointDir + ".");

	config.setString(CheckpointingOptions.STATE_BACKEND, "memory");
	config.setString(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDir.toURI().toString());
	config.set(CheckpointingOptions.FS_SMALL_FILE_THRESHOLD, MemorySize.ZERO);
	config.setString(CheckpointingOptions.SAVEPOINT_DIRECTORY, savepointDir.toURI().toString());
	config.setLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, 300L);

	return config;
}
 

private static Configuration getConfiguration() {
	Configuration config = new Configuration();
	config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
	config.setLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL);

	return config;
}
 

/**
 * Creates an HeartbeatServices instance from a {@link Configuration}.
 *
 * @param configuration Configuration to be used for the HeartbeatServices creation
 * @return An HeartbeatServices instance created from the given configuration
 */
public static HeartbeatServices fromConfiguration(Configuration configuration) {
	long heartbeatInterval = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL);

	long heartbeatTimeout = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT);

	return new HeartbeatServices(heartbeatInterval, heartbeatTimeout);
}
 

private static Configuration getConfiguration() {
	Configuration config = new Configuration();
	config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
	config.setLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL);

	return config;
}
 

/**
 * Creates an HeartbeatServices instance from a {@link Configuration}.
 *
 * @param configuration Configuration to be used for the HeartbeatServices creation
 * @return An HeartbeatServices instance created from the given configuration
 */
public static HeartbeatServices fromConfiguration(Configuration configuration) {
	long heartbeatInterval = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL);

	long heartbeatTimeout = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT);

	return new HeartbeatServices(heartbeatInterval, heartbeatTimeout);
}
 

private static Configuration getConfiguration() {
	Configuration config = new Configuration();
	config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
	config.setLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL);

	return config;
}
 

/**
 * Creates an HeartbeatServices instance from a {@link Configuration}.
 *
 * @param configuration Configuration to be used for the HeartbeatServices creation
 * @return An HeartbeatServices instance created from the given configuration
 */
public static HeartbeatServices fromConfiguration(Configuration configuration) {
	long heartbeatInterval = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL);

	long heartbeatTimeout = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT);

	return new HeartbeatServices(heartbeatInterval, heartbeatTimeout);
}