text:
package com.cdboo.music.service;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.jaudiotagger.audio.exceptions.CannotReadException;
import org.jaudiotagger.audio.exceptions.InvalidAudioFrameException;
import org.jaudiotagger.audio.exceptions.ReadOnlyFileException;
import org.jaudiotagger.tag.TagException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import com.cdboo.channel.dao.CdbooChannelDao;
import com.cdboo.channel.entity.CdbooChannel;
import com.cdboo.channelmusic.entity.CdbooChannelMusic;
import com.cdboo.channelmusic.service.CdbooChannelMusicService;
import com.cdboo.common.Constants;
import com.cdboo.music.dao.CdbooMusicDao;
import com.cdboo.music.entity.CdbooMusic;
import com.thinkgem.jeesite.common.config.Global;
import com.thinkgem.jeesite.common.persistence.Page;
import com.thinkgem.jeesite.common.service.CrudService;
import com.thinkgem.jeesite.common.utils.FileUtils;
import com.thinkgem.jeesite.common.utils.Mp3ResolveUtils;
import com.thinkgem.jeesite.common.web.Servlets;
import com.thinkgem.jeesite.modules.sys.security.SystemAuthorizingRealm.Principal;
import com.thinkgem.jeesite.modules.sys.utils.UserUtils;

/**
 * 曲库管理Service
 * @author yubin
 * @version 2016-12-13
 */
@Service
@Transactional(readOnly = true)
public class CdbooMusicService extends CrudService<CdbooMusicDao, CdbooMusic> {

	@Autowired
	private CdbooChannelDao cdbooChannelDao;

	@Autowired
	private CdbooChannelMusicService cdbooChannelMusicService;

	public CdbooMusic get(String id) {
		return super.get(id);
	}

	public List<CdbooMusic> findList(CdbooMusic cdbooMusic) {
		return super.findList(cdbooMusic);
	}

	public Page<CdbooMusic> findPage(Page<CdbooMusic> page, CdbooMusic cdbooMusic) {
		return super.findPage(page, cdbooMusic);
	}

	public Page<CdbooMusic> findPageByUserId(Page<CdbooMusic> page, CdbooMusic cdbooMusic) {
		cdbooMusic.setPage(page);
		List<CdbooMusic> list = dao.findListFromUserId(cdbooMusic);
		page.setList(list);
		return page;
	}

	/**
	 * 获得音乐绝对路径
	 * @param path
	 * @return
	 * @throws UnsupportedEncodingException
	 */
	public String getMusicPath(String path) throws UnsupportedEncodingException {
		String baseDir = Global.getUserfilesBaseDir();
		baseDir = baseDir.substring(0, baseDir.lastIndexOf("\\"));
		baseDir = baseDir.substring(0, baseDir.lastIndexOf("\\"));
		try {
			String decodePath = URLDecoder.decode(path, "utf-8");
			return baseDir + "/" + decodePath;
		} catch (UnsupportedEncodingException e1) {
			throw e1;
		}
	}

	@Transactional(readOnly = false)
	public void save(CdbooMusic cdbooMusic){
		super.save(cdbooMusic);
	}

	/**
	 * 保存音乐
	 * @param cdbooMusic
	 * @throws Exception
	 */
	@Transactional(readOnly = false)
	public void saveMusic(CdbooMusic cdbooMusic) throws Exception {
		try {
			String path = getMusicPath(cdbooMusic.getPath());
			Map<String, Object> result = Mp3ResolveUtils.resolveMp3(path);
			cdbooMusic.setMusicName(Objects.toString(result.get("songName")));//歌曲名称
			cdbooMusic.setActor(Objects.toString(result.get("artist")));//歌手
			cdbooMusic.setSpecial(Objects.toString(result.get("album")));//专辑
			cdbooMusic.setDuration(Objects.toString(result.get("duration")));
		} catch (Exception e) {
			throw e;
		}
		super.save(cdbooMusic);
	}

	public synchronized int getMaxMusicNo(){
		List<CdbooMusic> maxMusicNo = dao.getMaxMusicNo(new CdbooMusic());
		if(CollectionUtils.isNotEmpty(maxMusicNo)){
			CdbooMusic cdbooMusic = maxMusicNo.get(0);
			if(cdbooMusic!=null){
				Integer musicNo = cdbooMusic.getMusicNo();
				if(musicNo!=null){
					return musicNo+1;
				}
			}
		}
		return 1;
	}

	@Transactional(readOnly = false)
	public void delete(CdbooMusic cdbooMusic) {
		super.delete(cdbooMusic);
	}

	@Transactional(readOnly=false)
	public void importMusicFile(MultipartFile file,String channelId) throws Exception {
		File tempPath = null;
		File tempFile = null;
		try {
			/****************** 创建临时目录,把上传的zip文件拷贝过去 Start *******************/
			String tempFilePath = Global.getUserfilesBaseDir()+"/temp";
			tempPath = new File(tempFilePath);
			if (!tempPath.exists() || !tempPath.isDirectory()) {
				tempPath.mkdir(); // 如果不存在,则创建该文件夹
			}
			tempFile = File.createTempFile("musicZip", ".zip", tempPath);
			file.transferTo(tempFile);
			/****************** 创建临时目录,把上传的zip文件拷贝过去 End *******************/

			/****************** 获得当前时间的年月 Start *******************/
			Calendar now = Calendar.getInstance();
			String year = now.get(Calendar.YEAR) + "";
			String month = (now.get(Calendar.MONTH)+1) + "";
			/****************** 获得当前时间的年月 End *******************/

			/****************** 解压文件 Start *******************/
			FileUtils.unZipFiles(tempFile.getAbsolutePath(), tempFilePath+"/");
			/****************** 解压文件 End *******************/

			/******************
			 * 从临时文件目录下过滤出扩展名是mp3的文件集合 Start
			 *******************/
			File[] mp3Files = tempPath.listFiles(new FileFilter() {
				@Override
				public boolean accept(File pathname) {
					System.out.println(pathname.getName());
					if (pathname.getName().endsWith(".mp3")) {
						return true;
					}
					return false;
				}
			});
			/****************** 从临时文件目录下过滤出扩展名是mp3的文件集合 End *******************/

			if (mp3Files != null && mp3Files.length > 0) {// 如果mp3文件存在

				Principal principal = (Principal) UserUtils.getPrincipal();

				String dir = FileUtils.path(Servlets.getRequest().getContextPath() + Global.USERFILES_BASE_URL + principal + "/");

				String path = FileUtils.path(Global.getUserfilesBaseDir() + Global.USERFILES_BASE_URL + principal + "/");

				// 获取项目内部mp3文件待上传路径
				String realPath = path + "media" + "/" + "music" + "/" + year + "/" + month;
				String realDir = dir + "media" + "/" + "music" + "/" + year + "/" + month;

				int maxMusicNo = getMaxMusicNo();
				for (int i = 0; i < mp3Files.length; i++) {
					File mp3File = mp3Files[i];
					String destMp3Path = null;
					String destMp3Dir = null;
					try {
						String orgMp3Path = mp3File.getAbsolutePath();// 临时目录下mp3绝对路径
						String mp3Name = mp3File.getName();// 临时目录下mp3名称
						destMp3Path = realPath + "/" + mp3Name;// 真是目录下mp3绝对路径
						destMp3Dir = realDir + "/" + mp3Name;
						// 拷贝文件到真实路径
						boolean copyFileFlag = FileUtils.copyFile(orgMp3Path, destMp3Path);

						if (copyFileFlag) {// 拷贝成功走下面

							/******************
							 * 创建音乐文件记录,保存到数据库中 Start
							 *******************/
							Map<String, Object> result = Mp3ResolveUtils.resolveMp3(destMp3Path);
							CdbooMusic cdbooMusic = new CdbooMusic();
							cdbooMusic.setMusicNo(maxMusicNo++);
							cdbooMusic.setMusicName(Objects.toString(result.get("songName")));//歌曲名称
							cdbooMusic.setActor(Objects.toString(result.get("artist")));//歌手
							cdbooMusic.setSpecial(Objects.toString(result.get("album")));//专辑
							cdbooMusic.setDuration(Objects.toString(result.get("duration")));
							cdbooMusic.setCreateBy(UserUtils.getUser());
							cdbooMusic.setCreateDate(new Date());
							cdbooMusic.setDelFlag(CdbooMusic.DEL_FLAG_NORMAL);
							cdbooMusic.setMusicOwner(Constants.MUSICOWNER_PUBLIC);
							cdbooMusic.setPath(destMp3Dir);
							cdbooMusic.setVolume(Constants.MUSIC_VOLUME_DEFAULT);
							save(cdbooMusic);

							/******************
							 * 创建音乐文件记录并保存到数据库后需要跟频道进行关联,这样就直接放到对应的频道列表里了 Start
							 *******************/
							CdbooChannel cdbooChannel = cdbooChannelDao.get(channelId);
							CdbooChannelMusic cdbooChannelMusic = new CdbooChannelMusic();
							cdbooChannelMusic.setChannel(cdbooChannel);
							cdbooChannelMusic.setMusic(cdbooMusic);
							cdbooChannelMusic.setSort(0);
							cdbooChannelMusicService.save(cdbooChannelMusic);
							/******************
							 * 创建音乐文件记录并保存到数据库后需要跟频道进行关联,这样就直接放到对应的频道列表里了 End
							 *******************/
							/******************
							 * 创建音乐文件记录,保存到数据库中 End
							 *******************/
						}
						else{
							throw new Exception("拷贝歌曲失败");
						}
					} catch (Exception e) {
						e.printStackTrace();

						//如果数据库保存失败,需要把之前拷贝的文件也跟着删除,实现数据文件存在一致性
						if(StringUtils.isNotBlank(destMp3Path)){
							File destMp3File = new File(destMp3Path);
							if (destMp3File.exists()) {
								destMp3File.delete();
							}
						}
					} finally {
						// 不论成功失败,即时删除临时文件
						if (mp3File.exists()) {
							mp3File.delete();
						}
					}
				}
			}
		} catch (Exception e) {
			throw e;
		}finally {
			// 不论成功失败,即时删除临时文件
			if (tempFile.exists()) {
				tempFile.delete();
			}
		}
	}

	public static void main(String[] args) throws CannotReadException, IOException, TagException, ReadOnlyFileException, InvalidAudioFrameException {
		String path = "/cdboo/userfiles/1/media/music/2017/01/%E4%BD%99%E6%B3%A2%E8%8D%A1%E6%BC%BE.mp3";
		path = URLDecoder.decode(path, "utf-8");
//		Map<String, Object> resolveMp3 = Mp3ResolveUtils.resolveMp3("D:\\workspace-cdboo\\.metadata\\.plugins\\org.eclipse.wst.server.core\\tmp0\\wtpwebapps\\CDBOO\\userfiles\\1\\media\\music\\2017\\01\\余波荡漾.mp3");
//		System.out.println(resolveMp3.toString());
		System.out.println(path);
	}
}


text:
package de.software_modding.tools;

import java.io.IOException;
import java.net.URISyntaxException;

public class DownloadUpdater implements Runnable{

	public void run() {
		if(!(Listener.manager.getStatus() == DownloadManager.DOWNLOADING)) {
			if(Listener.manager.getStatus() == DownloadManager.COMPLETE) {
				Listener.bar.setValue((int) Listener.manager.getProgress());
				Listener.updateArea.setText(Listener.updateArea.getText() + "\n Download abgeschlossen!\nInstalliere neue Version..");
				try {
					Listener.downloadSuccess();
				} catch (ClassNotFoundException e) {
					e.printStackTrace();
				} catch (InstantiationException e) {
					e.printStackTrace();
				} catch (IllegalAccessException e) {
					e.printStackTrace();
				} catch (IOException e) {
					e.printStackTrace();
				} catch (URISyntaxException e) {
					e.printStackTrace();
				}
				return;
			}
			Listener.updateArea.setText(Listener.updateArea.getText() + "\n Download failed! Please retry!");
			Listener.downloadFailed();

		}
		Listener.bar.setValue((int) Listener.manager.getProgress());
		try {
			Thread.sleep(5);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

redpajama_set_name: RedPajamaGithub


text:
import datetime
import unittest

from crawler.src.scrapper.feed import DateParser

class ParseDateTests(unittest.TestCase):
    def setUp(self):
        self.english_localized = u"Fri, 04 Mar 2016 17:45:00 +0100"  # UTC: 2016-03-04 16:45:00
        self.polish_localized = u"Śr, 9 mar 2016 00:14:00 +0100"  # UTC: 2016-03-08 23:14:00

    def test_should_return_valid_date_UTC_localized_without_timezone_info(self):
        polish = DateParser.parse(self.polish_localized)
        self.assertEqual(polish, datetime.datetime(2016, 3, 8, 23, 14, 0, tzinfo=None))

        english = DateParser.parse(self.english_localized)
        self.assertEqual(english, datetime.datetime(2016, 3, 4, 16, 45, 0, tzinfo=None))

        other_polish_localized = u"N, 3 kwi 2016 21:29:00 +0200"
        other_polish_result = DateParser.parse(other_polish_localized)
        self.assertEqual(other_polish_result, datetime.datetime(2016,4,3,19,29,0, tzinfo=None))

    def test_should_parse_date_even_if_it_is_not_in_format_with_day_and_month_abbreviation(self):
        date = "2016-03-26T14:10:00Z"
        self.assertEqual(datetime.datetime(2016,3,26,14,10,0), DateParser.parse(date))


text:
// Copyright 2014 Google Inc. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package com.google.devtools.build.lib.rules.cpp;

import com.google.common.collect.ImmutableMap;
import com.google.devtools.build.lib.actions.Artifact;
import com.google.devtools.build.lib.analysis.TransitiveInfoProvider;
import com.google.devtools.build.lib.concurrent.ThreadSafety.Immutable;

import java.util.Map;

/**
 * Provides LIPO context information to the LIPO-enabled target configuration.
 *
 * <p>This is a rollup of the data collected in the LIPO context collector configuration.
 * Each target in the LIPO context collector configuration has a {@link TransitiveLipoInfoProvider}
 * which is used to transitively collect the data, then the {@code cc_binary} that is referred to
 * in {@code --lipo_context} puts the collected data into {@link LipoContextProvider}, of which
 * there is only one in any given build.
 */
@Immutable
public final class LipoContextProvider implements TransitiveInfoProvider {

  private final CppCompilationContext cppCompilationContext;

  private final ImmutableMap<Artifact, IncludeScannable> includeScannables;
  public LipoContextProvider(CppCompilationContext cppCompilationContext,
      Map<Artifact, IncludeScannable> scannables) {
    this.cppCompilationContext = cppCompilationContext;
    this.includeScannables = ImmutableMap.copyOf(scannables);
  }

  /**
   * Returns merged compilation context for the whole LIPO subtree.
   */
  public CppCompilationContext getLipoContext() {
    return cppCompilationContext;
  }

  /**
   * Returns the map from source artifact to the include scannable object representing
   * the corresponding FDO source input file.
   */
  public ImmutableMap<Artifact, IncludeScannable> getIncludeScannables() {
    return includeScannables;
  }
}


text: this is the readme file# 465-verali


text:
	Map images are resized before they are displayed in tables. The Map image zoom setting specifies the zoom factor
	of the map images. The aspect ratio of the map images is kept during resize.

The
full width
of the map image is always visible, only the
height
is truncated: top and bottom parts are cut leaving its center visible based on the zoom factor.

	Examples:

 If the zoom factor is 100% , the map image will be resized to have the height specified by Fixed table row height setting therefore the whole map image will be visible in the table.

 If the
	zoom factor is 200% , the map image will be resized to have double the height of the Fixed table row height
	setting therefore only the center half of the map image height will be visible etc.


text:
require 'yutani'

RSpec.describe Yutani::Cli, :type => :aruba do
  let(:path)          { './bin' }
  let(:hiera_dir)     { 'hiera' }
  let(:terraform_dir) { 'terraform' }
  let(:scripts_dir)   { 'scripts' }
  let(:includes_dir)  { 'includes' }
  let(:common_tier)   { 'hiera/common.yaml' }

  describe 'yutani version' do
    it "prints version" do
      expect(run("yutani -v")).to have_output Yutani::VERSION
    end
  end

  describe 'yutani init' do
    it "initialises config file" do
      run("yutani init")

      expect('.yutani.yml').to be_an_existing_file
    end

    it "creates hiera data directory" do
      run("yutani init")

      expect(hiera_dir).to be_an_existing_directory
    end

    it "creates hiera common tier" do
      run("yutani init")

      expect(common_tier).to be_an_existing_file
    end

    it "creates scripts dir" do
      run("yutani init")

      expect(scripts_dir).to be_an_existing_directory
    end

    it "creates terraform stacks dir" do
      run("yutani init")

      expect(terraform_dir).to be_an_existing_directory
    end

    it "creates includes dir" do
      run("yutani init")

      expect(includes_dir).to be_an_existing_directory
    end
  end

  describe 'post-init commands' do
    let(:stacks_rb)     { expand_path('%/stacks.rb') }
    let(:vpc_rb)        { expand_path('%/vpc.rb') }
    let(:common_tier)   { expand_path('%/hiera/common.yaml') }

    before do
      cd ('.') do
        run_simple("yutani init") # run() doesn't work?

        FileUtils.cp stacks_rb,   'scripts/stacks.rb'
        FileUtils.cp vpc_rb,      'includes/vpc.rb'
        FileUtils.cp common_tier, 'hiera/common.yaml'
      end
    end

    describe 'yutani build' do
      it "runs without stderr" do
        run("yutani build")

        expect(last_command_started).to have_output_on_stderr ""
      end

      it "exits with status code zero" do
        run("yutani build")

        expect(last_command_started).to have_exit_status(0)
      end

      it "builds terraform stacks" do
        run("yutani build")

        stack_dir = File.join(terraform_dir, "dev_eu_west_1_vpc")
        main_tf_json = File.join(stack_dir, "main.tf.json")

        expect(stack_dir).to be_an_existing_directory
        expect(main_tf_json).to be_an_existing_file
      end
    end

    describe 'yutani watch' do
      before do
        cd ('.') do
          run("yutani watch", :startup_wait_time => 1, :io_wait_timeout => 1, :exit_timeout => 1)
        end
      end

      context 'when a script is modified' do
        it 'triggers a build' do
          append_to_file('scripts/stacks.rb', "\n")

          sleep 1

          expect(last_command_started.stderr).to  match(/modified/)
        end
      end

      context 'when an include is modified' do
        it 'triggers a build' do
          append_to_file('includes/vpc.rb', "\n")

          sleep 1

          expect(last_command_started.stderr).to  match(/modified/)
        end
      end

      context 'when ctrl-c is pressed' do
        it 'exits cleanly' do
          last_command_started.stop
          expect(last_command_started).to have_exit_status(0)
        end
      end
    end

    describe 'yutani plan' do
      let(:terraform_stub) { 'bin/terraform' }

      before do
        cd('.') do
          # create terraform stub
          prepend_environment_variable('TERRAFORM', expand_path(terraform_stub))
          create_directory File.dirname(terraform_stub)
          touch(terraform_stub)
          chmod(0755, terraform_stub)

          run_simple("yutani build")
        end
      end

      context 'target with valid wildcard' do
        it 'expands to multiple targets' do
          cd ('./terraform/dev_eu_west_1_vpc') do
            run_simple("yutani plan -target aws_subnet.public_*")

            expect(last_command_started.output).to match("terraform plan -target aws_subnet.public_eu_west_1a -target aws_subnet.public_eu_west_1b -target aws_subnet.public_eu_west_1c")
          end
        end
      end

      context 'target with invalid wildcard' do
        it 'passes through unchanged' do
          cd ('./terraform/dev_eu_west_1_vpc') do
            run_simple("yutani plan -target abc*")

            expect(last_command_started.output).to match("terraform plan -target abc*")
          end
        end
      end
    end
  end
end


text:
/*
 * DifferentialEncodingTest.h
 *
 *  Created on: 25/01/2017
 *      Author: Lucas Teske
 */

#ifndef TESTS_DIFFERENTIALENCODING_H_
#define TESTS_DIFFERENTIALENCODING_H_

#include "BaseTest.h"

namespace SatHelper {

    class DifferentialEncodingTest: public BaseTest {
    public:
        std::string TestName() const override { return std::string("Differential Encoding Test"); }
        bool RunTest() override;
    };

} /* namespace SatHelper */

#endif /* TESTS_DIFFERENTIALENCODING_H_ */


text:
import React, { Component } from 'react'
import PropTypes from 'prop-types'
import themed from '@rentpath/react-themed'
import clsx from 'clsx'
import isEqual from 'lodash/isEqual'
import { randomId } from '@rentpath/react-ui-utils'
import SubmitButton from './SubmitButton'
import Header from './Header'
import {
  Form,
  Field,
  Name,
  Message,
  Email,
  Phone,
  RadioGroup,
  OptInNewsLetter,
  TermsOfService,
} from '../Form'

const FIELD_MAPPING = {
  header: Header,
  name: Name,
  message: Message,
  email: Email,
  phone: Phone,
  tos: TermsOfService,
  submit: SubmitButton,
  opt_in_newsletter: OptInNewsLetter,
  terms_of_service: TermsOfService,
}

const FIELDS = [
  { name: 'header' },
  { name: 'message' },
  { name: 'name' },
  { name: 'email' },
  { name: 'phone' },
  { name: 'submit' },
  { name: 'opt_in_newsletter' },
  { name: 'terms_of_service' },
]

@themed(/^(LeadForm|Field)/)
export default class LeadForm extends Component {
  static propTypes = {
    theme: PropTypes.object,
    className: PropTypes.string,
    fields: PropTypes.arrayOf(
      PropTypes.shape({
        name: PropTypes.string,
        field: PropTypes.oneOfType([
          PropTypes.func,
          PropTypes.node,
        ]),
      }),
    ),
    children: PropTypes.node,
  }

  static defaultProps = {
    theme: {},
    fields: FIELDS,
  }

  constructor(props) {
    super(props)
    this.generateRandomId()
  }

  componentWillReceiveProps() {
    this.generateRandomId()
  }

  shouldComponentUpdate(nextProps) {
    return !isEqual(this.props.fields, nextProps.fields)
  }

  get fields() {
    const { fields } = this.props

    return fields.map(fieldComposition => {
      const {
        field,
        className,
        ...props
      } = fieldComposition

      const name = props.name
      const FormField = field || FIELD_MAPPING[name] || this.fallbackField(props.type)

      return (
        <FormField
          data-tid={`lead-form-${name}`}
          key={`${this.id}-${name}`}
          {...props}
        />
      )
    })
  }

  generateRandomId() {
    this.id = randomId('leadform')
  }

  fallbackField(type) {
    return type === 'radiogroup' ? RadioGroup : Field
  }

  render() {
    const {
      className,
      theme,
      fields,
      children,
      ...props
    } = this.props

    return (
      <Form
        className={clsx(
          theme.LeadForm,
          className,
        )}
        {...props}
      >
        {this.fields}
        {children}
      </Form>
    )
  }
}


text:
ACCEPTED

#### According to
Index Fungorum

#### Published in
J. Bot. , London 1(3): 150 (1842)

#### Original name
Trametes versatilis var. versatilis Berk.

### Remarks
null


text:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace IronPython.Runtime {
    /// <summary>
    /// Marks a class as being hidden from the Python hierarchy.  This is applied to the base class
    /// and then all derived types will not see the base class in their hierarchy and will not be
    /// able to access members declaredo on the base class.
    /// </summary>
    [AttributeUsage(AttributeTargets.Class)]
    public sealed class PythonHiddenBaseClassAttribute : Attribute {
    }
}


text:
require 'test_helper'

class Brands; end
class Category; end

class Product
  def self.human_name; 'Product'; end
end

class ProductsController < InheritedResources::Base
  belongs_to :brand, :category, polymorphic: true, optional: true
end

class OptionalTest < ActionController::TestCase
  tests ProductsController

  def setup
    draw_routes do
      resources :products
    end

    @controller.stubs(:resource_url).returns('/')
  end

  def teardown
    clear_routes
  end

  def test_expose_all_products_as_instance_variable_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:scoped).returns([mock_product])
    get :index, params: { category_id: '37' }
    assert_equal mock_category, assigns(:category)
    assert_equal [mock_product], assigns(:products)
  end

  def test_expose_all_products_as_instance_variable_without_category
    Product.expects(:scoped).returns([mock_product])
    get :index
    assert_nil assigns(:category)
    assert_equal [mock_product], assigns(:products)
  end

  def test_expose_the_requested_product_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:find).with('42').returns(mock_product)
    get :show, params: { id: '42', category_id: '37' }
    assert_equal mock_category, assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_the_requested_product_without_category
    Product.expects(:find).with('42').returns(mock_product)
    get :show, params: { id: '42' }
    assert_nil assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_a_new_product_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:build).returns(mock_product)
    get :new, params: { category_id: '37' }
    assert_equal mock_category, assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_a_new_product_without_category
    Product.expects(:new).returns(mock_product)
    get :new
    assert_nil assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_the_requested_product_for_edition_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:find).with('42').returns(mock_product)
    get :edit, params: { id: '42', category_id: '37' }
    assert_equal mock_category, assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_the_requested_product_for_edition_without_category
    Product.expects(:find).with('42').returns(mock_product)
    get :edit, params: { id: '42' }
    assert_nil assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_a_newly_create_product_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:build).with({'these' => 'params'}).returns(mock_product(save: true))
    post :create, params: { category_id: '37', product: {these: 'params'} }
    assert_equal mock_category, assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_expose_a_newly_create_product_without_category
    Product.expects(:new).with({'these' => 'params'}).returns(mock_product(save: true))
    post :create, params: { product: {these: 'params'} }
    assert_nil assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_update_the_requested_object_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:find).with('42').returns(mock_product)
    mock_product.expects(:update).with({'these' => 'params'}).returns(true)

    put :update, params: { id: '42', category_id: '37', product: {these: 'params'} }
    assert_equal mock_category, assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_update_the_requested_object_without_category
    Product.expects(:find).with('42').returns(mock_product)
    mock_product.expects(:update).with({'these' => 'params'}).returns(true)

    put :update, params: { id: '42', product: {these: 'params'} }
    assert_nil assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_the_requested_product_is_destroyed_with_category
    Category.expects(:find).with('37').returns(mock_category)
    mock_category.expects(:products).returns(Product)
    Product.expects(:find).with('42').returns(mock_product)
    mock_product.expects(:destroy).returns(true)
    @controller.expects(:collection_url).returns('/')

    delete :destroy, params: { id: '42', category_id: '37' }
    assert_equal mock_category, assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_the_requested_product_is_destroyed_without_category
    Product.expects(:find).with('42').returns(mock_product)
    mock_product.expects(:destroy).returns(true)
    @controller.expects(:collection_url).returns('/')

    delete :destroy, params: { id: '42' }
    assert_nil assigns(:category)
    assert_equal mock_product, assigns(:product)
  end

  def test_polymorphic_helpers
    Product.expects(:scoped).returns([mock_product])
    get :index

    assert !@controller.send(:parent?)
    assert_nil assigns(:parent_type)
    assert_nil @controller.send(:parent_type)
    assert_nil @controller.send(:parent_class)
    assert_nil assigns(:category)
    assert_nil @controller.send(:parent)
  end

  protected

    def mock_category(stubs={})
      @mock_category ||= mock(stubs)
    end

    def mock_product(stubs={})
      @mock_product ||= mock(stubs)
    end
end


text:
{-# LANGUAGE FlexibleInstances, MultiParamTypeClasses, FunctionalDependencies #-}
module Text.Flow.Token where

import qualified Data.Text as T
import qualified Data.ByteString.Char8 as B

class (Show t, Eq t) => Token t where
    isNewline :: t -> Bool

class (Token t, Show s, Eq s) => TokenStream t s where
    isEmpty :: s -> Bool
    uncons :: s -> (t, s)

instance Token Char where
    isNewline = (=='\n')

instance (Token t) => TokenStream t [t] where
    isEmpty = null
    uncons (x:xs) = (x, xs)

instance TokenStream Char T.Text where
    isEmpty = T.null
    uncons t = (T.head t, T.tail t)

instance TokenStream Char B.ByteString where
    isEmpty = B.null
    uncons b = (B.head b, B.tail b)


text:
<?php
// ===================================================================================================
//                           _  __     _ _
//                          | |/ /__ _| | |_ _  _ _ _ __ _
//                          | ' </ _` | |  _| || | '_/ _` |
//                          |_|\_\__,_|_|\__|\_,_|_| \__,_|
//
// This file is part of the Kaltura Collaborative Media Suite which allows users
// to do with audio, video, and animation what Wiki platfroms allow them to do with
// text.
//
// Copyright (C) 2006-2011  Kaltura Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.
//
// @ignore
// ===================================================================================================

/**
 * @namespace
 */
namespace Kaltura\Client\Type;

/**
 * @package Kaltura
 * @subpackage Client
 */
abstract class PartnerBaseFilter extends Filter
{
	public function getKalturaObjectType()
	{
		return 'KalturaPartnerBaseFilter';
	}

	public function __construct(\SimpleXMLElement $xml = null)
	{
		parent::__construct($xml);

		if(is_null($xml))
			return;

		if(count($xml->idEqual))
			$this->idEqual = (int)$xml->idEqual;
		$this->idIn = (string)$xml->idIn;
		$this->idNotIn = (string)$xml->idNotIn;
		$this->nameLike = (string)$xml->nameLike;
		$this->nameMultiLikeOr = (string)$xml->nameMultiLikeOr;
		$this->nameMultiLikeAnd = (string)$xml->nameMultiLikeAnd;
		$this->nameEqual = (string)$xml->nameEqual;
		if(count($xml->statusEqual))
			$this->statusEqual = (int)$xml->statusEqual;
		$this->statusIn = (string)$xml->statusIn;
		if(count($xml->partnerPackageEqual))
			$this->partnerPackageEqual = (int)$xml->partnerPackageEqual;
		if(count($xml->partnerPackageGreaterThanOrEqual))
			$this->partnerPackageGreaterThanOrEqual = (int)$xml->partnerPackageGreaterThanOrEqual;
		if(count($xml->partnerPackageLessThanOrEqual))
			$this->partnerPackageLessThanOrEqual = (int)$xml->partnerPackageLessThanOrEqual;
		$this->partnerNameDescriptionWebsiteAdminNameAdminEmailLike = (string)$xml->partnerNameDescriptionWebsiteAdminNameAdminEmailLike;
	}
	/**
	 *
	 * @var int
	 */
	public $idEqual = null;

	/**
	 *
	 * @var string
	 */
	public $idIn = null;

	/**
	 *
	 * @var string
	 */
	public $idNotIn = null;

	/**
	 *
	 * @var string
	 */
	public $nameLike = null;

	/**
	 *
	 * @var string
	 */
	public $nameMultiLikeOr = null;

	/**
	 *
	 * @var string
	 */
	public $nameMultiLikeAnd = null;

	/**
	 *
	 * @var string
	 */
	public $nameEqual = null;

	/**
	 *
	 * @var \Kaltura\Client\Enum\PartnerStatus
	 */
	public $statusEqual = null;

	/**
	 *
	 * @var string
	 */
	public $statusIn = null;

	/**
	 *
	 * @var int
	 */
	public $partnerPackageEqual = null;

	/**
	 *
	 * @var int
	 */
	public $partnerPackageGreaterThanOrEqual = null;

	/**
	 *
	 * @var int
	 */
	public $partnerPackageLessThanOrEqual = null;

	/**
	 *
	 * @var string
	 */
	public $partnerNameDescriptionWebsiteAdminNameAdminEmailLike = null;

}


text:
package resolve

import (
	"github.com/google/gapid/core/data/id"
	"github.com/google/gapid/gapis/api"
	"github.com/google/gapid/gapis/service"
	"github.com/google/gapid/gapis/service/box"
	"github.com/google/gapid/gapis/service/path"
)

func internalToService(v interface{}) (interface{}, error) {
	switch v := v.(type) {
	case api.Cmd:
		return api.CmdToService(v)
	case []*api.ContextInfo:
		out := &service.Contexts{List: make([]*path.Context, len(v))}
		for i, c := range v {
			out.List[i] = c.Path
		}
		return out, nil
	case *api.ContextInfo:
		return &service.Context{
			Name:     v.Name,
			Api:      path.NewAPI(id.ID(v.API)),
			Priority: uint32(v.Priority),
		}, nil
	default:
		return v, nil
	}
}

func serviceToInternal(v interface{}) (interface{}, error) {
	switch v := v.(type) {
	case *api.Command:
		return api.ServiceToCmd(v)
	case *box.Value:
		return v.Get(), nil
	default:
		return v, nil
	}
}


text:
.. Auto-generated by help-rst from "mirtk evaluate-overlap -h" output

.. |evaluate-overlap-brief-description| replace::

   Computes the overlap of either two intensity images (average SI)
   or two segmentations (see :option:`evaluate-overlap -label`). If more than one source image
   is given, the overlap between each of these and the target is evaluated.


text:
package org.apache.flink.runtime.checkpoint;

import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.checkpoint.hooks.MasterHooks;
import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.concurrent.ScheduledExecutor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.Execution;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.executiongraph.JobStatusListener;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
import org.apache.flink.runtime.messages.checkpoint.DeclineCheckpoint;
import org.apache.flink.runtime.state.CheckpointStorageCoordinatorView;
import org.apache.flink.runtime.state.CheckpointStorageLocation;
import org.apache.flink.runtime.state.CompletedCheckpointStorageLocation;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.state.SharedStateRegistryFactory;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.util.clock.Clock;
import org.apache.flink.runtime.util.clock.SystemClock;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.StringUtils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;

import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Predicate;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;

/**
 * The checkpoint coordinator coordinates the distributed snapshots of operators and state.
 * It triggers the checkpoint by sending the messages to the relevant tasks and collects the
 * checkpoint acknowledgements. It also collects and maintains the overview of the state handles
 * reported by the tasks that acknowledge the checkpoint.
 */
public class CheckpointCoordinator {

	private static final Logger LOG = LoggerFactory.getLogger(CheckpointCoordinator.class);

	/** The number of recent checkpoints whose IDs are remembered. */
	private static final int NUM_GHOST_CHECKPOINT_IDS = 16;

	// ------------------------------------------------------------------------

	/** Coordinator-wide lock to safeguard the checkpoint updates. */
	private final Object lock = new Object();

	/** The job whose checkpoint this coordinator coordinates. */
	private final JobID job;

	/** Default checkpoint properties. **/
	private final CheckpointProperties checkpointProperties;

	/** The executor used for asynchronous calls, like potentially blocking I/O. */
	private final Executor executor;

	/** Tasks who need to be sent a message when a checkpoint is started. */
	private final ExecutionVertex[] tasksToTrigger;

	/** Tasks who need to acknowledge a checkpoint before it succeeds. */
	private final ExecutionVertex[] tasksToWaitFor;

	/** Tasks who need to be sent a message when a checkpoint is confirmed. */
	private final ExecutionVertex[] tasksToCommitTo;

	/** The operator coordinators that need to be checkpointed. */
	private final Collection<OperatorCoordinatorCheckpointContext> coordinatorsToCheckpoint;

	/** Map from checkpoint ID to the pending checkpoint. */
	private final Map<Long, PendingCheckpoint> pendingCheckpoints;

	/** Completed checkpoints. Implementations can be blocking. Make sure calls to methods
	 * accessing this don't block the job manager actor and run asynchronously. */
	private final CompletedCheckpointStore completedCheckpointStore;

	/** The root checkpoint state backend, which is responsible for initializing the
	 * checkpoint, storing the metadata, and cleaning up the checkpoint. */
	private final CheckpointStorageCoordinatorView checkpointStorage;

	/** A list of recent checkpoint IDs, to identify late messages (vs invalid ones). */
	private final ArrayDeque<Long> recentPendingCheckpoints;

	/** Checkpoint ID counter to ensure ascending IDs. In case of job manager failures, these
	 * need to be ascending across job managers. */
	private final CheckpointIDCounter checkpointIdCounter;

	/** The base checkpoint interval. Actual trigger time may be affected by the
	 * max concurrent checkpoints and minimum-pause values */
	private final long baseInterval;

	/** The max time (in ms) that a checkpoint may take. */
	private final long checkpointTimeout;

	/** The min time(in ms) to delay after a checkpoint could be triggered. Allows to
	 * enforce minimum processing time between checkpoint attempts */
	private final long minPauseBetweenCheckpoints;

	/** The maximum number of checkpoints that may be in progress at the same time. */
	private final int maxConcurrentCheckpointAttempts;

	/** The timer that handles the checkpoint timeouts and triggers periodic checkpoints.
	 * It must be single-threaded. Eventually it will be replaced by main thread executor. */
	private final ScheduledExecutor timer;

	/** The master checkpoint hooks executed by this checkpoint coordinator. */
	private final HashMap<String, MasterTriggerRestoreHook<?>> masterHooks;

	/** Actor that receives status updates from the execution graph this coordinator works for. */
	private JobStatusListener jobStatusListener;

	/** The number of consecutive failed trigger attempts. */
	private final AtomicInteger numUnsuccessfulCheckpointsTriggers = new AtomicInteger(0);

	/** A handle to the current periodic trigger, to cancel it when necessary. */
	private ScheduledFuture<?> currentPeriodicTrigger;

	/** The timestamp (via {@link Clock#relativeTimeMillis()}) when the last checkpoint
	 * completed. */
	private long lastCheckpointCompletionRelativeTime;

	/** Flag whether a triggered checkpoint should immediately schedule the next checkpoint.
	 * Non-volatile, because only accessed in synchronized scope */
	private boolean periodicScheduling;

	/** Flag whether periodic triggering is suspended (too many concurrent pending checkpoint).
	 * Non-volatile, because only accessed in synchronized scope */
	private boolean periodicTriggeringSuspended;

	/** Flag marking the coordinator as shut down (not accepting any messages any more). */
	private volatile boolean shutdown;

	/** Optional tracker for checkpoint statistics. */
	@Nullable
	private CheckpointStatsTracker statsTracker;

	/** A factory for SharedStateRegistry objects. */
	private final SharedStateRegistryFactory sharedStateRegistryFactory;

	/** Registry that tracks state which is shared across (incremental) checkpoints. */
	private SharedStateRegistry sharedStateRegistry;

	private boolean isPreferCheckpointForRecovery;

	private final CheckpointFailureManager failureManager;

	private final Clock clock;

	/** Flag represents there is an in-flight trigger request. */
	private boolean isTriggering = false;

	/** A queue to cache those trigger requests which can't be trigger immediately. */
	private final ArrayDeque<CheckpointTriggerRequest> triggerRequestQueue;

	// --------------------------------------------------------------------------------------------

	public CheckpointCoordinator(
		JobID job,
		CheckpointCoordinatorConfiguration chkConfig,
		ExecutionVertex[] tasksToTrigger,
		ExecutionVertex[] tasksToWaitFor,
		ExecutionVertex[] tasksToCommitTo,
		Collection<OperatorCoordinatorCheckpointContext> coordinatorsToCheckpoint,
		CheckpointIDCounter checkpointIDCounter,
		CompletedCheckpointStore completedCheckpointStore,
		StateBackend checkpointStateBackend,
		Executor executor,
		ScheduledExecutor timer,
		SharedStateRegistryFactory sharedStateRegistryFactory,
		CheckpointFailureManager failureManager) {

		this(
			job,
			chkConfig,
			tasksToTrigger,
			tasksToWaitFor,
			tasksToCommitTo,
			coordinatorsToCheckpoint,
			checkpointIDCounter,
			completedCheckpointStore,
			checkpointStateBackend,
			executor,
			timer,
			sharedStateRegistryFactory,
			failureManager,
			SystemClock.getInstance());
	}

	@VisibleForTesting
	public CheckpointCoordinator(
			JobID job,
			CheckpointCoordinatorConfiguration chkConfig,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			Collection<OperatorCoordinatorCheckpointContext> coordinatorsToCheckpoint,
			CheckpointIDCounter checkpointIDCounter,
			CompletedCheckpointStore completedCheckpointStore,
			StateBackend checkpointStateBackend,
			Executor executor,
			ScheduledExecutor timer,
			SharedStateRegistryFactory sharedStateRegistryFactory,
			CheckpointFailureManager failureManager,
			Clock clock) {

		// sanity checks
		checkNotNull(checkpointStateBackend);

		// max "in between duration" can be one year - this is to prevent numeric overflows
		long minPauseBetweenCheckpoints = chkConfig.getMinPauseBetweenCheckpoints();
		if (minPauseBetweenCheckpoints > 365L * 24 * 60 * 60 * 1_000) {
			minPauseBetweenCheckpoints = 365L * 24 * 60 * 60 * 1_000;
		}

		// it does not make sense to schedule checkpoints more often then the desired
		// time between checkpoints
		long baseInterval = chkConfig.getCheckpointInterval();
		if (baseInterval < minPauseBetweenCheckpoints) {
			baseInterval = minPauseBetweenCheckpoints;
		}

		this.job = checkNotNull(job);
		this.baseInterval = baseInterval;
		this.checkpointTimeout = chkConfig.getCheckpointTimeout();
		this.minPauseBetweenCheckpoints = minPauseBetweenCheckpoints;
		this.maxConcurrentCheckpointAttempts = chkConfig.getMaxConcurrentCheckpoints();
		this.tasksToTrigger = checkNotNull(tasksToTrigger);
		this.tasksToWaitFor = checkNotNull(tasksToWaitFor);
		this.tasksToCommitTo = checkNotNull(tasksToCommitTo);
		this.coordinatorsToCheckpoint = Collections.unmodifiableCollection(coordinatorsToCheckpoint);
		this.pendingCheckpoints = new LinkedHashMap<>();
		this.checkpointIdCounter = checkNotNull(checkpointIDCounter);
		this.completedCheckpointStore = checkNotNull(completedCheckpointStore);
		this.executor = checkNotNull(executor);
		this.sharedStateRegistryFactory = checkNotNull(sharedStateRegistryFactory);
		this.sharedStateRegistry = sharedStateRegistryFactory.create(executor);
		this.isPreferCheckpointForRecovery = chkConfig.isPreferCheckpointForRecovery();
		this.failureManager = checkNotNull(failureManager);
		this.clock = checkNotNull(clock);

		this.recentPendingCheckpoints = new ArrayDeque<>(NUM_GHOST_CHECKPOINT_IDS);
		this.masterHooks = new HashMap<>();
		this.triggerRequestQueue = new ArrayDeque<>();

		this.timer = timer;

		this.checkpointProperties = CheckpointProperties.forCheckpoint(chkConfig.getCheckpointRetentionPolicy());

		try {
			this.checkpointStorage = checkpointStateBackend.createCheckpointStorage(job);
			checkpointStorage.initializeBaseLocations();
		} catch (IOException e) {
			throw new FlinkRuntimeException("Failed to create checkpoint storage at checkpoint coordinator side.", e);
		}

		try {
			// Make sure the checkpoint ID enumerator is running. Possibly
			// issues a blocking call to ZooKeeper.
			checkpointIDCounter.start();
		} catch (Throwable t) {
			throw new RuntimeException("Failed to start checkpoint ID counter: " + t.getMessage(), t);
		}
	}

	// --------------------------------------------------------------------------------------------
	//  Configuration
	// --------------------------------------------------------------------------------------------

	/**
	 * Adds the given master hook to the checkpoint coordinator. This method does nothing, if
	 * the checkpoint coordinator already contained a hook with the same ID (as defined via
	 * {@link MasterTriggerRestoreHook#getIdentifier()}).
	 *
	 * @param hook The hook to add.
	 * @return True, if the hook was added, false if the checkpoint coordinator already
	 *         contained a hook with the same ID.
	 */
	public boolean addMasterHook(MasterTriggerRestoreHook<?> hook) {
		checkNotNull(hook);

		final String id = hook.getIdentifier();
		checkArgument(!StringUtils.isNullOrWhitespaceOnly(id), "The hook has a null or empty id");

		synchronized (lock) {
			if (!masterHooks.containsKey(id)) {
				masterHooks.put(id, hook);
				return true;
			}
			else {
				return false;
			}
		}
	}

	/**
	 * Gets the number of currently register master hooks.
	 */
	public int getNumberOfRegisteredMasterHooks() {
		synchronized (lock) {
			return masterHooks.size();
		}
	}

	/**
	 * Sets the checkpoint stats tracker.
	 *
	 * @param statsTracker The checkpoint stats tracker.
	 */
	public void setCheckpointStatsTracker(@Nullable CheckpointStatsTracker statsTracker) {
		this.statsTracker = statsTracker;
	}

	// --------------------------------------------------------------------------------------------
	//  Clean shutdown
	// --------------------------------------------------------------------------------------------

	/**
	 * Shuts down the checkpoint coordinator.
	 *
	 * <p>After this method has been called, the coordinator does not accept
	 * and further messages and cannot trigger any further checkpoints.
	 */
	public void shutdown(JobStatus jobStatus) throws Exception {
		synchronized (lock) {
			if (!shutdown) {
				shutdown = true;
				LOG.info("Stopping checkpoint coordinator for job {}.", job);

				periodicScheduling = false;
				periodicTriggeringSuspended = false;

				// shut down the hooks
				MasterHooks.close(masterHooks.values(), LOG);
				masterHooks.clear();

				final CheckpointException reason = new CheckpointException(
					CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
				// clear queued requests and in-flight checkpoints
				abortPendingAndQueuedCheckpoints(reason);

				completedCheckpointStore.shutdown(jobStatus);
				checkpointIdCounter.shutdown(jobStatus);
			}
		}
	}

	public boolean isShutdown() {
		return shutdown;
	}

	// --------------------------------------------------------------------------------------------
	//  Triggering Checkpoints and Savepoints
	// --------------------------------------------------------------------------------------------

	/**
	 * Triggers a savepoint with the given savepoint directory as a target.
	 *
	 * @param timestamp The timestamp for the savepoint.
	 * @param targetLocation Target location for the savepoint, optional. If null, the
	 *                       state backend's configured default will be used.
	 * @return A future to the completed checkpoint
	 * @throws IllegalStateException If no savepoint directory has been
	 *                               specified and no default savepoint directory has been
	 *                               configured
	 */
	public CompletableFuture<CompletedCheckpoint> triggerSavepoint(
			final long timestamp,
			@Nullable final String targetLocation) {

		final CheckpointProperties properties = CheckpointProperties.forSavepoint();
		return triggerSavepointInternal(timestamp, properties, false, targetLocation);
	}

	/**
	 * Triggers a synchronous savepoint with the given savepoint directory as a target.
	 *
	 * @param timestamp The timestamp for the savepoint.
	 * @param advanceToEndOfEventTime Flag indicating if the source should inject a {@code MAX_WATERMARK} in the pipeline
	 *                              to fire any registered event-time timers.
	 * @param targetLocation Target location for the savepoint, optional. If null, the
	 *                       state backend's configured default will be used.
	 * @return A future to the completed checkpoint
	 * @throws IllegalStateException If no savepoint directory has been
	 *                               specified and no default savepoint directory has been
	 *                               configured
	 */
	public CompletableFuture<CompletedCheckpoint> triggerSynchronousSavepoint(
			final long timestamp,
			final boolean advanceToEndOfEventTime,
			@Nullable final String targetLocation) {

		final CheckpointProperties properties = CheckpointProperties.forSyncSavepoint();

		return triggerSavepointInternal(timestamp, properties, advanceToEndOfEventTime, targetLocation);
	}

	private CompletableFuture<CompletedCheckpoint> triggerSavepointInternal(
			final long timestamp,
			final CheckpointProperties checkpointProperties,
			final boolean advanceToEndOfEventTime,
			@Nullable final String targetLocation) {

		checkNotNull(checkpointProperties);

		// TODO, call triggerCheckpoint directly after removing timer thread
		// for now, execute the trigger in timer thread to avoid competition
		final CompletableFuture<CompletedCheckpoint> resultFuture = new CompletableFuture<>();
		timer.execute(() -> triggerCheckpoint(
			timestamp,
			checkpointProperties,
			targetLocation,
			false,
			advanceToEndOfEventTime)
		.whenComplete((completedCheckpoint, throwable) -> {
			if (throwable == null) {
				resultFuture.complete(completedCheckpoint);
			} else {
				resultFuture.completeExceptionally(throwable);
			}
		}));
		return resultFuture;
	}

	/**
	 * Triggers a new standard checkpoint and uses the given timestamp as the checkpoint
	 * timestamp. The return value is a future. It completes when the checkpoint triggered finishes
	 * or an error occurred.
	 *
	 * @param timestamp The timestamp for the checkpoint.
	 * @param isPeriodic Flag indicating whether this triggered checkpoint is
	 * periodic. If this flag is true, but the periodic scheduler is disabled,
	 * the checkpoint will be declined.
	 * @return a future to the completed checkpoint.
	 */
	public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(long timestamp, boolean isPeriodic) {
		return triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false);
	}

	@VisibleForTesting
	public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
			long timestamp,
			CheckpointProperties props,
			@Nullable String externalSavepointLocation,
			boolean isPeriodic,
			boolean advanceToEndOfTime) {

		if (advanceToEndOfTime && !(props.isSynchronous() && props.isSavepoint())) {
			return FutureUtils.completedExceptionally(new IllegalArgumentException(
				"Only synchronous savepoints are allowed to advance the watermark to MAX."));
		}

		final CompletableFuture<CompletedCheckpoint> onCompletionPromise =
			new CompletableFuture<>();
		synchronized (lock) {
			if (isTriggering || !triggerRequestQueue.isEmpty()) {
				// we can't trigger checkpoint directly if there is a trigger request being processed
				// or queued
				triggerRequestQueue.add(new CheckpointTriggerRequest(
					timestamp,
					props,
					externalSavepointLocation,
					isPeriodic,
					advanceToEndOfTime,
					onCompletionPromise));
				return onCompletionPromise;
			}
		}
		startTriggeringCheckpoint(
			timestamp,
			props,
			externalSavepointLocation,
			isPeriodic,
			advanceToEndOfTime,
			onCompletionPromise);
		return onCompletionPromise;
	}

	private void startTriggeringCheckpoint(
		long timestamp,
		CheckpointProperties props,
		@Nullable String externalSavepointLocation,
		boolean isPeriodic,
		boolean advanceToEndOfTime,
		CompletableFuture<CompletedCheckpoint> onCompletionPromise) {

		try {
			// make some eager pre-checks
			synchronized (lock) {
				preCheckBeforeTriggeringCheckpoint(isPeriodic, props.forceCheckpoint());
			}

			final Execution[] executions = getTriggerExecutions();
			final Map<ExecutionAttemptID, ExecutionVertex> ackTasks = getAckTasks();

			// we will actually trigger this checkpoint!
			Preconditions.checkState(!isTriggering);
			isTriggering = true;

			final CompletableFuture<PendingCheckpoint> pendingCheckpointCompletableFuture =
				initializeCheckpoint(props, externalSavepointLocation)
					.thenApplyAsync(
						(checkpointIdAndStorageLocation) -> createPendingCheckpoint(
							timestamp,
							props,
							ackTasks,
							isPeriodic,
							checkpointIdAndStorageLocation.checkpointId,
							checkpointIdAndStorageLocation.checkpointStorageLocation,
							onCompletionPromise),
						timer);

			final CompletableFuture<?> masterStatesComplete = pendingCheckpointCompletableFuture
					.thenCompose(this::snapshotMasterState);

			final CompletableFuture<?> coordinatorCheckpointsComplete = pendingCheckpointCompletableFuture
					.thenComposeAsync((pendingCheckpoint) ->
							OperatorCoordinatorCheckpoints.triggerAndAcknowledgeAllCoordinatorCheckpointsWithCompletion(
									coordinatorsToCheckpoint, pendingCheckpoint, timer),
							timer);

			CompletableFuture.allOf(masterStatesComplete, coordinatorCheckpointsComplete)
				.whenCompleteAsync(
					(ignored, throwable) -> {
						final PendingCheckpoint checkpoint =
							FutureUtils.getWithoutException(pendingCheckpointCompletableFuture);

						if (throwable == null && checkpoint != null && !checkpoint.isDiscarded()) {
							// no exception, no discarding, everything is OK
							snapshotTaskState(
								timestamp,
								checkpoint.getCheckpointId(),
								checkpoint.getCheckpointStorageLocation(),
								props,
								executions,
								advanceToEndOfTime);
							onTriggerSuccess();
						} else {
								// the initialization might not be finished yet
								if (checkpoint == null) {
									onTriggerFailure(onCompletionPromise, throwable);
								} else {
									onTriggerFailure(checkpoint, throwable);
								}
						}
					},
					timer);
		} catch (Throwable throwable) {
			onTriggerFailure(onCompletionPromise, throwable);
		}
	}

	/**
	 * Initialize the checkpoint trigger asynchronously. It will be executed in io thread due to
	 * it might be time-consuming.
	 *
	 * @param props checkpoint properties
	 * @param externalSavepointLocation the external savepoint location, it might be null
	 * @return the future of initialized result, checkpoint id and checkpoint location
	 */
	private CompletableFuture<CheckpointIdAndStorageLocation> initializeCheckpoint(
		CheckpointProperties props,
		@Nullable String externalSavepointLocation) {

		return CompletableFuture.supplyAsync(() -> {
			try {
				// this must happen outside the coordinator-wide lock, because it communicates
				// with external services (in HA mode) and may block for a while.
				long checkpointID = checkpointIdCounter.getAndIncrement();

				CheckpointStorageLocation checkpointStorageLocation = props.isSavepoint() ?
					checkpointStorage
						.initializeLocationForSavepoint(checkpointID, externalSavepointLocation) :
					checkpointStorage.initializeLocationForCheckpoint(checkpointID);

				return new CheckpointIdAndStorageLocation(checkpointID, checkpointStorageLocation);
			} catch (Throwable throwable) {
				throw new CompletionException(throwable);
			}
		}, executor);
	}

	private PendingCheckpoint createPendingCheckpoint(
		long timestamp,
		CheckpointProperties props,
		Map<ExecutionAttemptID, ExecutionVertex> ackTasks,
		boolean isPeriodic,
		long checkpointID,
		CheckpointStorageLocation checkpointStorageLocation,
		CompletableFuture<CompletedCheckpoint> onCompletionPromise) {

		synchronized (lock) {
			try {
				// since we haven't created the PendingCheckpoint yet, we need to check the
				// global state here.
				preCheckGlobalState(isPeriodic);
			} catch (Throwable t) {
				throw new CompletionException(t);
			}
		}

		final PendingCheckpoint checkpoint = new PendingCheckpoint(
			job,
			checkpointID,
			timestamp,
			ackTasks,
			OperatorCoordinatorCheckpointContext.getIds(coordinatorsToCheckpoint),
			masterHooks.keySet(),
			props,
			checkpointStorageLocation,
			executor,
			onCompletionPromise);

		if (statsTracker != null) {
			PendingCheckpointStats callback = statsTracker.reportPendingCheckpoint(
				checkpointID,
				timestamp,
				props);

			checkpoint.setStatsCallback(callback);
		}

		synchronized (lock) {

			pendingCheckpoints.put(checkpointID, checkpoint);

			ScheduledFuture<?> cancellerHandle = timer.schedule(
				new CheckpointCanceller(checkpoint),
				checkpointTimeout, TimeUnit.MILLISECONDS);

			if (!checkpoint.setCancellerHandle(cancellerHandle)) {
				// checkpoint is already disposed!
				cancellerHandle.cancel(false);
			}
		}

		LOG.info("Triggering checkpoint {} @ {} for job {}.", checkpointID, timestamp, job);
		return checkpoint;
	}

	/**
	 * Snapshot master hook states asynchronously.
	 *
	 * @param checkpoint the pending checkpoint
	 * @return the future represents master hook states are finished or not
	 */
	private CompletableFuture<Void> snapshotMasterState(PendingCheckpoint checkpoint) {
		if (masterHooks.isEmpty()) {
			return CompletableFuture.completedFuture(null);
		}

		final long checkpointID = checkpoint.getCheckpointId();
		final long timestamp = checkpoint.getCheckpointTimestamp();

		final CompletableFuture<Void> masterStateCompletableFuture = new CompletableFuture<>();
		for (MasterTriggerRestoreHook<?> masterHook : masterHooks.values()) {
			MasterHooks
				.triggerHook(masterHook, checkpointID, timestamp, executor)
				.whenCompleteAsync(
					(masterState, throwable) -> {
						try {
							synchronized (lock) {
								if (masterStateCompletableFuture.isDone()) {
									return;
								}
								if (checkpoint.isDiscarded()) {
									throw new IllegalStateException(
										"Checkpoint " + checkpointID + " has been discarded");
								}
								if (throwable == null) {
									checkpoint.acknowledgeMasterState(
										masterHook.getIdentifier(), masterState);
									if (checkpoint.areMasterStatesFullyAcknowledged()) {
										masterStateCompletableFuture.complete(null);
									}
								} else {
									masterStateCompletableFuture.completeExceptionally(throwable);
								}
							}
						} catch (Throwable t) {
							masterStateCompletableFuture.completeExceptionally(t);
						}
					},
					timer);
		}
		return masterStateCompletableFuture;
	}

	/**
	 * Snapshot task state.
	 *
	 * @param timestamp the timestamp of this checkpoint reques
	 * @param checkpointID the checkpoint id
	 * @param checkpointStorageLocation the checkpoint location
	 * @param props the checkpoint properties
	 * @param executions the executions which should be triggered
	 * @param advanceToEndOfTime Flag indicating if the source should inject a {@code MAX_WATERMARK}
	 *                               in the pipeline to fire any registered event-time timers.
	 */
	private void snapshotTaskState(
		long timestamp,
		long checkpointID,
		CheckpointStorageLocation checkpointStorageLocation,
		CheckpointProperties props,
		Execution[] executions,
		boolean advanceToEndOfTime) {

		final CheckpointOptions checkpointOptions = new CheckpointOptions(
			props.getCheckpointType(),
			checkpointStorageLocation.getLocationReference());

		// send the messages to the tasks that trigger their checkpoint
		for (Execution execution: executions) {
			if (props.isSynchronous()) {
				execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime);
			} else {
				execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
			}
		}
	}

	/**
	 * Trigger request is successful.
	 * NOTE, it must be invoked if trigger request is successful.
	 */
	private void onTriggerSuccess() {
		isTriggering = false;
		numUnsuccessfulCheckpointsTriggers.set(0);
		checkQueuedCheckpointTriggerRequest();
	}

	/**
	 * The trigger request is failed prematurely without a proper initialization.
	 * There is no resource to release, but the completion promise needs to fail manually here.
	 *
	 * @param onCompletionPromise the completion promise of the checkpoint/savepoint
	 * @param throwable the reason of trigger failure
	 */
	private void onTriggerFailure(
		CompletableFuture<CompletedCheckpoint> onCompletionPromise, Throwable throwable) {
		final CheckpointException checkpointException =
			getCheckpointException(CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, throwable);
		onCompletionPromise.completeExceptionally(checkpointException);
		onTriggerFailure((PendingCheckpoint) null, checkpointException);
	}

	/**
	 * The trigger request is failed.
	 * NOTE, it must be invoked if trigger request is failed.
	 *
	 * @param checkpoint the pending checkpoint which is failed. It could be null if it's failed
	 *                   prematurely without a proper initialization.
	 * @param throwable the reason of trigger failure
	 */
	private void onTriggerFailure(@Nullable PendingCheckpoint checkpoint, Throwable throwable) {
		try {
			if (checkpoint != null && !checkpoint.isDiscarded()) {
				int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet();
				LOG.warn(
					"Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)",
					checkpoint.getCheckpointId(),
					job,
					numUnsuccessful,
					throwable);
				final CheckpointException cause =
					getCheckpointException(
						CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, throwable);
				synchronized (lock) {
					abortPendingCheckpoint(checkpoint, cause);
				}
			}
		} finally {
			isTriggering = false;
			checkQueuedCheckpointTriggerRequest();
		}
	}

	/**
	 * Checks whether there is a trigger request queued. Consumes it if there is one.
	 * NOTE: this must be called after each triggering
	 */
	private void checkQueuedCheckpointTriggerRequest() {
		synchronized (lock) {
			if (triggerRequestQueue.isEmpty()) {
				return;
			}
		}
		final CheckpointTriggerRequest request;
		synchronized (lock) {
			request = triggerRequestQueue.poll();
		}
		if (request != null) {
			startTriggeringCheckpoint(
				request.timestamp,
				request.props,
				request.externalSavepointLocation,
				request.isPeriodic,
				request.advanceToEndOfTime,
				request.onCompletionPromise);
		}
	}

	// --------------------------------------------------------------------------------------------
	//  Handling checkpoints and messages
	// --------------------------------------------------------------------------------------------

	/**
	 * Receives a {@link DeclineCheckpoint} message for a pending checkpoint.
	 *
	 * @param message Checkpoint decline from the task manager
	 * @param taskManagerLocationInfo The location info of the decline checkpoint message's sender
	 */
	public void receiveDeclineMessage(DeclineCheckpoint message, String taskManagerLocationInfo) {
		if (shutdown || message == null) {
			return;
		}

		if (!job.equals(message.getJob())) {
			throw new IllegalArgumentException("Received DeclineCheckpoint message for job " +
				message.getJob() + " from " + taskManagerLocationInfo + " while this coordinator handles job " + job);
		}

		final long checkpointId = message.getCheckpointId();
		final String reason = (message.getReason() != null ? message.getReason().getMessage() : "");

		PendingCheckpoint checkpoint;

		synchronized (lock) {
			// we need to check inside the lock for being shutdown as well, otherwise we
			// get races and invalid error log messages
			if (shutdown) {
				return;
			}

			checkpoint = pendingCheckpoints.get(checkpointId);

			if (checkpoint != null) {
				Preconditions.checkState(
					!checkpoint.isDiscarded(),
					"Received message for discarded but non-removed checkpoint " + checkpointId);
				LOG.info("Decline checkpoint {} by task {} of job {} at {}.",
					checkpointId,
					message.getTaskExecutionId(),
					job,
					taskManagerLocationInfo);
				final CheckpointException checkpointException;
				if (message.getReason() == null) {
					checkpointException =
						new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED);
				} else {
					checkpointException = getCheckpointException(
						CheckpointFailureReason.JOB_FAILURE, message.getReason());
				}
				abortPendingCheckpoint(
					checkpoint,
					checkpointException,
					message.getTaskExecutionId());
			} else if (LOG.isDebugEnabled()) {
				if (recentPendingCheckpoints.contains(checkpointId)) {
					// message is for an unknown checkpoint, or comes too late (checkpoint disposed)
					LOG.debug("Received another decline message for now expired checkpoint attempt {} from task {} of job {} at {} : {}",
							checkpointId, message.getTaskExecutionId(), job, taskManagerLocationInfo, reason);
				} else {
					// message is for an unknown checkpoint. might be so old that we don't even remember it any more
					LOG.debug("Received decline message for unknown (too old?) checkpoint attempt {} from task {} of job {} at {} : {}",
							checkpointId, message.getTaskExecutionId(), job, taskManagerLocationInfo, reason);
				}
			}
		}
	}

	/**
	 * Receives an AcknowledgeCheckpoint message and returns whether the
	 * message was associated with a pending checkpoint.
	 *
	 * @param message Checkpoint ack from the task manager
	 *
	 * @param taskManagerLocationInfo The location of the acknowledge checkpoint message's sender
	 * @return Flag indicating whether the ack'd checkpoint was associated
	 * with a pending checkpoint.
	 *
	 * @throws CheckpointException If the checkpoint cannot be added to the completed checkpoint store.
	 */
	public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message, String taskManagerLocationInfo) throws CheckpointException {
		if (shutdown || message == null) {
			return false;
		}

		if (!job.equals(message.getJob())) {
			LOG.error("Received wrong AcknowledgeCheckpoint message for job {} from {} : {}", job, taskManagerLocationInfo, message);
			return false;
		}

		final long checkpointId = message.getCheckpointId();

		synchronized (lock) {
			// we need to check inside the lock for being shutdown as well, otherwise we
			// get races and invalid error log messages
			if (shutdown) {
				return false;
			}

			final PendingCheckpoint checkpoint = pendingCheckpoints.get(checkpointId);

			if (checkpoint != null && !checkpoint.isDiscarded()) {

				switch (checkpoint.acknowledgeTask(message.getTaskExecutionId(), message.getSubtaskState(), message.getCheckpointMetrics())) {
					case SUCCESS:
						LOG.debug("Received acknowledge message for checkpoint {} from task {} of job {} at {}.",
							checkpointId, message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);

						if (checkpoint.areTasksFullyAcknowledged()) {
							completePendingCheckpoint(checkpoint);
						}
						break;
					case DUPLICATE:
						LOG.debug("Received a duplicate acknowledge message for checkpoint {}, task {}, job {}, location {}.",
							message.getCheckpointId(), message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);
						break;
					case UNKNOWN:
						LOG.warn("Could not acknowledge the checkpoint {} for task {} of job {} at {}, " +
								"because the task's execution attempt id was unknown. Discarding " +
								"the state handle to avoid lingering state.", message.getCheckpointId(),
							message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);

						discardSubtaskState(message.getJob(), message.getTaskExecutionId(), message.getCheckpointId(), message.getSubtaskState());

						break;
					case DISCARDED:
						LOG.warn("Could not acknowledge the checkpoint {} for task {} of job {} at {}, " +
							"because the pending checkpoint had been discarded. Discarding the " +
								"state handle tp avoid lingering state.",
							message.getCheckpointId(), message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);

						discardSubtaskState(message.getJob(), message.getTaskExecutionId(), message.getCheckpointId(), message.getSubtaskState());
				}

				return true;
			}
			else if (checkpoint != null) {
				// this should not happen
				throw new IllegalStateException(
						"Received message for discarded but non-removed checkpoint " + checkpointId);
			}
			else {
				boolean wasPendingCheckpoint;

				// message is for an unknown checkpoint, or comes too late (checkpoint disposed)
				if (recentPendingCheckpoints.contains(checkpointId)) {
					wasPendingCheckpoint = true;
					LOG.warn("Received late message for now expired checkpoint attempt {} from task " +
						"{} of job {} at {}.", checkpointId, message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);
				}
				else {
					LOG.debug("Received message for an unknown checkpoint {} from task {} of job {} at {}.",
						checkpointId, message.getTaskExecutionId(), message.getJob(), taskManagerLocationInfo);
					wasPendingCheckpoint = false;
				}

				// try to discard the state so that we don't have lingering state lying around
				discardSubtaskState(message.getJob(), message.getTaskExecutionId(), message.getCheckpointId(), message.getSubtaskState());

				return wasPendingCheckpoint;
			}
		}
	}

	/**
	 * Try to complete the given pending checkpoint.
	 *
	 * <p>Important: This method should only be called in the checkpoint lock scope.
	 *
	 * @param pendingCheckpoint to complete
	 * @throws CheckpointException if the completion failed
	 */
	private void completePendingCheckpoint(PendingCheckpoint pendingCheckpoint) throws CheckpointException {
		final long checkpointId = pendingCheckpoint.getCheckpointId();
		final CompletedCheckpoint completedCheckpoint;

		// As a first step to complete the checkpoint, we register its state with the registry
		Map<OperatorID, OperatorState> operatorStates = pendingCheckpoint.getOperatorStates();
		sharedStateRegistry.registerAll(operatorStates.values());

		try {
			try {
				completedCheckpoint = pendingCheckpoint.finalizeCheckpoint();
				failureManager.handleCheckpointSuccess(pendingCheckpoint.getCheckpointId());
			}
			catch (Exception e1) {
				// abort the current pending checkpoint if we fails to finalize the pending checkpoint.
				if (!pendingCheckpoint.isDiscarded()) {
					abortPendingCheckpoint(
						pendingCheckpoint,
						new CheckpointException(
							CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, e1));
				}

				throw new CheckpointException("Could not finalize the pending checkpoint " + checkpointId + '.',
					CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, e1);
			}

			// the pending checkpoint must be discarded after the finalization
			Preconditions.checkState(pendingCheckpoint.isDiscarded() && completedCheckpoint != null);

			try {
				completedCheckpointStore.addCheckpoint(completedCheckpoint);
			} catch (Exception exception) {
				// we failed to store the completed checkpoint. Let's clean up
				executor.execute(new Runnable() {
					@Override
					public void run() {
						try {
							completedCheckpoint.discardOnFailedStoring();
						} catch (Throwable t) {
							LOG.warn("Could not properly discard completed checkpoint {}.", completedCheckpoint.getCheckpointID(), t);
						}
					}
				});

				throw new CheckpointException("Could not complete the pending checkpoint " + checkpointId + '.',
					CheckpointFailureReason.FINALIZE_CHECKPOINT_FAILURE, exception);
			}
		} finally {
			pendingCheckpoints.remove(checkpointId);

			resumePeriodicTriggering();
		}

		rememberRecentCheckpointId(checkpointId);

		// drop those pending checkpoints that are at prior to the completed one
		dropSubsumedCheckpoints(checkpointId);

		// record the time when this was completed, to calculate
		// the 'min delay between checkpoints'
		lastCheckpointCompletionRelativeTime = clock.relativeTimeMillis();

		LOG.info("Completed checkpoint {} for job {} ({} bytes in {} ms).", checkpointId, job,
			completedCheckpoint.getStateSize(), completedCheckpoint.getDuration());

		if (LOG.isDebugEnabled()) {
			StringBuilder builder = new StringBuilder();
			builder.append("Checkpoint state: ");
			for (OperatorState state : completedCheckpoint.getOperatorStates().values()) {
				builder.append(state);
				builder.append(", ");
			}
			// Remove last two chars ", "
			builder.setLength(builder.length() - 2);

			LOG.debug(builder.toString());
		}

		// send the "notify complete" call to all vertices, coordinators, etc.
		sendAcknowledgeMessages(checkpointId, completedCheckpoint.getTimestamp());
	}

	private void sendAcknowledgeMessages(long checkpointId, long timestamp) {
		// commit tasks
		for (ExecutionVertex ev : tasksToCommitTo) {
			Execution ee = ev.getCurrentExecutionAttempt();
			if (ee != null) {
				ee.notifyCheckpointComplete(checkpointId, timestamp);
			}
		}

		// commit coordinators
		for (OperatorCoordinatorCheckpointContext coordinatorContext : coordinatorsToCheckpoint) {
			coordinatorContext.coordinator().checkpointComplete(checkpointId);
		}
	}

	/**
	 * Fails all pending checkpoints which have not been acknowledged by the given execution
	 * attempt id.
	 *
	 * @param executionAttemptId for which to discard unacknowledged pending checkpoints
	 * @param cause of the failure
	 */
	public void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause) {
		synchronized (lock) {
			abortPendingCheckpoints(
				checkpoint -> !checkpoint.isAcknowledgedBy(executionAttemptId),
				new CheckpointException(CheckpointFailureReason.TASK_FAILURE, cause));
		}
	}

	private void rememberRecentCheckpointId(long id) {
		if (recentPendingCheckpoints.size() >= NUM_GHOST_CHECKPOINT_IDS) {
			recentPendingCheckpoints.removeFirst();
		}
		recentPendingCheckpoints.addLast(id);
	}

	private void dropSubsumedCheckpoints(long checkpointId) {
		abortPendingCheckpoints(
			checkpoint -> checkpoint.getCheckpointId() < checkpointId && checkpoint.canBeSubsumed(),
			new CheckpointException(CheckpointFailureReason.CHECKPOINT_SUBSUMED));
	}

	/**
	 * Resumes suspended periodic triggering.
	 *
	 * <p>NOTE: The caller of this method must hold the lock when invoking the method!
	 */
	private void resumePeriodicTriggering() {
		assert(Thread.holdsLock(lock));

		if (shutdown || !periodicScheduling) {
			return;
		}
		if (periodicTriggeringSuspended) {
			periodicTriggeringSuspended = false;

			// trigger the checkpoint from the trigger timer, to finish the work of this thread before
			// starting with the next checkpoint
			if (currentPeriodicTrigger != null) {
				currentPeriodicTrigger.cancel(false);
			}
			currentPeriodicTrigger = scheduleTriggerWithDelay(0L);
		}
	}

	// --------------------------------------------------------------------------------------------
	//  Checkpoint State Restoring
	// --------------------------------------------------------------------------------------------

	/**
	 * Restores the latest checkpointed state.
	 *
	 * @param tasks Map of job vertices to restore. State for these vertices is
	 * restored via {@link Execution#setInitialState(JobManagerTaskRestore)}.
	 * @param errorIfNoCheckpoint Fail if no completed checkpoint is available to
	 * restore from.
	 * @param allowNonRestoredState Allow checkpoint state that cannot be mapped
	 * to any job vertex in tasks.
	 * @return <code>true</code> if state was restored, <code>false</code> otherwise.
	 * @throws IllegalStateException If the CheckpointCoordinator is shut down.
	 * @throws IllegalStateException If no completed checkpoint is available and
	 *                               the <code>failIfNoCheckpoint</code> flag has been set.
	 * @throws IllegalStateException If the checkpoint contains state that cannot be
	 *                               mapped to any job vertex in <code>tasks</code> and the
	 *                               <code>allowNonRestoredState</code> flag has not been set.
	 * @throws IllegalStateException If the max parallelism changed for an operator
	 *                               that restores state from this checkpoint.
	 * @throws IllegalStateException If the parallelism changed for an operator
	 *                               that restores <i>non-partitioned</i> state from this
	 *                               checkpoint.
	 */
	@Deprecated
	public boolean restoreLatestCheckpointedState(
			Map<JobVertexID, ExecutionJobVertex> tasks,
			boolean errorIfNoCheckpoint,
			boolean allowNonRestoredState) throws Exception {

		return restoreLatestCheckpointedState(new HashSet<>(tasks.values()), errorIfNoCheckpoint, allowNonRestoredState);
	}

	/**
	 * Restores the latest checkpointed state.
	 *
	 * @param tasks Set of job vertices to restore. State for these vertices is
	 * restored via {@link Execution#setInitialState(JobManagerTaskRestore)}.
	 * @param errorIfNoCheckpoint Fail if no completed checkpoint is available to
	 * restore from.
	 * @param allowNonRestoredState Allow checkpoint state that cannot be mapped
	 * to any job vertex in tasks.
	 * @return <code>true</code> if state was restored, <code>false</code> otherwise.
	 * @throws IllegalStateException If the CheckpointCoordinator is shut down.
	 * @throws IllegalStateException If no completed checkpoint is available and
	 *                               the <code>failIfNoCheckpoint</code> flag has been set.
	 * @throws IllegalStateException If the checkpoint contains state that cannot be
	 *                               mapped to any job vertex in <code>tasks</code> and the
	 *                               <code>allowNonRestoredState</code> flag has not been set.
	 * @throws IllegalStateException If the max parallelism changed for an operator
	 *                               that restores state from this checkpoint.
	 * @throws IllegalStateException If the parallelism changed for an operator
	 *                               that restores <i>non-partitioned</i> state from this
	 *                               checkpoint.
	 */
	public boolean restoreLatestCheckpointedState(
			final Set<ExecutionJobVertex> tasks,
			final boolean errorIfNoCheckpoint,
			final boolean allowNonRestoredState) throws Exception {

		synchronized (lock) {
			if (shutdown) {
				throw new IllegalStateException("CheckpointCoordinator is shut down");
			}

			// We create a new shared state registry object, so that all pending async disposal requests from previous
			// runs will go against the old object (were they can do no harm).
			// This must happen under the checkpoint lock.
			sharedStateRegistry.close();
			sharedStateRegistry = sharedStateRegistryFactory.create(executor);

			// Recover the checkpoints, TODO this could be done only when there is a new leader, not on each recovery
			completedCheckpointStore.recover();

			// Now, we re-register all (shared) states from the checkpoint store with the new registry
			for (CompletedCheckpoint completedCheckpoint : completedCheckpointStore.getAllCheckpoints()) {
				completedCheckpoint.registerSharedStatesAfterRestored(sharedStateRegistry);
			}

			LOG.debug("Status of the shared state registry of job {} after restore: {}.", job, sharedStateRegistry);

			// Restore from the latest checkpoint
			CompletedCheckpoint latest = completedCheckpointStore.getLatestCheckpoint(isPreferCheckpointForRecovery);

			if (latest == null) {
				if (errorIfNoCheckpoint) {
					throw new IllegalStateException("No completed checkpoint available");
				} else {
					LOG.debug("Resetting the master hooks.");
					MasterHooks.reset(masterHooks.values(), LOG);

					return false;
				}
			}

			LOG.info("Restoring job {} from latest valid checkpoint: {}.", job, latest);

			// re-assign the task states
			final Map<OperatorID, OperatorState> operatorStates = latest.getOperatorStates();

			StateAssignmentOperation stateAssignmentOperation =
					new StateAssignmentOperation(latest.getCheckpointID(), tasks, operatorStates, allowNonRestoredState);

			stateAssignmentOperation.assignStates();

			// call master hooks for restore

			MasterHooks.restoreMasterHooks(
					masterHooks,
					latest.getMasterHookStates(),
					latest.getCheckpointID(),
					allowNonRestoredState,
					LOG);

			// update metrics

			if (statsTracker != null) {
				long restoreTimestamp = System.currentTimeMillis();
				RestoredCheckpointStats restored = new RestoredCheckpointStats(
					latest.getCheckpointID(),
					latest.getProperties(),
					restoreTimestamp,
					latest.getExternalPointer());

				statsTracker.reportRestoredCheckpoint(restored);
			}

			return true;
		}
	}

	/**
	 * Restore the state with given savepoint.
	 *
	 * @param savepointPointer The pointer to the savepoint.
	 * @param allowNonRestored True if allowing checkpoint state that cannot be
	 *                         mapped to any job vertex in tasks.
	 * @param tasks            Map of job vertices to restore. State for these
	 *                         vertices is restored via
	 *                         {@link Execution#setInitialState(JobManagerTaskRestore)}.
	 * @param userClassLoader  The class loader to resolve serialized classes in
	 *                         legacy savepoint versions.
	 */
	public boolean restoreSavepoint(
			String savepointPointer,
			boolean allowNonRestored,
			Map<JobVertexID, ExecutionJobVertex> tasks,
			ClassLoader userClassLoader) throws Exception {

		Preconditions.checkNotNull(savepointPointer, "The savepoint path cannot be null.");

		LOG.info("Starting job {} from savepoint {} ({})",
				job, savepointPointer, (allowNonRestored ? "allowing non restored state" : ""));

		final CompletedCheckpointStorageLocation checkpointLocation = checkpointStorage.resolveCheckpoint(savepointPointer);

		// Load the savepoint as a checkpoint into the system
		CompletedCheckpoint savepoint = Checkpoints.loadAndValidateCheckpoint(
				job, tasks, checkpointLocation, userClassLoader, allowNonRestored);

		completedCheckpointStore.addCheckpoint(savepoint);

		// Reset the checkpoint ID counter
		long nextCheckpointId = savepoint.getCheckpointID() + 1;
		checkpointIdCounter.setCount(nextCheckpointId);

		LOG.info("Reset the checkpoint ID of job {} to {}.", job, nextCheckpointId);

		return restoreLatestCheckpointedState(new HashSet<>(tasks.values()), true, allowNonRestored);
	}

	// ------------------------------------------------------------------------
	//  Accessors
	// ------------------------------------------------------------------------

	public int getNumberOfPendingCheckpoints() {
		return this.pendingCheckpoints.size();
	}

	public int getNumberOfRetainedSuccessfulCheckpoints() {
		synchronized (lock) {
			return completedCheckpointStore.getNumberOfRetainedCheckpoints();
		}
	}

	public Map<Long, PendingCheckpoint> getPendingCheckpoints() {
		synchronized (lock) {
			return new HashMap<>(this.pendingCheckpoints);
		}
	}

	public List<CompletedCheckpoint> getSuccessfulCheckpoints() throws Exception {
		synchronized (lock) {
			return completedCheckpointStore.getAllCheckpoints();
		}
	}

	public CheckpointStorageCoordinatorView getCheckpointStorage() {
		return checkpointStorage;
	}

	public CompletedCheckpointStore getCheckpointStore() {
		return completedCheckpointStore;
	}

	public long getCheckpointTimeout() {
		return checkpointTimeout;
	}

	public ArrayDeque<CheckpointTriggerRequest> getTriggerRequestQueue() {
		return triggerRequestQueue;
	}

	public boolean isTriggering() {
		return isTriggering;
	}

	@VisibleForTesting
	boolean isCurrentPeriodicTriggerAvailable() {
		return currentPeriodicTrigger != null;
	}

	/**
	 * Returns whether periodic checkpointing has been configured.
	 *
	 * @return <code>true</code> if periodic checkpoints have been configured.
	 */
	public boolean isPeriodicCheckpointingConfigured() {
		return baseInterval != Long.MAX_VALUE;
	}

	// --------------------------------------------------------------------------------------------
	//  Periodic scheduling of checkpoints
	// --------------------------------------------------------------------------------------------

	public void startCheckpointScheduler() {
		synchronized (lock) {
			if (shutdown) {
				throw new IllegalArgumentException("Checkpoint coordinator is shut down");
			}

			// make sure all prior timers are cancelled
			stopCheckpointScheduler();

			periodicScheduling = true;
			currentPeriodicTrigger = scheduleTriggerWithDelay(getRandomInitDelay());
		}
	}

	public void stopCheckpointScheduler() {
		synchronized (lock) {
			periodicTriggeringSuspended = false;
			periodicScheduling = false;

			if (currentPeriodicTrigger != null) {
				currentPeriodicTrigger.cancel(false);
				currentPeriodicTrigger = null;
			}

			final CheckpointException reason =
				new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SUSPEND);
			abortPendingAndQueuedCheckpoints(reason);

			numUnsuccessfulCheckpointsTriggers.set(0);
		}
	}

	/**
	 * Aborts all the pending checkpoints due to en exception.
	 * @param exception The exception.
	 */
	public void abortPendingCheckpoints(CheckpointException exception) {
		synchronized (lock) {
			abortPendingCheckpoints(ignored -> true, exception);
		}
	}

	private void abortPendingCheckpoints(
		Predicate<PendingCheckpoint> checkpointToFailPredicate,
		CheckpointException exception) {

		assert Thread.holdsLock(lock);

		final PendingCheckpoint[] pendingCheckpointsToFail = pendingCheckpoints
			.values()
			.stream()
			.filter(checkpointToFailPredicate)
			.toArray(PendingCheckpoint[]::new);

		// do not traverse pendingCheckpoints directly, because it might be changed during traversing
		for (PendingCheckpoint pendingCheckpoint : pendingCheckpointsToFail) {
			abortPendingCheckpoint(pendingCheckpoint, exception);
		}
	}

	/**
	 * If too many checkpoints are currently in progress, we need to mark that a request is queued.
	 *
	 * @throws CheckpointException If too many checkpoints are currently in progress.
	 */
	private void checkConcurrentCheckpoints() throws CheckpointException {
		if (pendingCheckpoints.size() >= maxConcurrentCheckpointAttempts) {
			periodicTriggeringSuspended = true;
			if (currentPeriodicTrigger != null) {
				currentPeriodicTrigger.cancel(false);
				currentPeriodicTrigger = null;
			}
			throw new CheckpointException(CheckpointFailureReason.TOO_MANY_CONCURRENT_CHECKPOINTS);
		}
	}

	/**
	 * Make sure the minimum interval between checkpoints has passed.
	 *
	 * @throws CheckpointException If the minimum interval between checkpoints has not passed.
	 */
	private void checkMinPauseBetweenCheckpoints() throws CheckpointException {
		final long nextCheckpointTriggerRelativeTime =
			lastCheckpointCompletionRelativeTime + minPauseBetweenCheckpoints;
		final long durationTillNextMillis =
			nextCheckpointTriggerRelativeTime - clock.relativeTimeMillis();

		if (durationTillNextMillis > 0) {
			if (currentPeriodicTrigger != null) {
				currentPeriodicTrigger.cancel(false);
				currentPeriodicTrigger = null;
			}
			// Reassign the new trigger to the currentPeriodicTrigger
			currentPeriodicTrigger = scheduleTriggerWithDelay(durationTillNextMillis);

			throw new CheckpointException(CheckpointFailureReason.MINIMUM_TIME_BETWEEN_CHECKPOINTS);
		}
	}

	private long getRandomInitDelay() {
		return ThreadLocalRandom.current().nextLong(minPauseBetweenCheckpoints, baseInterval + 1L);
	}

	private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {
		return timer.scheduleAtFixedRate(
			new ScheduledTrigger(),
			initDelay, baseInterval, TimeUnit.MILLISECONDS);
	}

	// ------------------------------------------------------------------------
	//  job status listener that schedules / cancels periodic checkpoints
	// ------------------------------------------------------------------------

	public JobStatusListener createActivatorDeactivator() {
		synchronized (lock) {
			if (shutdown) {
				throw new IllegalArgumentException("Checkpoint coordinator is shut down");
			}

			if (jobStatusListener == null) {
				jobStatusListener = new CheckpointCoordinatorDeActivator(this);
			}

			return jobStatusListener;
		}
	}

	// ------------------------------------------------------------------------

	private final class ScheduledTrigger implements Runnable {

		@Override
		public void run() {
			try {
				triggerCheckpoint(System.currentTimeMillis(), true);
			}
			catch (Exception e) {
				LOG.error("Exception while triggering checkpoint for job {}.", job, e);
			}
		}
	}

	/**
	 * Discards the given state object asynchronously belonging to the given job, execution attempt
	 * id and checkpoint id.
	 *
	 * @param jobId identifying the job to which the state object belongs
	 * @param executionAttemptID identifying the task to which the state object belongs
	 * @param checkpointId of the state object
	 * @param subtaskState to discard asynchronously
	 */
	private void discardSubtaskState(
			final JobID jobId,
			final ExecutionAttemptID executionAttemptID,
			final long checkpointId,
			final TaskStateSnapshot subtaskState) {

		if (subtaskState != null) {
			executor.execute(new Runnable() {
				@Override
				public void run() {

					try {
						subtaskState.discardState();
					} catch (Throwable t2) {
						LOG.warn("Could not properly discard state object of checkpoint {} " +
							"belonging to task {} of job {}.", checkpointId, executionAttemptID, jobId, t2);
					}
				}
			});
		}
	}

	private void abortPendingCheckpoint(
		PendingCheckpoint pendingCheckpoint,
		CheckpointException exception) {

		abortPendingCheckpoint(pendingCheckpoint, exception, null);
	}

	private void abortPendingCheckpoint(
		PendingCheckpoint pendingCheckpoint,
		CheckpointException exception,
		@Nullable final ExecutionAttemptID executionAttemptID) {

		assert(Thread.holdsLock(lock));

		if (!pendingCheckpoint.isDiscarded()) {
			try {
				// release resource here
				pendingCheckpoint.abort(
					exception.getCheckpointFailureReason(), exception.getCause());

				if (pendingCheckpoint.getProps().isSavepoint() &&
					pendingCheckpoint.getProps().isSynchronous()) {
					failureManager.handleSynchronousSavepointFailure(exception);
				} else if (executionAttemptID != null) {
					failureManager.handleTaskLevelCheckpointException(
						exception, pendingCheckpoint.getCheckpointId(), executionAttemptID);
				} else {
					failureManager.handleJobLevelCheckpointException(
						exception, pendingCheckpoint.getCheckpointId());
				}
			} finally {
				pendingCheckpoints.remove(pendingCheckpoint.getCheckpointId());
				rememberRecentCheckpointId(pendingCheckpoint.getCheckpointId());

				resumePeriodicTriggering();
			}
		}
	}

	private void preCheckBeforeTriggeringCheckpoint(boolean isPeriodic, boolean forceCheckpoint) throws CheckpointException {
		preCheckGlobalState(isPeriodic);

		if (!forceCheckpoint) {
			checkConcurrentCheckpoints();
			checkMinPauseBetweenCheckpoints();
		}
	}

	private void preCheckGlobalState(boolean isPeriodic) throws CheckpointException {
		// abort if the coordinator has been shutdown in the meantime
		if (shutdown) {
			throw new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
		}

		// Don't allow periodic checkpoint if scheduling has been disabled
		if (isPeriodic && !periodicScheduling) {
			throw new CheckpointException(CheckpointFailureReason.PERIODIC_SCHEDULER_SHUTDOWN);
		}
	}

	/**
	 * Check if all tasks that we need to trigger are running. If not, abort the checkpoint.
	 *
	 * @return the executions need to be triggered.
	 * @throws CheckpointException the exception fails checking
	 */
	private Execution[] getTriggerExecutions() throws CheckpointException {
		Execution[] executions = new Execution[tasksToTrigger.length];
		for (int i = 0; i < tasksToTrigger.length; i++) {
			Execution ee = tasksToTrigger[i].getCurrentExecutionAttempt();
			if (ee == null) {
				LOG.info(
					"Checkpoint triggering task {} of job {} is not being executed at the moment. Aborting checkpoint.",
					tasksToTrigger[i].getTaskNameWithSubtaskIndex(),
					job);
				throw new CheckpointException(
					CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
			} else if (ee.getState() == ExecutionState.RUNNING) {
				executions[i] = ee;
			} else {
				LOG.info(
					"Checkpoint triggering task {} of job {} is not in state {} but {} instead. Aborting checkpoint.",
					tasksToTrigger[i].getTaskNameWithSubtaskIndex(),
					job,
					ExecutionState.RUNNING,
					ee.getState());
				throw new CheckpointException(
					CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
			}
		}
		return executions;
	}

	/**
	 * Check if all tasks that need to acknowledge the checkpoint are running.
	 * If not, abort the checkpoint
	 *
	 * @return the execution vertices which should give an ack response
	 * @throws CheckpointException the exception fails checking
	 */
	private Map<ExecutionAttemptID, ExecutionVertex> getAckTasks() throws CheckpointException {
		Map<ExecutionAttemptID, ExecutionVertex> ackTasks = new HashMap<>(tasksToWaitFor.length);

		for (ExecutionVertex ev : tasksToWaitFor) {
			Execution ee = ev.getCurrentExecutionAttempt();
			if (ee != null) {
				ackTasks.put(ee.getAttemptId(), ev);
			} else {
				LOG.info(
					"Checkpoint acknowledging task {} of job {} is not being executed at the moment. Aborting checkpoint.",
					ev.getTaskNameWithSubtaskIndex(),
					job);
				throw new CheckpointException(
					CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
			}
		}
		return ackTasks;
	}

	private void abortPendingAndQueuedCheckpoints(CheckpointException exception) {
		assert(Thread.holdsLock(lock));
		CheckpointTriggerRequest request;
		while ((request = triggerRequestQueue.poll()) != null) {
			request.onCompletionPromise.completeExceptionally(exception);
		}
		abortPendingCheckpoints(exception);
	}

	/**
	 * The canceller of checkpoint. The checkpoint might be cancelled if it doesn't finish in a
	 * configured period.
	 */
	private class CheckpointCanceller implements Runnable {

		private final PendingCheckpoint pendingCheckpoint;

		private CheckpointCanceller(PendingCheckpoint pendingCheckpoint) {
			this.pendingCheckpoint = checkNotNull(pendingCheckpoint);
		}

		@Override
		public void run() {
			synchronized (lock) {
				// only do the work if the checkpoint is not discarded anyways
				// note that checkpoint completion discards the pending checkpoint object
				if (!pendingCheckpoint.isDiscarded()) {
					LOG.info("Checkpoint {} of job {} expired before completing.",
						pendingCheckpoint.getCheckpointId(), job);

					abortPendingCheckpoint(
						pendingCheckpoint,
						new CheckpointException(CheckpointFailureReason.CHECKPOINT_EXPIRED));
				}
			}
		}
	}

	private static CheckpointException getCheckpointException(
		CheckpointFailureReason defaultReason, Throwable throwable) {

		final Optional<CheckpointException> checkpointExceptionOptional =
			ExceptionUtils.findThrowable(throwable, CheckpointException.class);
		return checkpointExceptionOptional
			.orElseGet(() -> new CheckpointException(defaultReason, throwable));
	}

	private static class CheckpointIdAndStorageLocation {
		private final long checkpointId;
		private final CheckpointStorageLocation checkpointStorageLocation;

		CheckpointIdAndStorageLocation(
			long checkpointId,
			CheckpointStorageLocation checkpointStorageLocation) {

			this.checkpointId = checkpointId;
			this.checkpointStorageLocation = checkNotNull(checkpointStorageLocation);
		}
	}

	private static class CheckpointTriggerRequest {
		private final long timestamp;
		private final CheckpointProperties props;
		private final @Nullable String externalSavepointLocation;
		private final boolean isPeriodic;
		private final boolean advanceToEndOfTime;
		private final CompletableFuture<CompletedCheckpoint> onCompletionPromise;

		CheckpointTriggerRequest(
			long timestamp,
			CheckpointProperties props,
			@Nullable String externalSavepointLocation,
			boolean isPeriodic,
			boolean advanceToEndOfTime,
			CompletableFuture<CompletedCheckpoint> onCompletionPromise) {

			this.timestamp = timestamp;
			this.props = checkNotNull(props);
			this.externalSavepointLocation = externalSavepointLocation;
			this.isPeriodic = isPeriodic;
			this.advanceToEndOfTime = advanceToEndOfTime;
			this.onCompletionPromise = checkNotNull(onCompletionPromise);
		}
	}
}