text:
import ReactDOM from 'react-dom';
import React from 'react';
import { Provider } from 'react-redux';
import { Router, Route, IndexRoute, hashHistory } from 'react-router';
import App from './components/App';
import store from './store';
import Home from './components/Home';
import Login from './components/Login';
import Register from './components/Register';
import Settings from './components/Settings';
import Animal from './components/Animal';
import AnimalForm from './components/Animal/AnimalForm';

function hashLinkScroll() {
  const { hash } = window.location;
  if (hash !== '') {
    setTimeout(() => {
      const id = hash.replace('#', '');
      const element = document.getElementById(id);
      if (element) element.scrollIntoView();
    }, 0);
  }
}

ReactDOM.render(
  <Provider store={store}>
    <Router history={hashHistory} onUpdate={hashLinkScroll}>
      <Route path="/" component={App}>
        <IndexRoute component={Home} />
        <Route path="login" component={Login} />
        <Route path="register" component={Register} />
        <Route path="settings" component={Settings} />
        <Route path="animals" component={Animal} />
        <Route path="animalform" component={AnimalForm} />
        <Route path="animalform/:slug" component={AnimalForm} />
      </Route>
    </Router>
  </Provider>,
  document.getElementById('root')
);


text:
var count = 0;

function createTemplate(template, data) {
  if (!Array.isArray(data)) {
    var o = [];
    for (var i=0; i< data; i++) {
      o.push(i);
    }
    data = o;
  }
  count = data.length;

  var clone = template.cloneNode(true);
  document.body.appendChild(clone);
  clone.bindingDelegate = new PolymerExpressions();
  clone.setAttribute('repeat', '');
  clone.model = data;
}

function test(template, data) {
  timeTest(function() {
    createTemplate(template, data);
  });
}

function timeTest(action) {
  var start = performance.now();
  action();
  new MutationObserver(function() {
    requestAnimationFrame(function() {
      var end = performance.now();
      var time = end - start;
      var itemTime = (time / count).toFixed(3);
      console.log('Time to generate ' + count + ' items: ' + time +
        ' ms, or ' + itemTime + 'ms/item');
    });
  }).observe(document.body, {childList: true});

}

redpajama_set_name: RedPajamaGithub


text:
"""
server.py

File contains all routes for Flask app
"""

from flask import Flask, request, render_template, jsonify
from jinja2 import StrictUndefined
import os

app = Flask(__name__)

app.secret_key = os.environ.get("FLASK_SECRET_KEY", "ABCDEF")
app.jinja_env.undefined = StrictUndefined

@app.route("/")
def home():
    """ Home page """

    return render_template("index.html")

if __name__ == "__main__":

    PORT = int(os.environ.get("PORT", 5000))
    DEBUG = "NO_DEBUG" not in os.environ

    app.run(debug=DEBUG, host="0.0.0.0", port=PORT)


text:
New-Item -ItemType Directory -Force -Path TestAndCoverage

# Clear the Coverage folder
Remove-Item TestAndCoverage\* -recurse

# Define the BuildConfiguration
New-Variable -Name BuildConfiguration -Value ($env:BuildConfiguration)
if ([string]::IsNullOrEmpty($BuildConfiguration))
{
		Set-Variable -Name BuildConfiguration -Value "Debug"
}

# Run Unit Tests and Coverage
.\packages\OpenCover.4.6.519\tools\OpenCover.Console.exe -register:user -output:TestAndCoverage\coverage.xml -target:"packages\xunit.runner.console.2.3.1\tools\net452\xunit.console.x86.exe" -targetargs:"tests\DocFunctions.Lib.Unit\bin\$BuildConfiguration\DocFunctions.Lib.Unit.dll -noshadow -xml TestAndCoverage\UnitTestResults.xml" -filter:"+[DocFunctions.Lib*]* -[DocFunctions.Lib.Unit]*" -hideskipped:Attribute

# Convert Coverage Report to Cobertura format
.\packages\OpenCoverToCoberturaConverter.0.2.6.0\tools\OpenCoverToCoberturaConverter.exe -input:TestAndCoverage\coverage.xml -output:TestAndCoverage\cobertura.xml

# Generate Coverage Report
.\packages\ReportGenerator.2.5.7\tools\ReportGenerator.exe TestAndCoverage\coverage.xml TestAndCoverage\coverage


text:
var GetXSConfigurationSkillResponse_Skills = {

};


text:
import java.lang.reflect.Method;

class InvokeReflect {

  static String  methodToRun         = "run";
  static String  signatureToPrintOut = "()Z";
  static Class[] noparams            = {};

  public static void main(String[] argv) throws Exception {
    if (argv.length == 0) { printUsage(); return; }

    for (int iArg=0; iArg<argv.length; iArg++) {
      String arg = argv[iArg];
      if (arg.startsWith("-h")) { printUsage(); return; }

      // invoking methodToRun() in the class specified by arg
      System.out.println("**** START OF EXECUTION of " + arg + "." +
                         methodToRun + " " + signatureToPrintOut + " ****.");
      Class  klass = Class.forName(arg);
      Method method = klass.getDeclaredMethod(methodToRun, noparams);
      Object result = method.invoke(null, (Object[])noparams);
      System.out.println("**** RESULT: " + result);
    }
  }

  // self-test
  static boolean run() throws Exception {
    methodToRun         = "sampleRun";
    String[] sampleArgv = {"InvokeReflect", "-h"};
    System.out.println("Running `InvokeReflect InvokeReflect -h'");
    main(sampleArgv);
    return true;
  }

  static boolean sampleRun() {
    System.out.println("This is InvokeReflect.sampleRun()");
    return true;
  }

  static void printUsage() {
    System.out.println("\nUsage:  InvokeReflect className ...\n");
  }
}


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">

Contains classes for managing softwares and upgrading devices.

In addition to the normal page classes this package also contains a FileComparator that is used to sort the list of Files.

There is a help file, FirmwarePage.html, that contains generic information about the Software concept. The naming differences (Firmware/Software) is due to name refactoring.


text:
"""Runs Android's lint tool."""

import argparse
import os
import re
import sys
import traceback
from xml.dom import minidom

from util import build_utils

_LINT_MD_URL = 'https://chromium.googlesource.com/chromium/src/+/master/build/android/docs/lint.md' # pylint: disable=line-too-long
_SRC_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__),
                                         '..', '..', '..'))

def _OnStaleMd5(lint_path, config_path, processed_config_path,
                manifest_path, result_path, product_dir, sources, jar_path,
                cache_dir, android_sdk_version, resource_sources,
                classpath=None, can_fail_build=False, silent=False):
  def _RebasePath(path):
    """Returns relative path to top-level src dir.

    Args:
      path: A path relative to cwd.
    """
    ret = os.path.relpath(os.path.abspath(path), _SRC_ROOT)
    # If it's outside of src/, just use abspath.
    if ret.startswith('..'):
      ret = os.path.abspath(path)
    return ret

  def _ProcessConfigFile():
    if not config_path or not processed_config_path:
      return
    if not build_utils.IsTimeStale(processed_config_path, [config_path]):
      return

    with open(config_path, 'rb') as f:
      content = f.read().replace(
          'PRODUCT_DIR', _RebasePath(product_dir))

    with open(processed_config_path, 'wb') as f:
      f.write(content)

  def _ProcessResultFile():
    with open(result_path, 'rb') as f:
      content = f.read().replace(
          _RebasePath(product_dir), 'PRODUCT_DIR')

    with open(result_path, 'wb') as f:
      f.write(content)

  def _ParseAndShowResultFile():
    dom = minidom.parse(result_path)
    issues = dom.getElementsByTagName('issue')
    if not silent:
      print >> sys.stderr
      for issue in issues:
        issue_id = issue.attributes['id'].value
        message = issue.attributes['message'].value
        location_elem = issue.getElementsByTagName('location')[0]
        path = location_elem.attributes['file'].value
        line = location_elem.getAttribute('line')
        if line:
          error = '%s:%s %s: %s [warning]' % (path, line, message, issue_id)
        else:
          # Issues in class files don't have a line number.
          error = '%s %s: %s [warning]' % (path, message, issue_id)
        print >> sys.stderr, error.encode('utf-8')
        for attr in ['errorLine1', 'errorLine2']:
          error_line = issue.getAttribute(attr)
          if error_line:
            print >> sys.stderr, error_line.encode('utf-8')
    return len(issues)

  with build_utils.TempDir() as temp_dir:
    _ProcessConfigFile()

    cmd = [
        _RebasePath(lint_path), '-Werror', '--exitcode', '--showall',
        '--xml', _RebasePath(result_path),
    ]
    if jar_path:
      # --classpath is just for .class files for this one target.
      cmd.extend(['--classpath', _RebasePath(jar_path)])
    if processed_config_path:
      cmd.extend(['--config', _RebasePath(processed_config_path)])

    tmp_dir_counter = [0]
    def _NewTempSubdir(prefix, append_digit=True):
      # Helper function to create a new sub directory based on the number of
      # subdirs created earlier.
      if append_digit:
        tmp_dir_counter[0] += 1
        prefix += str(tmp_dir_counter[0])
      new_dir = os.path.join(temp_dir, prefix)
      os.makedirs(new_dir)
      return new_dir

    resource_dirs = []
    for resource_source in resource_sources:
      if os.path.isdir(resource_source):
        resource_dirs.append(resource_source)
      else:
        # This is a zip file with generated resources (e. g. strings from GRD).
        # Extract it to temporary folder.
        resource_dir = _NewTempSubdir(_RebasePath(resource_source),
                                      append_digit=False)
        resource_dirs.append(resource_dir)
        build_utils.ExtractAll(resource_source, path=resource_dir)

    for resource_dir in resource_dirs:
      cmd.extend(['--resources', _RebasePath(resource_dir)])

    if classpath:
      # --libraries is the classpath (excluding active target).
      cp = ':'.join(_RebasePath(p) for p in classpath)
      cmd.extend(['--libraries', cp])

    # There may be multiple source files with the same basename (but in
    # different directories). It is difficult to determine what part of the path
    # corresponds to the java package, and so instead just link the source files
    # into temporary directories (creating a new one whenever there is a name
    # conflict).
    def PathInDir(d, src):
      subpath = os.path.join(d, _RebasePath(src))
      subdir = os.path.dirname(subpath)
      if not os.path.exists(subdir):
        os.makedirs(subdir)
      return subpath

    src_dirs = []
    for src in sources:
      src_dir = None
      for d in src_dirs:
        if not os.path.exists(PathInDir(d, src)):
          src_dir = d
          break
      if not src_dir:
        src_dir = _NewTempSubdir('SRC_ROOT')
        src_dirs.append(src_dir)
        cmd.extend(['--sources', _RebasePath(src_dir)])
      os.symlink(os.path.abspath(src), PathInDir(src_dir, src))

    project_dir = _NewTempSubdir('SRC_ROOT')
    if android_sdk_version:
      # Create dummy project.properies file in a temporary "project" directory.
      # It is the only way to add Android SDK to the Lint's classpath. Proper
      # classpath is necessary for most source-level checks.
      with open(os.path.join(project_dir, 'project.properties'), 'w') \
          as propfile:
        print >> propfile, 'target=android-{}'.format(android_sdk_version)

    # Put the manifest in a temporary directory in order to avoid lint detecting
    # sibling res/ and src/ directories (which should be pass explicitly if they
    # are to be included).
    if manifest_path:
      os.symlink(os.path.abspath(manifest_path),
                 os.path.join(project_dir, 'AndroidManifest.xml'))
    cmd.append(project_dir)

    if os.path.exists(result_path):
      os.remove(result_path)

    env = {}
    stderr_filter = None
    if cache_dir:
      env['_JAVA_OPTIONS'] = '-Duser.home=%s' % _RebasePath(cache_dir)
      # When _JAVA_OPTIONS is set, java prints to stderr:
      # Picked up _JAVA_OPTIONS: ...
      #
      # We drop all lines that contain _JAVA_OPTIONS from the output
      stderr_filter = lambda l: re.sub(r'.*_JAVA_OPTIONS.*\n?', '', l)

    try:
      build_utils.CheckOutput(cmd, cwd=_SRC_ROOT, env=env or None,
                              stderr_filter=stderr_filter)
    except build_utils.CalledProcessError:
      # There is a problem with lint usage
      if not os.path.exists(result_path):
        raise

      # Sometimes produces empty (almost) files:
      if os.path.getsize(result_path) < 10:
        if can_fail_build:
          raise
        elif not silent:
          traceback.print_exc()
        return

      # There are actual lint issues
      try:
        num_issues = _ParseAndShowResultFile()
      except Exception: # pylint: disable=broad-except
        if not silent:
          print 'Lint created unparseable xml file...'
          print 'File contents:'
          with open(result_path) as f:
            print f.read()
        if not can_fail_build:
          return

      if can_fail_build and not silent:
        traceback.print_exc()

      # There are actual lint issues
      try:
        num_issues = _ParseAndShowResultFile()
      except Exception: # pylint: disable=broad-except
        if not silent:
          print 'Lint created unparseable xml file...'
          print 'File contents:'
          with open(result_path) as f:
            print f.read()
        raise

      _ProcessResultFile()
      msg = ('\nLint found %d new issues.\n'
             ' - For full explanation, please refer to %s\n'
             ' - For more information about lint and how to fix lint issues,'
             ' please refer to %s\n' %
             (num_issues, _RebasePath(result_path), _LINT_MD_URL))
      if not silent:
        print >> sys.stderr, msg
      if can_fail_build:
        raise Exception('Lint failed.')

def main():
  parser = argparse.ArgumentParser()
  build_utils.AddDepfileOption(parser)

  parser.add_argument('--lint-path', required=True,
                      help='Path to lint executable.')
  parser.add_argument('--product-dir', required=True,
                      help='Path to product dir.')
  parser.add_argument('--result-path', required=True,
                      help='Path to XML lint result file.')
  parser.add_argument('--cache-dir', required=True,
                      help='Path to the directory in which the android cache '
                           'directory tree should be stored.')
  parser.add_argument('--platform-xml-path', required=True,
                      help='Path to api-platforms.xml')
  parser.add_argument('--android-sdk-version',
                      help='Version (API level) of the Android SDK used for '
                           'building.')
  parser.add_argument('--create-cache', action='store_true',
                      help='Mark the lint cache file as an output rather than '
                      'an input.')
  parser.add_argument('--can-fail-build', action='store_true',
                      help='If set, script will exit with nonzero exit status'
                           ' if lint errors are present')
  parser.add_argument('--config-path',
                      help='Path to lint suppressions file.')
  parser.add_argument('--enable', action='store_true',
                      help='Run lint instead of just touching stamp.')
  parser.add_argument('--jar-path',
                      help='Jar file containing class files.')
  parser.add_argument('--java-sources-file',
                      help='File containing a list of java files.')
  parser.add_argument('--manifest-path',
                      help='Path to AndroidManifest.xml')
  parser.add_argument('--classpath', default=[], action='append',
                      help='GYP-list of classpath .jar files')
  parser.add_argument('--processed-config-path',
                      help='Path to processed lint suppressions file.')
  parser.add_argument('--resource-dir',
                      help='Path to resource dir.')
  parser.add_argument('--resource-sources', default=[], action='append',
                      help='GYP-list of resource sources (directories with '
                      'resources or archives created by resource-generating '
                      'tasks.')
  parser.add_argument('--silent', action='store_true',
                      help='If set, script will not log anything.')
  parser.add_argument('--src-dirs',
                      help='Directories containing java files.')
  parser.add_argument('--stamp',
                      help='Path to touch on success.')

  args = parser.parse_args(build_utils.ExpandFileArgs(sys.argv[1:]))

  if args.enable:
    sources = []
    if args.src_dirs:
      src_dirs = build_utils.ParseGnList(args.src_dirs)
      sources = build_utils.FindInDirectories(src_dirs, '*.java')
    elif args.java_sources_file:
      sources.extend(build_utils.ReadSourcesList(args.java_sources_file))

    if args.config_path and not args.processed_config_path:
      parser.error('--config-path specified without --processed-config-path')
    elif args.processed_config_path and not args.config_path:
      parser.error('--processed-config-path specified without --config-path')

    input_paths = [
        args.lint_path,
        args.platform_xml_path,
    ]
    if args.config_path:
      input_paths.append(args.config_path)
    if args.jar_path:
      input_paths.append(args.jar_path)
    if args.manifest_path:
      input_paths.append(args.manifest_path)
    if sources:
      input_paths.extend(sources)
    classpath = []
    for gyp_list in args.classpath:
      classpath.extend(build_utils.ParseGnList(gyp_list))
    input_paths.extend(classpath)

    resource_sources = []
    if args.resource_dir:
      # Backward compatibility with GYP
      resource_sources += [ args.resource_dir ]

    for gyp_list in args.resource_sources:
      resource_sources += build_utils.ParseGnList(gyp_list)

    for resource_source in resource_sources:
      if os.path.isdir(resource_source):
        input_paths.extend(build_utils.FindInDirectory(resource_source, '*'))
      else:
        input_paths.append(resource_source)

    input_strings = [
      args.can_fail_build,
      args.silent,
    ]
    if args.android_sdk_version:
      input_strings.append(args.android_sdk_version)
    if args.processed_config_path:
      input_strings.append(args.processed_config_path)

    output_paths = [ args.result_path ]

    build_utils.CallAndWriteDepfileIfStale(
        lambda: _OnStaleMd5(args.lint_path,
                            args.config_path,
                            args.processed_config_path,
                            args.manifest_path, args.result_path,
                            args.product_dir, sources,
                            args.jar_path,
                            args.cache_dir,
                            args.android_sdk_version,
                            resource_sources,
                            classpath=classpath,
                            can_fail_build=args.can_fail_build,
                            silent=args.silent),
        args,
        input_paths=input_paths,
        input_strings=input_strings,
        output_paths=output_paths,
        depfile_deps=classpath)

if __name__ == '__main__':
  sys.exit(main())


text:
QACI0 ; OAKOIFO/TKW - DATA MIGRATION - PRE-MIGRATION ERROR CHECKING AND REPORT ;11/30/06  12:22
 ;;2.0;Patient Representative;**19**;07/25/1995;Build 55
EN ; Build report of errors found in both Reference Tables and ROCs.
 ; Make sure list of valid sites has been downloaded from the EMC
 W !!,"This option is meant to be run prior to migrating data from the old",!,"Patient Representative System in VistA to the new PATS system."
 W !!,"The option reports data problems that would prevent the data from",!,"being migrated.",!
 I '$D(^XTMP("QACMIGR","STDINSTITUTION")) D STAERR^QACI2 Q
 Q:$$ASK'=1
 W !
 N QACI0 S QACI0=1
 D EN0^QACI2
 F TYPE="ROC","HL","USER","PT","CC","EMPINV","FSOS" D
 . K ^XTMP("QACMIGR",TYPE,"U") Q
 Q
 ;
 ;
ASK() ; Ask user whether they're sure they want to run the option.
 N DIR,X,Y
 S DIR(0)="YO",DIR("A")="Are you sure",DIR("B")="YES"
 D ^DIR
 Q Y
 ;
 ;


text:
<?xml version="1.0"?>
<!DOCTYPE view_configuration SYSTEM "..\viewconfig.dtd">

<!--
        Show instruction TLB (ITLB) measurements
        View configuration
        Date:    28 June 2010
        Version: 1.2
        Processor: AMD family 15h
        Copyright (c) 2006-2010 Advanced Micro Devices, Inc.
                      All rights reserved.

        The itlb_focus view should be available when instruction cache fetches,
        ITLB_L1M_L2H, ITLB_L1M_L2M, and retired instructions are available.

        Related data collection configurations: inst_access.xml
 -->

<view_configuration>

  <view name="ITLB report"
        separate_cpus="F"
        separate_processes="F"
        separate_threads="F"
        >

    <data>
      <event id="Ret_instructions" select="c0" mask="00" />
      <event id="IC_fetches"       select="80" mask="00" />
      <event id="ITLB_L1M_L2H"     select="84" mask="00" />
      <event id="ITLB_L1M_L2M"     select="85" mask="07" />
    </data>

    <output>
      <column title="Ret inst" sort="descending">
        <value id="Ret_instructions" />
      </column>
      <column title="ITLB requests" sort="none">
        <value id="IC_fetches" />
      </column>
      <column title="ITLB L1M L2H" sort="none">
        <value id="ITLB_L1M_L2H" />
      </column>
      <column title="ITLB L1M L2M" sort="none">
        <value id="ITLB_L1M_L2M" />
      </column>
      <column title="ITLB request rate" sort="none">
        <ratio left="IC_fetches" right="Ret_instructions" />
      </column>
      <column title="ITLB L1M L2H rate" sort="none">
        <ratio left="ITLB_L1M_L2H" right="Ret_instructions" />
      </column>
      <column title="ITLB L1M L2M rate" sort="none">
        <ratio left="ITLB_L1M_L2M" right="Ret_instructions" />
      </column>
    </output>

    <tool_tip>
      Show detailed ITLB measurements
    </tool_tip>

    <description>
      Use this view to analyze and break out instruction translation
      lookaside buffer (ITLB) miss rates by level (L1 and L2.)
    </description>

  </view>

</view_configuration>


text:
import vue from 'vue'
import orderDetail from './order-detail.vue'

let vm = new vue({
	el: '#order-detail',
	components: {
		'order-detail': orderDetail,
	},
})


text:
platform_is :windows do
  require 'win32ole'

  describe "WIN32OLE#ole_obj_help" do
    before :each do
      @ie = WIN32OLE.new('InternetExplorer.Application')
    end

    after :each do
      @ie.Quit
    end

    it "raises ArgumentError if argument is given" do
      lambda { @ie.ole_obj_help(1) }.should raise_error ArgumentError
    end

    it "returns an instance of WIN32OLE_TYPE" do
      @ie.ole_obj_help.kind_of?(WIN32OLE_TYPE).should be_true
    end
  end
end


text:
/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package io.theothercompany.theothertodolist.web;

import io.theothercompany.theothertodolist.service.AtService;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

/**
 *
 * @author The Other Company
 */
@Controller
public class AtController {
    @Autowired
    protected AtService atService;

    @RequestMapping(value = "/rest/at/populate.json", method = RequestMethod.GET, produces = MediaType.APPLICATION_JSON_VALUE)
    @ResponseBody
    public Map<String, List<Integer>> populateItems() {
        atService.parse(1, "pears @grocery");
        atService.parse(2, "Spring @dev");
        atService.parse(3, "something @test @ignored");
        return atService.collectTypes();
    }

    @RequestMapping(value = "/rest/at/list.json", method = RequestMethod.GET, produces = MediaType.APPLICATION_JSON_VALUE)
    @ResponseBody
    public Map<String, List<Integer>> getTodoItems() {
        return atService.collectTypes();
    }
}


text:
<!DOCTYPE html>

	<meta charset="utf-8">

	<meta name='viewport' content='width=device-width, initial-scale=1'>

	<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css">

	<link rel="stylesheet" href="styles.css">

	<title>Bootstrap</title>

HEADER WITH HALF CONTAINER

FLUID CONTAINER

This is a fluid container with a class='container-fluid'. Always adjusts to the size of the device.
		Lorem ipsum dolor sit amet, consectetur adipisicing elit. Cumque ipsum accusamus earum illum reiciendis quae dicta quo asperiores voluptatum, tempore, consequatur, obcaecati! Mollitia minus ea itaque. Atque molestiae quod perspiciatis.

FIXED-CONTAINER

This is a fixed container with a class='container'. Show a specific size depending the size of the viewport. Adjusts to media query breakpoints - Extra Small, Small, Medium and Large.
		Lorem ipsum dolor sit amet, consectetur adipisicing elit. Cumque ipsum accusamus earum illum reiciendis quae dicta quo asperiores voluptatum, tempore, consequatur, obcaecati! Mollitia minus ea itaque. Atque molestiae quod perspiciatis.


text:
set -e
set -u
set -o pipefail

GATEWAY_API="${GATEWAY_API:-false}"

# @setup profile=default

kubectl label namespace default istio-injection- --overwrite

# Generate client and server certificates and keys
snip_generate_client_and_server_certificates_and_keys_1
snip_generate_client_and_server_certificates_and_keys_2

# Deploy an NGINX server
snip_deploy_an_nginx_server_1

snip_deploy_an_nginx_server_2

snip_deploy_an_nginx_server_3

snip_deploy_an_nginx_server_4

# waiting for nginx deployment to start
_wait_for_deployment default my-nginx

# validate NGINX server was deployed successfully
_verify_contains snip_deploy_an_nginx_server_5 "subject: CN=nginx.example.com"

# configure an ingress gateway
if [ "$GATEWAY_API" == "true" ]; then
    snip_configure_an_ingress_gateway_2
    _wait_for_gateway default mygateway

    snip_configure_an_ingress_gateway_4
    snip_configure_an_ingress_gateway_5

    echo "addr: ${INGRESS_HOST}:${SECURE_INGRESS_PORT}"
else
    snip_configure_an_ingress_gateway_1
    snip_configure_an_ingress_gateway_3

    # wait for configuration to propagate
    _wait_for_istio gateway default mygateway
    _wait_for_istio virtualservice default nginx

    _set_ingress_environment_variables
fi

# validate the output
_verify_contains snip_configure_an_ingress_gateway_6 "SSL certificate verify ok."

# @cleanup
if [ "$GATEWAY_API" != "true" ]; then
    snip_cleanup_1
    snip_cleanup_3
    snip_cleanup_4
fi


text:
using JustLogic.Core;
using System.Collections.Generic;
using UnityEngine;

[UnitMenu("Camera/Set Target Texture")]
[UnitFriendlyName("Camera.Set Target Texture")]
[UnitUsage(typeof(RenderTexture))]
public class JLCameraSetTargetTexture : JLExpression
{
    [Parameter(ExpressionType = typeof(Camera))]
    public JLExpression OperandValue;

    [Parameter(ExpressionType = typeof(RenderTexture))]
    public JLExpression Value;

    public override object GetAnyResult(IExecutionContext context)
    {
        Camera opValue = OperandValue.GetResult<Camera>(context);
        return opValue.targetTexture = Value.GetResult<RenderTexture>(context);
    }
}


text:
class BatteryWidget : public QProgressBar
{
    Q_OBJECT

    int warnLevel;
    QPixmap backgroundPixmap;
    int backgroundImageWidth;
    int backgroundImageHeight;
    int maxChargeLevelRectWidth;
    int chargeLevelRectHeight;
    std::map< qreal, QColor > normalGradMap, warnGradMap;
    QFont font;
    std::unique_ptr< QFontMetrics > fontMetric;

protected:
    void paintEvent(QPaintEvent *) override;

public:
    explicit BatteryWidget(QWidget *parent = 0);

    void setWarnLevel(int level);

public slots:
    void setValue(int value);
};

#endif // BATTERY_WIDGET_H


text:
package com.quark.rest.controller;

import com.quark.common.dto.SocketMessage;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.messaging.simp.annotation.SendToUser;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Author LHR
 * Create By 2017/9/6
 */

@Api(value = "WebSocket接口", description = "用于服务器端对客户都进行消息推送")
@RestController
public class WebSocketController {

    public SimpMessagingTemplate template;

    @Autowired
    public WebSocketController(SimpMessagingTemplate template) {
        this.template = template;
    }

    @ApiOperation("WebSocket广播接口:客户端可以在/topic/all监听并接受服务端发回的消息")
    @SendTo("/topic/all")
    public SocketMessage sendToAll(SocketMessage message){
        return message;
    }

    @ApiOperation("WebSocket单播:客户端接收一对一消息的主题应该是"/user/" + 用户Id + "/message" ")
    @SendToUser("/message")
    public SocketMessage sendToUser(SocketMessage message){
        return message;
    }

}


text:
namespace demo
{

namespace res
{

// MEMBER FUNCTIONS
void TextureFactory::create( const String& path, rndr::Texture::Type type,
                             rndr::TexturePtr out )
{
    assert( out != nullptr );

    const char* filePath = path.c_str();

    // load image using FreeImage library
    FREE_IMAGE_FORMAT format = FreeImage_GetFileType( filePath, 0 );

    // ensure the image was found
    assert( format != 1 );

    // use file name to determine type if no magic number found
    if ( format == FIF_UNKNOWN )
    {
        // get format from file type and ensure it is supported
        format = FreeImage_GetFIFFromFilename( filePath );
        assert( FreeImage_FIFSupportsReading( format ) );
    }

    // determine the desired bits-per-pixel for the texture type
    uint8 desiredBpp;
    switch ( type )
    {
        case rndr::Texture::Type::BUMP:
        case rndr::Texture::Type::DIFFUSE:
        case rndr::Texture::Type::SPECULAR:
        default:
            desiredBpp = 32;
            break;
    }

    // read image
    FIBITMAP* bitmap = FreeImage_Load( format, filePath );
    FIBITMAP* formattedBitmap;

    bool isDesiredBpp = FreeImage_GetBPP( bitmap ) == desiredBpp;
    if ( isDesiredBpp )
    {
        formattedBitmap = bitmap;
    }
    else
    {
        switch (desiredBpp)
        {
            // 8-bit images (height maps, masks, transparency maps, etc.)
            case 8:
                formattedBitmap = FreeImage_ConvertTo8Bits( bitmap );
                break;

            // 24-bit images (most common alpha-less images)
            case 24:
                formattedBitmap = FreeImage_ConvertTo24Bits( bitmap );
                break;

            // 32-bit images (images with alpha [unnecessary in most cases])
            case 32:
            default:
                formattedBitmap = FreeImage_ConvertTo32Bits( bitmap );
                break;
        }
    }

    uint32 imageWidth = FreeImage_GetWidth( formattedBitmap );
    uint32 imageHeight = FreeImage_GetHeight( formattedBitmap );
    uint8* imageData = FreeImage_GetBits( formattedBitmap );
    uint32 size = imageWidth * imageHeight * ( desiredBpp / 8 );

    // load texture with data
    out->load( type, cntr::FixedArray<uint8>::copy( imageData, size, size ),
               imageWidth, imageHeight, desiredBpp );

    // release FreeImage resources
    FreeImage_Unload( formattedBitmap );
    if ( !isDesiredBpp )
    {
        FreeImage_Unload( bitmap );
    }
}

} // End nspc res

} // End nspc demo


text:
var MyClass = /** @class */ (function () {
    function MyClass(value) {
        this.value = value;
    }
    MyClass.prototype.getValue = function () {
        return this.value ? this.value : "not assigned";
    };
    return MyClass;
}());
function showName(ctr) {
    console.log("name:" + ctr["name"]);
}
function showFunc(ctr) {
    if (typeof ctr === "function")
        console.log(ctr["name"]);
    else
        console.log("not a func");
}
showName(MyClass);
var a = new MyClass();
var b = new MyClass("this a value");
console.log(a.getValue());
console.log(b.getValue());
console.log(typeof MyClass);
showFunc(MyClass);


text:
import noop from 'lodash-es/noop';
import reduce from 'lodash-es/reduce';
import {createLogicMiddleware} from 'redux-logic';
import configureStore from 'redux-mock-store';
import {first} from 'rxjs/operators';

import rootReducer from '../../reducers';

export function makeTestLogic(logic) {
  return async (action, {state = {}, afterDispatch = noop} = {}) => {
    const logicMiddleware = createLogicMiddleware([logic]);
    const mockStore = configureStore([logicMiddleware]);
    const store = mockStore(state);

    const logicDone = logicMiddleware.monitor$
      .pipe(first(({op}) => op === 'end'))
      .toPromise();

    store.dispatch(action);
    await Promise.resolve();
    afterDispatch(store);
    await logicDone;

    const dispatch = jest.fn();
    for (const dispatchedAction of store.getActions()) {
      dispatch(dispatchedAction);
    }
    return dispatch;
  };
}

export function applyActions(...actions) {
  return reduce(
    actions,
    (state, action) => rootReducer(state, action),
    undefined,
  );
}


text:
.class public final Landroid/provider/ContactsContract$CommonDataKinds$Callable;
.super Ljava/lang/Object;
.source "ContactsContract.java"

# interfaces
.implements Landroid/provider/ContactsContract$DataColumnsWithJoins;
.implements Landroid/provider/ContactsContract$CommonDataKinds$CommonColumns;
.implements Landroid/provider/ContactsContract$ContactCounts;

# annotations
.annotation system Ldalvik/annotation/EnclosingClass;
    value = Landroid/provider/ContactsContract$CommonDataKinds;
.end annotation

.annotation system Ldalvik/annotation/InnerClass;
    accessFlags = 0x19
    name = "Callable"
.end annotation

# static fields
.field public static final CONTENT_FILTER_URI:Landroid/net/Uri;

.field public static final CONTENT_URI:Landroid/net/Uri;

# direct methods
.method static constructor <clinit>()V
    .locals 2

    .prologue
    .line 7327
    sget-object v0, Landroid/provider/ContactsContract$Data;->CONTENT_URI:Landroid/net/Uri;

    .line 7328
    const-string/jumbo v1, "callables"

    .line 7327
    invoke-static {v0, v1}, Landroid/net/Uri;->withAppendedPath(Landroid/net/Uri;Ljava/lang/String;)Landroid/net/Uri;

    move-result-object v0

    sput-object v0, Landroid/provider/ContactsContract$CommonDataKinds$Callable;->CONTENT_URI:Landroid/net/Uri;

    .line 7333
    sget-object v0, Landroid/provider/ContactsContract$CommonDataKinds$Callable;->CONTENT_URI:Landroid/net/Uri;

    .line 7334
    const-string/jumbo v1, "filter"

    .line 7333
    invoke-static {v0, v1}, Landroid/net/Uri;->withAppendedPath(Landroid/net/Uri;Ljava/lang/String;)Landroid/net/Uri;

    move-result-object v0

    sput-object v0, Landroid/provider/ContactsContract$CommonDataKinds$Callable;->CONTENT_FILTER_URI:Landroid/net/Uri;

    .line 7321
    return-void
.end method

.method public constructor <init>()V
    .locals 0

    .prologue
    .line 7321
    invoke-direct {p0}, Ljava/lang/Object;-><init>()V

    return-void
.end method


text:
package org.jgroups.tests;

import org.jgroups.*;
import org.jgroups.protocols.TP;
import org.jgroups.util.Util;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Tests the concurrent stack (TP)
 * @author Bela Ban
 */
@Test(groups=Global.STACK_DEPENDENT,sequential=true)
public class FifoOrderTest extends ChannelTestBase {
    JChannel ch1, ch2, ch3;
    final static int NUM=25, EXPECTED=NUM * 3;
    final static long SLEEPTIME=100;
    CyclicBarrier barrier;

    @BeforeMethod
    void setUp() throws Exception {
        barrier=new CyclicBarrier(4);
        ch1=createChannel(true,3, "A");
        ch2=createChannel(ch1,    "B");
        ch3=createChannel(ch1,    "C");
    }

    @AfterMethod
    protected void tearDown() throws Exception {
        Util.close(ch3, ch2, ch1);
        barrier.reset();
    }

    @Test
    public void testFifoDelivery() throws Exception {
        long start, stop, diff;
        modifyDefaultThreadPool(ch1);
        modifyDefaultThreadPool(ch2);
        modifyDefaultThreadPool(ch3);

        MyReceiver r1=new MyReceiver("R1"), r2=new MyReceiver("R2"), r3=new MyReceiver("R3");
        ch1.setReceiver(r1); ch2.setReceiver(r2); ch3.setReceiver(r3);

        ch1.connect("FifoOrderTest");
        ch2.connect("FifoOrderTest");
        ch3.connect("FifoOrderTest");
        View v=ch3.getView();
        assert v.size() == 3 : "view is " + v;

        new Thread(new Sender(ch1)) {}.start();
        new Thread(new Sender(ch2)) {}.start();
        new Thread(new Sender(ch3)) {}.start();
        barrier.await(); // start senders
        start=System.currentTimeMillis();

        Exception ex=null;

        try {
            System.out.println("waiting for all messages to be received");
            barrier.await((long)(EXPECTED * SLEEPTIME * 1.5), TimeUnit.MILLISECONDS); // wait for all receivers
        }
        catch(java.util.concurrent.TimeoutException e) {
            ex=e;
        }

        stop=System.currentTimeMillis();
        diff=stop - start;

        System.out.println("Total time: " + diff + " ms\n");

        checkFIFO(r1);
        checkFIFO(r2);
        checkFIFO(r3);
        if(ex != null)
            throw ex;
    }

    private void checkFIFO(MyReceiver r) {
        Map<Address,List<Integer>> map=r.getMessages();

        boolean fifo=true;
        List<Address> incorrect_receivers=new LinkedList<Address>();
        System.out.println("Checking FIFO for " + r.getName() + ":");
        for(Address addr: map.keySet()) {
            List<Integer> list=map.get(addr);
            print(addr, list);
            if(!verifyFIFO(list)) {
                fifo=false;
                incorrect_receivers.add(addr);
            }
        }
        System.out.print("\n");

        if(!fifo)
            assert false : "The following receivers didn't receive all messages in FIFO order: " + incorrect_receivers;
    }

    private static boolean verifyFIFO(List<Integer> list) {
        List<Integer> list2=new LinkedList<Integer>(list);
        Collections.sort(list2);
        return list.equals(list2);
    }

    private static void print(Address addr, List<Integer> list) {
        StringBuilder sb=new StringBuilder();
        sb.append(addr).append(": ");
        for(Integer i: list)
            sb.append(i).append(" ");
        System.out.println(sb);
    }

    private static void modifyDefaultThreadPool(JChannel ch1) {
        TP transport=ch1.getProtocolStack().getTransport();
        ThreadPoolExecutor default_pool=(ThreadPoolExecutor)transport.getDefaultThreadPool();
        if(default_pool != null) {
            default_pool.setCorePoolSize(1);
            default_pool.setMaximumPoolSize(100);
        }
        transport.setThreadPoolQueueEnabled(false);
    }

    private class Sender implements Runnable {
        final Channel ch;
        final Address local_addr;

        public Sender(Channel ch) {
            this.ch=ch;
            local_addr=ch.getAddress();
        }

        public void run() {
            Message msg;
            try {
                barrier.await();
            }
            catch(Throwable t) {
                return;
            }

            for(int i=1; i <= NUM; i++) {
                msg=new Message(null, null, new Integer(i));
                try {
                    ch.send(msg);
                }
                catch(Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private class Pair<K,V> {
        K key;
        V val;

        public Pair(K key, V val) {
            this.key=key;
            this.val=val;
        }

        public String toString() {
            return key + "::" + val;
        }
    }

    private class MyReceiver extends ReceiverAdapter {
        String name;
        final ConcurrentMap<Address,List<Integer>> msgs=new ConcurrentHashMap<Address,List<Integer>>();

        AtomicInteger count=new AtomicInteger(0);

        public MyReceiver(String name) {
            this.name=name;
        }

        public void receive(Message msg) {
            Util.sleep(SLEEPTIME);

            Address sender=msg.getSrc();
            List<Integer> list=msgs.get(sender);
            if(list == null) {
                list=new LinkedList<Integer>();
                List<Integer> tmp=msgs.putIfAbsent(sender, list);
                if(tmp != null)
                    list=tmp;
            }
            Integer num=(Integer)msg.getObject();
            list.add(num); // no concurrent access: FIFO per sender ! (No need to synchronize on list)

            if(count.incrementAndGet() >= EXPECTED) {
                System.out.println("[" + name + "]: received all messages (" + count.get() + ")");
                try {
                    barrier.await();
                }
                catch(Exception e) {
                    e.printStackTrace();
                }
            }
        }

        public ConcurrentMap<Address,List<Integer>> getMessages() {return msgs;}

        public String getName() {
            return name;
        }
    }

}


text:
.class public final Lcom/android/internal/R$color;
.super Ljava/lang/Object;
.source "R.java"

# annotations
.annotation system Ldalvik/annotation/EnclosingClass;
    value = Lcom/android/internal/R;
.end annotation

.annotation system Ldalvik/annotation/InnerClass;
    accessFlags = 0x19
    name = "color"
.end annotation

# static fields
.field public static final background_dark:I = 0x106000e

.field public static final background_light:I = 0x106000f

.field public static final battery_saver_mode_color:I = 0x106005d

.field public static final black:I = 0x106000c

.field public static final config_defaultNotificationColor:I = 0x106010f

.field public static final darker_gray:I = 0x1060000

.field public static final datepicker_default_circle_background_color_holo_light:I = 0x10600b2

.field public static final datepicker_default_circle_background_color_material_light:I = 0x106010b

.field public static final datepicker_default_disabled_text_color_holo_light:I = 0x10600ac

.field public static final datepicker_default_disabled_text_color_material_light:I = 0x1060105

.field public static final datepicker_default_header_dayofweek_background_color_holo_light:I = 0x10600a8

.field public static final datepicker_default_header_dayofweek_background_color_material_light:I = 0x1060101

.field public static final datepicker_default_header_selector_background_holo_light:I = 0x10600a6

.field public static final datepicker_default_header_selector_background_material_light:I = 0x10600ff

.field public static final datepicker_default_normal_text_color_holo_light:I = 0x10600aa

.field public static final datepicker_default_normal_text_color_material_light:I = 0x1060103

.field public static final holo_blue_bright:I = 0x106001b

.field public static final holo_blue_dark:I = 0x1060013

.field public static final holo_blue_light:I = 0x1060012

.field public static final holo_green_dark:I = 0x1060015

.field public static final holo_green_light:I = 0x1060014

.field public static final holo_orange_dark:I = 0x1060019

.field public static final holo_orange_light:I = 0x1060018

.field public static final holo_purple:I = 0x106001a

.field public static final holo_red_dark:I = 0x1060017

.field public static final holo_red_light:I = 0x1060016

.field public static final input_method_navigation_guard:I = 0x1060035

.field public static final lock_pattern_view_error_color:I = 0x106004f

.field public static final lock_pattern_view_regular_color:I = 0x106004d

.field public static final lock_pattern_view_success_color:I = 0x106004e

.field public static final notification_action_color_filter:I = 0x1060053

.field public static final notification_icon_bg_color:I = 0x1060052

.field public static final notification_media_primary_color:I = 0x1060054

.field public static final notification_media_secondary_color:I = 0x1060055

.field public static final notification_progress_background_color:I = 0x1060056

.field public static final primary_text_dark:I = 0x1060001

.field public static final primary_text_dark_nodisable:I = 0x1060002

.field public static final primary_text_light:I = 0x1060003

.field public static final primary_text_light_nodisable:I = 0x1060004

.field public static final secondary_text_dark:I = 0x1060005

.field public static final secondary_text_dark_nodisable:I = 0x1060006

.field public static final secondary_text_light:I = 0x1060007

.field public static final secondary_text_light_nodisable:I = 0x1060008

.field public static final system_notification_accent_color:I = 0x106005c

.field public static final tab_indicator_text:I = 0x1060009

.field public static final tab_indicator_text_v4:I = 0x106013c

.field public static final tertiary_text_dark:I = 0x1060010

.field public static final tertiary_text_light:I = 0x1060011

.field public static final timepicker_default_ampm_selected_background_color_holo_light:I = 0x10600a2

.field public static final timepicker_default_ampm_selected_background_color_material:I = 0x10600fb

.field public static final timepicker_default_ampm_unselected_background_color_holo_light:I = 0x10600a4

.field public static final timepicker_default_ampm_unselected_background_color_material:I = 0x10600fc

.field public static final timepicker_default_numbers_background_color_material:I = 0x10600fe

.field public static final timepicker_default_selector_color_material:I = 0x10600fd

.field public static final timepicker_default_text_color_holo_light:I = 0x10600a0

.field public static final timepicker_default_text_color_material:I = 0x10600fa

.field public static final transparent:I = 0x106000d

.field public static final user_icon_1:I = 0x106005e

.field public static final user_icon_2:I = 0x106005f

.field public static final user_icon_3:I = 0x1060060

.field public static final user_icon_4:I = 0x1060061

.field public static final user_icon_5:I = 0x1060062

.field public static final user_icon_6:I = 0x1060063

.field public static final user_icon_7:I = 0x1060064

.field public static final user_icon_8:I = 0x1060065

.field public static final user_icon_default_gray:I = 0x1060066

.field public static final user_icon_default_white:I = 0x1060067

.field public static final white:I = 0x106000b

.field public static final widget_edittext_dark:I = 0x106000a

# direct methods
.method public constructor <init>()V
    .locals 0

    .prologue
    .line 14873
    invoke-direct {p0}, Ljava/lang/Object;-><init>()V

    return-void
.end method


text:
package org.apache.flink.streaming.connectors.kafka.table;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.KafkaTestBase;
import org.apache.flink.streaming.connectors.kafka.KafkaTestBaseWithFlink;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.descriptors.KafkaValidator;
import org.apache.flink.test.util.SuccessException;
import org.apache.flink.types.Row;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.collectRows;
import static org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestUtils.readLines;
import static org.apache.flink.table.utils.TableTestMatchers.deepEqualTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;

/** Basic IT cases for the Kafka table source and sink. */
@RunWith(Parameterized.class)
public class KafkaTableITCase extends KafkaTestBaseWithFlink {

    private static final String JSON_FORMAT = "json";
    private static final String AVRO_FORMAT = "avro";
    private static final String CSV_FORMAT = "csv";

    @Parameterized.Parameter public boolean isLegacyConnector;

    @Parameterized.Parameter(1)
    public String format;

    @Parameterized.Parameters(name = "legacy = {0}, format = {1}")
    public static Object[] parameters() {
        return new Object[][] {
            // cover all 3 formats for new and old connector
            new Object[] {false, JSON_FORMAT},
            new Object[] {false, AVRO_FORMAT},
            new Object[] {false, CSV_FORMAT},
            new Object[] {true, JSON_FORMAT},
            new Object[] {true, AVRO_FORMAT},
            new Object[] {true, CSV_FORMAT}
        };
    }

    protected StreamExecutionEnvironment env;
    protected StreamTableEnvironment tEnv;

    @Before
    public void setup() {
        env = StreamExecutionEnvironment.getExecutionEnvironment();
        tEnv =
                StreamTableEnvironment.create(
                        env,
                        EnvironmentSettings.newInstance()
                                // Watermark is only supported in blink planner
                                .useBlinkPlanner()
                                .inStreamingMode()
                                .build());
        env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
        // we have to use single parallelism,
        // because we will count the messages in sink to terminate the job
        env.setParallelism(1);
    }

    @Test
    public void testKafkaSourceSink() throws Exception {
        // we always use a different topic name for each parameterized topic,
        // in order to make sure the topic can be created.
        final String topic = "tstopic_" + format + "_" + isLegacyConnector;
        createTestTopic(topic, 1, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");

        final String createTable;
        if (!isLegacyConnector) {
            createTable =
                    String.format(
                            "create table kafka (\n"
                                    + "  `computed-price` as price + 1.0,\n"
                                    + "  price decimal(38, 18),\n"
                                    + "  currency string,\n"
                                    + "  log_date date,\n"
                                    + "  log_time time(3),\n"
                                    + "  log_ts timestamp(3),\n"
                                    + "  ts as log_ts + INTERVAL '1' SECOND,\n"
                                    + "  watermark for ts as ts\n"
                                    + ") with (\n"
                                    + "  'connector' = '%s',\n"
                                    + "  'topic' = '%s',\n"
                                    + "  'properties.bootstrap.servers' = '%s',\n"
                                    + "  'properties.group.id' = '%s',\n"
                                    + "  'scan.startup.mode' = 'earliest-offset',\n"
                                    + "  %s\n"
                                    + ")",
                            KafkaDynamicTableFactory.IDENTIFIER,
                            topic,
                            bootstraps,
                            groupId,
                            formatOptions());
        } else {
            createTable =
                    String.format(
                            "create table kafka (\n"
                                    + "  `computed-price` as price + 1.0,\n"
                                    + "  price decimal(38, 18),\n"
                                    + "  currency string,\n"
                                    + "  log_date date,\n"
                                    + "  log_time time(3),\n"
                                    + "  log_ts timestamp(3),\n"
                                    + "  ts as log_ts + INTERVAL '1' SECOND,\n"
                                    + "  watermark for ts as ts\n"
                                    + ") with (\n"
                                    + "  'connector.type' = 'kafka',\n"
                                    + "  'connector.version' = '%s',\n"
                                    + "  'connector.topic' = '%s',\n"
                                    + "  'connector.properties.bootstrap.servers' = '%s',\n"
                                    + "  'connector.properties.group.id' = '%s',\n"
                                    + "  'connector.startup-mode' = 'earliest-offset',\n"
                                    + "  'update-mode' = 'append',\n"
                                    + "  %s\n"
                                    + ")",
                            KafkaValidator.CONNECTOR_VERSION_VALUE_UNIVERSAL,
                            topic,
                            bootstraps,
                            groupId,
                            formatOptions());
        }

        tEnv.executeSql(createTable);

        String initialValues =
                "INSERT INTO kafka\n"
                        + "SELECT CAST(price AS DECIMAL(10, 2)), currency, "
                        + " CAST(d AS DATE), CAST(t AS TIME(0)), CAST(ts AS TIMESTAMP(3))\n"
                        + "FROM (VALUES (2.02,'Euro','2019-12-12', '00:00:01', '2019-12-12 00:00:01.001001'), \n"
                        + "  (1.11,'US Dollar','2019-12-12', '00:00:02', '2019-12-12 00:00:02.002001'), \n"
                        + "  (50,'Yen','2019-12-12', '00:00:03', '2019-12-12 00:00:03.004001'), \n"
                        + "  (3.1,'Euro','2019-12-12', '00:00:04', '2019-12-12 00:00:04.005001'), \n"
                        + "  (5.33,'US Dollar','2019-12-12', '00:00:05', '2019-12-12 00:00:05.006001'), \n"
                        + "  (0,'DUMMY','2019-12-12', '00:00:10', '2019-12-12 00:00:10'))\n"
                        + "  AS orders (price, currency, d, t, ts)";
        tEnv.executeSql(initialValues).await();

        // ---------- Consume stream from Kafka -------------------

        String query =
                "SELECT\n"
                        + "  CAST(TUMBLE_END(ts, INTERVAL '5' SECOND) AS VARCHAR),\n"
                        + "  CAST(MAX(log_date) AS VARCHAR),\n"
                        + "  CAST(MAX(log_time) AS VARCHAR),\n"
                        + "  CAST(MAX(ts) AS VARCHAR),\n"
                        + "  COUNT(*),\n"
                        + "  CAST(MAX(price) AS DECIMAL(10, 2))\n"
                        + "FROM kafka\n"
                        + "GROUP BY TUMBLE(ts, INTERVAL '5' SECOND)";

        DataStream<RowData> result = tEnv.toAppendStream(tEnv.sqlQuery(query), RowData.class);
        TestingSinkFunction sink = new TestingSinkFunction(2);
        result.addSink(sink).setParallelism(1);

        try {
            env.execute("Job_2");
        } catch (Throwable e) {
            // we have to use a specific exception to indicate the job is finished,
            // because the registered Kafka source is infinite.
            if (!isCausedByJobFinished(e)) {
                // re-throw
                throw e;
            }
        }

        List<String> expected =
                Arrays.asList(
                        "+I(2019-12-12 00:00:05.000,2019-12-12,00:00:03,2019-12-12 00:00:04.004,3,50.00)",
                        "+I(2019-12-12 00:00:10.000,2019-12-12,00:00:05,2019-12-12 00:00:06.006,2,5.33)");

        assertEquals(expected, TestingSinkFunction.rows);

        // ------------- cleanup -------------------

        deleteTestTopic(topic);
    }

    @Test
    public void testKafkaTableWithMultipleTopics() throws Exception {
        if (isLegacyConnector) {
            return;
        }
        // ---------- create source and sink tables -------------------
        String tableTemp =
                "create table %s (\n"
                        + "  currency string\n"
                        + ") with (\n"
                        + "  'connector' = '%s',\n"
                        + "  'topic' = '%s',\n"
                        + "  'properties.bootstrap.servers' = '%s',\n"
                        + "  'properties.group.id' = '%s',\n"
                        + "  'scan.startup.mode' = 'earliest-offset',\n"
                        + "  %s\n"
                        + ")";
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");
        List<String> currencies = Arrays.asList("Euro", "Dollar", "Yen", "Dummy");
        List<String> topics =
                currencies.stream()
                        .map(currency -> String.format("%s_%s", currency, format))
                        .collect(Collectors.toList());
        // Because kafka connector currently doesn't support write data into multiple topic
        // together,
        // we have to create multiple sink tables.
        IntStream.range(0, 4)
                .forEach(
                        index -> {
                            createTestTopic(topics.get(index), 1, 1);
                            tEnv.executeSql(
                                    String.format(
                                            tableTemp,
                                            currencies.get(index).toLowerCase(),
                                            KafkaDynamicTableFactory.IDENTIFIER,
                                            topics.get(index),
                                            bootstraps,
                                            groupId,
                                            formatOptions()));
                        });
        // create source table
        tEnv.executeSql(
                String.format(
                        tableTemp,
                        "currencies_topic_list",
                        KafkaDynamicTableFactory.IDENTIFIER,
                        String.join(";", topics),
                        bootstraps,
                        groupId,
                        formatOptions()));

        // ---------- Prepare data in Kafka topics -------------------
        String insertTemp =
                "INSERT INTO %s\n"
                        + "SELECT currency\n"
                        + " FROM (VALUES ('%s'))\n"
                        + " AS orders (currency)";
        currencies.forEach(
                currency -> {
                    try {
                        tEnv.executeSql(String.format(insertTemp, currency.toLowerCase(), currency))
                                .await();
                    } catch (Exception e) {
                        fail(e.getMessage());
                    }
                });

        // ------------- test the topic-list kafka source -------------------
        DataStream<RowData> result =
                tEnv.toAppendStream(
                        tEnv.sqlQuery("SELECT currency FROM currencies_topic_list"), RowData.class);
        TestingSinkFunction sink = new TestingSinkFunction(4); // expect to receive 4 records
        result.addSink(sink);

        try {
            env.execute("Job_3");
        } catch (Throwable e) {
            // we have to use a specific exception to indicate the job is finished,
            // because the registered Kafka source is infinite.
            if (!isCausedByJobFinished(e)) {
                // re-throw
                throw e;
            }
        }
        List<String> expected = Arrays.asList("+I(Dollar)", "+I(Dummy)", "+I(Euro)", "+I(Yen)");
        TestingSinkFunction.rows.sort(Comparator.naturalOrder());
        assertEquals(expected, TestingSinkFunction.rows);

        // ------------- cleanup -------------------
        topics.forEach(KafkaTestBase::deleteTestTopic);
    }

    @Test
    public void testKafkaSourceSinkWithMetadata() throws Exception {
        if (isLegacyConnector) {
            return;
        }
        // we always use a different topic name for each parameterized topic,
        // in order to make sure the topic can be created.
        final String topic = "metadata_topic_" + format;
        createTestTopic(topic, 1, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");

        final String createTable =
                String.format(
                        "CREATE TABLE kafka (\n"
                                + "  `physical_1` STRING,\n"
                                + "  `physical_2` INT,\n"
                                // metadata fields are out of order on purpose
                                // offset is ignored because it might not be deterministic
                                + "  `timestamp-type` STRING METADATA VIRTUAL,\n"
                                + "  `timestamp` TIMESTAMP(3) METADATA,\n"
                                + "  `leader-epoch` INT METADATA VIRTUAL,\n"
                                + "  `headers` MAP<STRING, BYTES> METADATA,\n"
                                + "  `partition` INT METADATA VIRTUAL,\n"
                                + "  `topic` STRING METADATA VIRTUAL,\n"
                                + "  `physical_3` BOOLEAN\n"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  %s\n"
                                + ")",
                        topic, bootstraps, groupId, formatOptions());

        tEnv.executeSql(createTable);

        String initialValues =
                "INSERT INTO kafka\n"
                        + "VALUES\n"
                        + " ('data 1', 1, TIMESTAMP '2020-03-08 13:12:11.123', MAP['k1', X'C0FFEE', 'k2', X'BABE'], TRUE),\n"
                        + " ('data 2', 2, TIMESTAMP '2020-03-09 13:12:11.123', CAST(NULL AS MAP<STRING, BYTES>), FALSE),\n"
                        + " ('data 3', 3, TIMESTAMP '2020-03-10 13:12:11.123', MAP['k1', X'10', 'k2', X'20'], TRUE)";
        tEnv.executeSql(initialValues).await();

        // ---------- Consume stream from Kafka -------------------

        final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM kafka"), 3);

        final Map<String, byte[]> headers1 = new HashMap<>();
        headers1.put("k1", new byte[] {(byte) 0xC0, (byte) 0xFF, (byte) 0xEE});
        headers1.put("k2", new byte[] {(byte) 0xBA, (byte) 0xBE});

        final Map<String, byte[]> headers3 = new HashMap<>();
        headers3.put("k1", new byte[] {(byte) 0x10});
        headers3.put("k2", new byte[] {(byte) 0x20});

        final List<Row> expected =
                Arrays.asList(
                        Row.of(
                                "data 1",
                                1,
                                "CreateTime",
                                LocalDateTime.parse("2020-03-08T13:12:11.123"),
                                0,
                                headers1,
                                0,
                                topic,
                                true),
                        Row.of(
                                "data 2",
                                2,
                                "CreateTime",
                                LocalDateTime.parse("2020-03-09T13:12:11.123"),
                                0,
                                Collections.emptyMap(),
                                0,
                                topic,
                                false),
                        Row.of(
                                "data 3",
                                3,
                                "CreateTime",
                                LocalDateTime.parse("2020-03-10T13:12:11.123"),
                                0,
                                headers3,
                                0,
                                topic,
                                true));

        assertThat(result, deepEqualTo(expected, true));

        // ------------- cleanup -------------------

        deleteTestTopic(topic);
    }

    @Test
    public void testKafkaSourceSinkWithKeyAndPartialValue() throws Exception {
        if (isLegacyConnector) {
            return;
        }
        // we always use a different topic name for each parameterized topic,
        // in order to make sure the topic can be created.
        final String topic = "key_partial_value_topic_" + format;
        createTestTopic(topic, 1, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");

        // k_user_id and user_id have different data types to verify the correct mapping,
        // fields are reordered on purpose
        final String createTable =
                String.format(
                        "CREATE TABLE kafka (\n"
                                + "  `k_user_id` BIGINT,\n"
                                + "  `name` STRING,\n"
                                + "  `timestamp` TIMESTAMP(3) METADATA,\n"
                                + "  `k_event_id` BIGINT,\n"
                                + "  `user_id` INT,\n"
                                + "  `payload` STRING\n"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'key.format' = '%s',\n"
                                + "  'key.fields' = 'k_event_id; k_user_id',\n"
                                + "  'key.fields-prefix' = 'k_',\n"
                                + "  'value.format' = '%s',\n"
                                + "  'value.fields-include' = 'EXCEPT_KEY'\n"
                                + ")",
                        topic, bootstraps, groupId, format, format);

        tEnv.executeSql(createTable);

        String initialValues =
                "INSERT INTO kafka\n"
                        + "VALUES\n"
                        + " (1, 'name 1', TIMESTAMP '2020-03-08 13:12:11.123', 100, 41, 'payload 1'),\n"
                        + " (2, 'name 2', TIMESTAMP '2020-03-09 13:12:11.123', 101, 42, 'payload 2'),\n"
                        + " (3, 'name 3', TIMESTAMP '2020-03-10 13:12:11.123', 102, 43, 'payload 3')";
        tEnv.executeSql(initialValues).await();

        // ---------- Consume stream from Kafka -------------------

        final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM kafka"), 3);

        final List<Row> expected =
                Arrays.asList(
                        Row.of(
                                1L,
                                "name 1",
                                LocalDateTime.parse("2020-03-08T13:12:11.123"),
                                100L,
                                41,
                                "payload 1"),
                        Row.of(
                                2L,
                                "name 2",
                                LocalDateTime.parse("2020-03-09T13:12:11.123"),
                                101L,
                                42,
                                "payload 2"),
                        Row.of(
                                3L,
                                "name 3",
                                LocalDateTime.parse("2020-03-10T13:12:11.123"),
                                102L,
                                43,
                                "payload 3"));

        assertThat(result, deepEqualTo(expected, true));

        // ------------- cleanup -------------------

        deleteTestTopic(topic);
    }

    @Test
    public void testKafkaSourceSinkWithKeyAndFullValue() throws Exception {
        if (isLegacyConnector) {
            return;
        }
        // we always use a different topic name for each parameterized topic,
        // in order to make sure the topic can be created.
        final String topic = "key_full_value_topic_" + format;
        createTestTopic(topic, 1, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");

        // compared to the partial value test we cannot support both k_user_id and user_id in a full
        // value due to duplicate names after key prefix stripping,
        // fields are reordered on purpose,
        // fields for keys and values are overlapping
        final String createTable =
                String.format(
                        "CREATE TABLE kafka (\n"
                                + "  `user_id` BIGINT,\n"
                                + "  `name` STRING,\n"
                                + "  `timestamp` TIMESTAMP(3) METADATA,\n"
                                + "  `event_id` BIGINT,\n"
                                + "  `payload` STRING\n"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'key.format' = '%s',\n"
                                + "  'key.fields' = 'event_id; user_id',\n"
                                + "  'value.format' = '%s',\n"
                                + "  'value.fields-include' = 'ALL'\n"
                                + ")",
                        topic, bootstraps, groupId, format, format);

        tEnv.executeSql(createTable);

        String initialValues =
                "INSERT INTO kafka\n"
                        + "VALUES\n"
                        + " (1, 'name 1', TIMESTAMP '2020-03-08 13:12:11.123', 100, 'payload 1'),\n"
                        + " (2, 'name 2', TIMESTAMP '2020-03-09 13:12:11.123', 101, 'payload 2'),\n"
                        + " (3, 'name 3', TIMESTAMP '2020-03-10 13:12:11.123', 102, 'payload 3')";
        tEnv.executeSql(initialValues).await();

        // ---------- Consume stream from Kafka -------------------

        final List<Row> result = collectRows(tEnv.sqlQuery("SELECT * FROM kafka"), 3);

        final List<Row> expected =
                Arrays.asList(
                        Row.of(
                                1L,
                                "name 1",
                                LocalDateTime.parse("2020-03-08T13:12:11.123"),
                                100L,
                                "payload 1"),
                        Row.of(
                                2L,
                                "name 2",
                                LocalDateTime.parse("2020-03-09T13:12:11.123"),
                                101L,
                                "payload 2"),
                        Row.of(
                                3L,
                                "name 3",
                                LocalDateTime.parse("2020-03-10T13:12:11.123"),
                                102L,
                                "payload 3"));

        assertThat(result, deepEqualTo(expected, true));

        // ------------- cleanup -------------------

        deleteTestTopic(topic);
    }

    @Test
    public void testKafkaTemporalJoinChangelog() throws Exception {
        if (isLegacyConnector) {
            return;
        }

        // Set the session time zone to UTC, because the next `METADATA FROM
        // 'value.source.timestamp'` DDL
        // will use the session time zone when convert the changelog time from milliseconds to
        // timestamp
        tEnv.getConfig().getConfiguration().setString("table.local-time-zone", "UTC");

        // we always use a different topic name for each parameterized topic,
        // in order to make sure the topic can be created.
        final String orderTopic = "temporal_join_topic_order_" + format;
        createTestTopic(orderTopic, 1, 1);

        final String productTopic = "temporal_join_topic_product_" + format;
        createTestTopic(productTopic, 1, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");

        // create order table and set initial values
        final String orderTableDDL =
                String.format(
                        "CREATE TABLE ordersTable (\n"
                                + "  order_id STRING,\n"
                                + "  product_id STRING,\n"
                                + "  order_time TIMESTAMP(3),\n"
                                + "  quantity INT,\n"
                                + "  purchaser STRING,\n"
                                + "  WATERMARK FOR order_time AS order_time - INTERVAL '1' SECOND\n"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'format' = '%s'\n"
                                + ")",
                        orderTopic, bootstraps, groupId, format);
        tEnv.executeSql(orderTableDDL);
        String orderInitialValues =
                "INSERT INTO ordersTable\n"
                        + "VALUES\n"
                        + "('o_001', 'p_001', TIMESTAMP '2020-10-01 00:01:00', 1, 'Alice'),"
                        + "('o_002', 'p_002', TIMESTAMP '2020-10-01 00:02:00', 1, 'Bob'),"
                        + "('o_003', 'p_001', TIMESTAMP '2020-10-01 12:00:00', 2, 'Tom'),"
                        + "('o_004', 'p_002', TIMESTAMP '2020-10-01 12:00:00', 2, 'King'),"
                        + "('o_005', 'p_001', TIMESTAMP '2020-10-01 18:00:00', 10, 'Leonard'),"
                        + "('o_006', 'p_002', TIMESTAMP '2020-10-01 18:00:00', 10, 'Leonard'),"
                        + "('o_007', 'p_002', TIMESTAMP '2020-10-01 18:00:01', 10, 'Robinson')"; // used to advance watermark
        tEnv.executeSql(orderInitialValues).await();

        // create product table and set initial values
        final String productTableDDL =
                String.format(
                        "CREATE TABLE productChangelogTable (\n"
                                + "  product_id STRING,\n"
                                + "  product_name STRING,\n"
                                + "  product_price DECIMAL(10, 4),\n"
                                + "  update_time TIMESTAMP(3) METADATA FROM 'value.source.timestamp' VIRTUAL,\n"
                                + "  PRIMARY KEY(product_id) NOT ENFORCED,\n"
                                + "  WATERMARK FOR update_time AS update_time - INTERVAL '1' SECOND\n"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'value.format' = 'debezium-json'\n"
                                + ")",
                        productTopic, bootstraps, groupId);
        tEnv.executeSql(productTableDDL);

        // use raw format to initial the changelog data
        initialProductChangelog(productTopic, bootstraps);

        // ---------- query temporal join result from Kafka -------------------
        final List<String> result =
                collectRows(
                                tEnv.sqlQuery(
                                        "SELECT"
                                                + "  order_id,"
                                                + "  order_time,"
                                                + "  P.product_id,"
                                                + "  P.update_time as product_update_time,"
                                                + "  product_price,"
                                                + "  purchaser,"
                                                + "  product_name,"
                                                + "  quantity,"
                                                + "  quantity * product_price AS order_amount "
                                                + "FROM ordersTable AS O "
                                                + "LEFT JOIN productChangelogTable FOR SYSTEM_TIME AS OF O.order_time AS P "
                                                + "ON O.product_id = P.product_id"),
                                6)
                        .stream()
                        .map(row -> row.toString())
                        .sorted()
                        .collect(Collectors.toList());

        final List<String> expected =
                Arrays.asList(
                        "+I[o_001, 2020-10-01T00:01, p_001, 1970-01-01T00:00, 11.1100, Alice, scooter, 1, 11.1100]",
                        "+I[o_002, 2020-10-01T00:02, p_002, 1970-01-01T00:00, 23.1100, Bob, basketball, 1, 23.1100]",
                        "+I[o_003, 2020-10-01T12:00, p_001, 2020-10-01T12:00, 12.9900, Tom, scooter, 2, 25.9800]",
                        "+I[o_004, 2020-10-01T12:00, p_002, 2020-10-01T12:00, 19.9900, King, basketball, 2, 39.9800]",
                        "+I[o_005, 2020-10-01T18:00, p_001, 2020-10-01T18:00, 11.9900, Leonard, scooter, 10, 119.9000]",
                        "+I[o_006, 2020-10-01T18:00, null, null, null, Leonard, null, 10, null]");

        assertEquals(expected, result);

        // ------------- cleanup -------------------

        deleteTestTopic(orderTopic);
        deleteTestTopic(productTopic);
    }

    private void initialProductChangelog(String topic, String bootstraps) throws Exception {
        String productChangelogDDL =
                String.format(
                        "CREATE TABLE productChangelog (\n"
                                + "  changelog STRING"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'format' = 'raw'\n"
                                + ")",
                        topic, bootstraps);
        tEnv.executeSql(productChangelogDDL);
        String[] allChangelog = readLines("product_changelog.txt").toArray(new String[0]);

        StringBuilder insertSqlSb = new StringBuilder();
        insertSqlSb.append("INSERT INTO productChangelog VALUES ");
        for (String log : allChangelog) {
            insertSqlSb.append("('" + log + "'),");
        }
        // trim the last comma
        String insertSql = insertSqlSb.substring(0, insertSqlSb.toString().length() - 1);
        tEnv.executeSql(insertSql).await();
    }

    @Test
    public void testPerPartitionWatermarkKafka() throws Exception {
        if (isLegacyConnector) {
            return;
        }
        // we always use a different topic name for each parameterized topic,
        // in order to make sure the topic can be created.
        final String topic = "per_partition_watermark_topic_" + format;
        createTestTopic(topic, 4, 1);

        // ---------- Produce an event time stream into Kafka -------------------
        String groupId = standardProps.getProperty("group.id");
        String bootstraps = standardProps.getProperty("bootstrap.servers");

        final String createTable =
                String.format(
                        "CREATE TABLE kafka (\n"
                                + "  `partition_id` INT,\n"
                                + "  `name` STRING,\n"
                                + "  `timestamp` TIMESTAMP(3),\n"
                                + "  WATERMARK FOR `timestamp` AS `timestamp`\n"
                                + ") WITH (\n"
                                + "  'connector' = 'kafka',\n"
                                + "  'topic' = '%s',\n"
                                + "  'properties.bootstrap.servers' = '%s',\n"
                                + "  'properties.group.id' = '%s',\n"
                                + "  'scan.startup.mode' = 'earliest-offset',\n"
                                + "  'sink.partitioner' = '%s',\n"
                                + "  'format' = '%s'\n"
                                + ")",
                        topic, bootstraps, groupId, TestPartitioner.class.getName(), format);

        tEnv.executeSql(createTable);

        // make every partition have more than one record
        String initialValues =
                "INSERT INTO kafka\n"
                        + "VALUES\n"
                        + " (0, 'partition-0-name-0', TIMESTAMP '2020-03-08 13:12:11.123'),\n"
                        + " (0, 'partition-0-name-1', TIMESTAMP '2020-03-08 14:12:12.223'),\n"
                        + " (0, 'partition-0-name-2', TIMESTAMP '2020-03-08 15:12:13.323'),\n"
                        + " (1, 'partition-1-name-0', TIMESTAMP '2020-03-09 13:13:11.123'),\n"
                        + " (1, 'partition-1-name-1', TIMESTAMP '2020-03-09 15:13:11.133'),\n"
                        + " (1, 'partition-1-name-2', TIMESTAMP '2020-03-09 16:13:11.143'),\n"
                        + " (2, 'partition-2-name-0', TIMESTAMP '2020-03-10 13:12:14.123'),\n"
                        + " (2, 'partition-2-name-1', TIMESTAMP '2020-03-10 14:12:14.123'),\n"
                        + " (2, 'partition-2-name-2', TIMESTAMP '2020-03-10 14:13:14.123'),\n"
                        + " (2, 'partition-2-name-3', TIMESTAMP '2020-03-10 14:14:14.123'),\n"
                        + " (2, 'partition-2-name-4', TIMESTAMP '2020-03-10 14:15:14.123'),\n"
                        + " (2, 'partition-2-name-5', TIMESTAMP '2020-03-10 14:16:14.123'),\n"
                        + " (3, 'partition-3-name-0', TIMESTAMP '2020-03-11 17:12:11.123'),\n"
                        + " (3, 'partition-3-name-1', TIMESTAMP '2020-03-11 18:12:11.123')";
        tEnv.executeSql(initialValues).await();

        // ---------- Consume stream from Kafka -------------------

        env.setParallelism(1);
        String createSink =
                "CREATE TABLE MySink(\n"
                        + "  id INT,\n"
                        + "  name STRING,\n"
                        + "  ts TIMESTAMP(3),\n"
                        + "  WATERMARK FOR ts as ts\n"
                        + ") WITH (\n"
                        + "  'connector' = 'values',\n"
                        + "  'sink.drop-late-event' = 'true'\n"
                        + ")";
        tEnv.executeSql(createSink);
        TableResult tableResult = tEnv.executeSql("INSERT INTO MySink SELECT * FROM kafka");
        final List<String> expected =
                Arrays.asList(
                        "+I[0, partition-0-name-0, 2020-03-08T13:12:11.123]",
                        "+I[0, partition-0-name-1, 2020-03-08T14:12:12.223]",
                        "+I[0, partition-0-name-2, 2020-03-08T15:12:13.323]",
                        "+I[1, partition-1-name-0, 2020-03-09T13:13:11.123]",
                        "+I[1, partition-1-name-1, 2020-03-09T15:13:11.133]",
                        "+I[1, partition-1-name-2, 2020-03-09T16:13:11.143]",
                        "+I[2, partition-2-name-0, 2020-03-10T13:12:14.123]",
                        "+I[2, partition-2-name-1, 2020-03-10T14:12:14.123]",
                        "+I[2, partition-2-name-2, 2020-03-10T14:13:14.123]",
                        "+I[2, partition-2-name-3, 2020-03-10T14:14:14.123]",
                        "+I[2, partition-2-name-4, 2020-03-10T14:15:14.123]",
                        "+I[2, partition-2-name-5, 2020-03-10T14:16:14.123]",
                        "+I[3, partition-3-name-0, 2020-03-11T17:12:11.123]",
                        "+I[3, partition-3-name-1, 2020-03-11T18:12:11.123]");
        KafkaTableTestUtils.waitingExpectedResults("MySink", expected, Duration.ofSeconds(5));

        // ------------- cleanup -------------------

        tableResult.getJobClient().ifPresent(JobClient::cancel);
        deleteTestTopic(topic);
    }

    // --------------------------------------------------------------------------------------------
    // Utilities
    // --------------------------------------------------------------------------------------------

    /** Extract the partition id from the row and set it on the record. */
    public static class TestPartitioner extends FlinkKafkaPartitioner<RowData> {

        private static final long serialVersionUID = 1L;
        private static final int PARTITION_ID_FIELD_IN_SCHEMA = 0;

        @Override
        public int partition(
                RowData record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            return partitions[record.getInt(PARTITION_ID_FIELD_IN_SCHEMA) % partitions.length];
        }
    }

    private String formatOptions() {
        if (!isLegacyConnector) {
            return String.format("'format' = '%s'", format);
        } else {
            String formatType = String.format("'format.type' = '%s'", format);
            if (format.equals(AVRO_FORMAT)) {
                // legacy connector requires to specify avro-schema
                String avroSchema =
                        "{\"type\":\"record\",\"name\":\"row_0\",\"fields\":"
                                + "[{\"name\":\"price\",\"type\":{\"type\":\"bytes\",\"logicalType\":\"decimal\","
                                + "\"precision\":38,\"scale\":18}},{\"name\":\"currency\",\"type\":[\"string\","
                                + "\"null\"]},{\"name\":\"log_date\",\"type\":{\"type\":\"int\",\"logicalType\":"
                                + "\"date\"}},{\"name\":\"log_time\",\"type\":{\"type\":\"int\",\"logicalType\":"
                                + "\"time-millis\"}},{\"name\":\"log_ts\",\"type\":{\"type\":\"long\","
                                + "\"logicalType\":\"timestamp-millis\"}}]}";
                return formatType + String.format(", 'format.avro-schema' = '%s'", avroSchema);
            } else {
                return formatType;
            }
        }
    }

    private static final class TestingSinkFunction implements SinkFunction<RowData> {

        private static final long serialVersionUID = 455430015321124493L;
        private static List<String> rows = new ArrayList<>();

        private final int expectedSize;

        private TestingSinkFunction(int expectedSize) {
            this.expectedSize = expectedSize;
            rows.clear();
        }

        @Override
        public void invoke(RowData value, Context context) {
            rows.add(value.toString());
            if (rows.size() >= expectedSize) {
                // job finish
                throw new SuccessException();
            }
        }
    }

    private static boolean isCausedByJobFinished(Throwable e) {
        if (e instanceof SuccessException) {
            return true;
        } else if (e.getCause() != null) {
            return isCausedByJobFinished(e.getCause());
        } else {
            return false;
        }
    }
}