text:
package org.apache.beam.runners.flink.streaming;

import static org.apache.beam.runners.flink.streaming.StreamRecordStripper.stripStreamRecordFromWindowedValue;
import static org.hamcrest.Matchers.emptyIterable;
import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;

import com.google.common.base.Function;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import java.util.Collections;
import java.util.HashMap;
import java.util.Optional;
import javax.annotation.Nullable;
import org.apache.beam.runners.core.StatefulDoFnRunner;
import org.apache.beam.runners.flink.FlinkPipelineOptions;
import org.apache.beam.runners.flink.translation.types.CoderTypeInformation;
import org.apache.beam.runners.flink.translation.types.CoderTypeSerializer;
import org.apache.beam.runners.flink.translation.wrappers.streaming.DoFnOperator;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.state.StateSpec;
import org.apache.beam.sdk.state.StateSpecs;
import org.apache.beam.sdk.state.TimeDomain;
import org.apache.beam.sdk.state.Timer;
import org.apache.beam.sdk.state.TimerSpec;
import org.apache.beam.sdk.state.TimerSpecs;
import org.apache.beam.sdk.state.ValueState;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.View;
import org.apache.beam.sdk.transforms.join.RawUnionValue;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.FixedWindows;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.util.WindowedValue.FullWindowedValueCoder;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionView;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.WindowingStrategy;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.OperatorStateHandles;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.TwoInputStreamOperatorTestHarness;
import org.apache.flink.util.OutputTag;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/**
 * Tests for {@link DoFnOperator}.
 */
@RunWith(JUnit4.class)
public class DoFnOperatorTest {

  // views and windows for testing side inputs
  private static final long WINDOW_MSECS_1 = 100;
  private static final long WINDOW_MSECS_2 = 500;
  private PCollectionView<Iterable<String>> view1;
  private PCollectionView<Iterable<String>> view2;

  @Before
  public void setUp() {
    PCollection<String> pc = Pipeline.create().apply(Create.of("1"));
    view1 =
        pc.apply(Window.into(FixedWindows.of(new Duration(WINDOW_MSECS_1))))
            .apply(View.asIterable());
    view2 =
        pc.apply(Window.into(FixedWindows.of(new Duration(WINDOW_MSECS_2))))
            .apply(View.asIterable());
  }

  @Test
  @SuppressWarnings("unchecked")
  public void testSingleOutput() throws Exception {

    Coder<WindowedValue<String>> coder = WindowedValue.getValueOnlyCoder(StringUtf8Coder.of());

    TupleTag<String> outputTag = new TupleTag<>("main-output");

    DoFnOperator<String, String> doFnOperator =
        new DoFnOperator<>(
            new IdentityDoFn<>(),
            "stepName",
            coder,
            outputTag,
            Collections.emptyList(),
            new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
            WindowingStrategy.globalDefault(),
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            PipelineOptionsFactory.as(FlinkPipelineOptions.class),
            null);

    OneInputStreamOperatorTestHarness<WindowedValue<String>, WindowedValue<String>> testHarness =
        new OneInputStreamOperatorTestHarness<>(doFnOperator);

    testHarness.open();

    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("Hello")));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(WindowedValue.valueInGlobalWindow("Hello")));

    testHarness.close();
  }

  @Test
  @SuppressWarnings("unchecked")
  public void testMultiOutputOutput() throws Exception {

    WindowedValue.ValueOnlyWindowedValueCoder<String> coder =
        WindowedValue.getValueOnlyCoder(StringUtf8Coder.of());

    TupleTag<String> mainOutput = new TupleTag<>("main-output");
    TupleTag<String> additionalOutput1 = new TupleTag<>("output-1");
    TupleTag<String> additionalOutput2 = new TupleTag<>("output-2");
    ImmutableMap<TupleTag<?>, OutputTag<?>> tagsToOutputTags =
        ImmutableMap.<TupleTag<?>, OutputTag<?>>builder()
            .put(additionalOutput1, new OutputTag<String>(additionalOutput1.getId()){})
            .put(additionalOutput2, new OutputTag<String>(additionalOutput2.getId()){})
            .build();
    ImmutableMap<TupleTag<?>, Coder<WindowedValue<?>>> tagsToCoders =
        ImmutableMap.<TupleTag<?>, Coder<WindowedValue<?>>>builder()
            .put(mainOutput, (Coder) coder)
            .put(additionalOutput1, coder)
            .put(additionalOutput2, coder)
            .build();
    ImmutableMap<TupleTag<?>, Integer> tagsToIds =
        ImmutableMap.<TupleTag<?>, Integer>builder()
            .put(mainOutput, 0)
            .put(additionalOutput1, 1)
            .put(additionalOutput2, 2)
            .build();

    DoFnOperator<String, String> doFnOperator =
        new DoFnOperator<>(
            new MultiOutputDoFn(additionalOutput1, additionalOutput2),
            "stepName",
            coder,
            mainOutput,
            ImmutableList.of(additionalOutput1, additionalOutput2),
            new DoFnOperator.MultiOutputOutputManagerFactory(
                mainOutput, tagsToOutputTags, tagsToCoders, tagsToIds),
            WindowingStrategy.globalDefault(),
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            PipelineOptionsFactory.as(FlinkPipelineOptions.class),
            null);

    OneInputStreamOperatorTestHarness<WindowedValue<String>, WindowedValue<String>> testHarness =
        new OneInputStreamOperatorTestHarness<>(doFnOperator);

    testHarness.open();

    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("one")));
    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("two")));
    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("hello")));

    assertThat(
        this.stripStreamRecord(testHarness.getOutput()),
        contains(
            WindowedValue.valueInGlobalWindow("got: hello")));

    assertThat(
        this.stripStreamRecord(testHarness.getSideOutput(tagsToOutputTags.get(additionalOutput1))),
        contains(
            WindowedValue.valueInGlobalWindow("extra: one"),
            WindowedValue.valueInGlobalWindow("got: hello")));

    assertThat(
        this.stripStreamRecord(testHarness.getSideOutput(tagsToOutputTags.get(additionalOutput2))),
        contains(
            WindowedValue.valueInGlobalWindow("extra: two"),
            WindowedValue.valueInGlobalWindow("got: hello")));

    testHarness.close();
  }

  /**
   * This test specifically verifies that we correctly map Flink watermarks to Beam watermarks. In
   * Beam, a watermark {@code T} guarantees there will not be elements with a timestamp
   * {@code < T} in the future. In Flink, a watermark {@code T} guarantees there will not be
   * elements with a timestamp {@code <= T} in the future. We have to make sure to take this into
   * account when firing timers.
   *
   * <p>This not test the timer API in general or processing-time timers because there are generic
   * tests for this in {@code ParDoTest}.
   */
  @Test
  public void testWatermarkContract() throws Exception {

    final Instant timerTimestamp = new Instant(1000);
    final String outputMessage = "Timer fired";

    WindowingStrategy<Object, IntervalWindow> windowingStrategy =
        WindowingStrategy.of(FixedWindows.of(new Duration(10_000)));

    DoFn<Integer, String> fn = new DoFn<Integer, String>() {
      private static final String EVENT_TIMER_ID = "eventTimer";

      @TimerId(EVENT_TIMER_ID)
      private final TimerSpec eventTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);

      @ProcessElement
      public void processElement(ProcessContext context, @TimerId(EVENT_TIMER_ID) Timer timer) {
        timer.set(timerTimestamp);
      }

      @OnTimer(EVENT_TIMER_ID)
      public void onEventTime(OnTimerContext context) {
        assertEquals(
            "Timer timestamp must match set timestamp.", timerTimestamp, context.timestamp());
        context.outputWithTimestamp(outputMessage, context.timestamp());
      }
    };

    WindowedValue.FullWindowedValueCoder<Integer> inputCoder =
        WindowedValue.getFullCoder(
            VarIntCoder.of(),
            windowingStrategy.getWindowFn().windowCoder());

    WindowedValue.FullWindowedValueCoder<String> outputCoder =
        WindowedValue.getFullCoder(
            StringUtf8Coder.of(),
            windowingStrategy.getWindowFn().windowCoder());

    TupleTag<String> outputTag = new TupleTag<>("main-output");

    DoFnOperator<Integer, String> doFnOperator =
        new DoFnOperator<>(
            fn,
            "stepName",
            inputCoder,
            outputTag,
            Collections.emptyList(),
            new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, outputCoder),
            windowingStrategy,
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            PipelineOptionsFactory.as(FlinkPipelineOptions.class),
            VarIntCoder.of() /* key coder */);

    OneInputStreamOperatorTestHarness<WindowedValue<Integer>, WindowedValue<String>> testHarness =
        new KeyedOneInputStreamOperatorTestHarness<>(
            doFnOperator, WindowedValue::getValue, new CoderTypeInformation<>(VarIntCoder.of()));

    testHarness.setup(new CoderTypeSerializer<>(outputCoder));

    testHarness.open();

    testHarness.processWatermark(0);

    IntervalWindow window1 = new IntervalWindow(new Instant(0), Duration.millis(10_000));

    // this should register a timer
    testHarness.processElement(
        new StreamRecord<>(WindowedValue.of(13, new Instant(0), window1, PaneInfo.NO_FIRING)));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        emptyIterable());

    // this does not yet fire the timer (in vanilla Flink it would)
    testHarness.processWatermark(timerTimestamp.getMillis());

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        emptyIterable());

    testHarness.getOutput().clear();

    // this must fire the timer
    testHarness.processWatermark(timerTimestamp.getMillis() + 1);

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(
            WindowedValue.of(
                outputMessage, new Instant(timerTimestamp), window1, PaneInfo.NO_FIRING)));

    testHarness.close();
  }

  @Test
  public void testLateDroppingForStatefulFn() throws Exception {

    WindowingStrategy<Object, IntervalWindow> windowingStrategy =
        WindowingStrategy.of(FixedWindows.of(new Duration(10)));

    DoFn<Integer, String> fn = new DoFn<Integer, String>() {

      @StateId("state")
      private final StateSpec<ValueState<String>> stateSpec =
          StateSpecs.value(StringUtf8Coder.of());

      @ProcessElement
      public void processElement(ProcessContext context) {
        context.output(context.element().toString());
      }
    };

    Coder<WindowedValue<Integer>> inputCoder = WindowedValue.getFullCoder(
        VarIntCoder.of(), windowingStrategy.getWindowFn().windowCoder());
    Coder<WindowedValue<String>> outputCoder = WindowedValue.getFullCoder(
        StringUtf8Coder.of(), windowingStrategy.getWindowFn().windowCoder());

    TupleTag<String> outputTag = new TupleTag<>("main-output");

    DoFnOperator<Integer, String> doFnOperator =
        new DoFnOperator<>(
            fn,
            "stepName",
            inputCoder,
            outputTag,
            Collections.emptyList(),
            new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, outputCoder),
            windowingStrategy,
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            PipelineOptionsFactory.as(FlinkPipelineOptions.class),
            VarIntCoder.of() /* key coder */);

    OneInputStreamOperatorTestHarness<WindowedValue<Integer>, WindowedValue<String>> testHarness =
        new KeyedOneInputStreamOperatorTestHarness<>(
            doFnOperator, WindowedValue::getValue, new CoderTypeInformation<>(VarIntCoder.of()));

    testHarness.open();

    testHarness.processWatermark(0);

    IntervalWindow window1 = new IntervalWindow(new Instant(0), Duration.millis(10));

    // this should not be late
    testHarness.processElement(
        new StreamRecord<>(WindowedValue.of(13, new Instant(0), window1, PaneInfo.NO_FIRING)));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(WindowedValue.of("13", new Instant(0), window1, PaneInfo.NO_FIRING)));

    testHarness.getOutput().clear();

    testHarness.processWatermark(9);

    // this should still not be considered late
    testHarness.processElement(
        new StreamRecord<>(WindowedValue.of(17, new Instant(0), window1, PaneInfo.NO_FIRING)));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(WindowedValue.of("17", new Instant(0), window1, PaneInfo.NO_FIRING)));

    testHarness.getOutput().clear();

    testHarness.processWatermark(10);

    // this should now be considered late
    testHarness.processElement(
        new StreamRecord<>(WindowedValue.of(17, new Instant(0), window1, PaneInfo.NO_FIRING)));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        emptyIterable());

    testHarness.close();
  }

  @Test
  public void testStateGCForStatefulFn() throws Exception {

    WindowingStrategy<Object, IntervalWindow> windowingStrategy =
        WindowingStrategy.of(FixedWindows.of(new Duration(10))).withAllowedLateness(Duration.ZERO);

    final String timerId = "boo";
    final String stateId = "dazzle";

    final int offset = 5000;
    final int timerOutput = 4093;

    DoFn<KV<String, Integer>, KV<String, Integer>> fn =
        new DoFn<KV<String, Integer>, KV<String, Integer>>() {

          @TimerId(timerId)
          private final TimerSpec spec = TimerSpecs.timer(TimeDomain.EVENT_TIME);

          @StateId(stateId)
          private final StateSpec<ValueState<String>> stateSpec =
              StateSpecs.value(StringUtf8Coder.of());

          @ProcessElement
          public void processElement(
              ProcessContext context,
              @TimerId(timerId) Timer timer,
              @StateId(stateId) ValueState<String> state,
              BoundedWindow window) {
            timer.set(window.maxTimestamp());
            state.write(context.element().getKey());
            context.output(
                KV.of(context.element().getKey(), context.element().getValue() + offset));
          }

          @OnTimer(timerId)
          public void onTimer(OnTimerContext context, @StateId(stateId) ValueState<String> state) {
            context.output(KV.of(state.read(), timerOutput));
          }
        };

    WindowedValue.FullWindowedValueCoder<KV<String, Integer>> coder =
        WindowedValue.getFullCoder(
            KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of()),
            windowingStrategy.getWindowFn().windowCoder());

    TupleTag<KV<String, Integer>> outputTag = new TupleTag<>("main-output");

    DoFnOperator<KV<String, Integer>, KV<String, Integer>> doFnOperator =
        new DoFnOperator<>(
            fn,
            "stepName",
            coder,
            outputTag,
            Collections.emptyList(),
            new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
            windowingStrategy,
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            PipelineOptionsFactory.as(FlinkPipelineOptions.class),
            StringUtf8Coder.of() /* key coder */);

    KeyedOneInputStreamOperatorTestHarness<
            String, WindowedValue<KV<String, Integer>>, WindowedValue<KV<String, Integer>>>
        testHarness =
            new KeyedOneInputStreamOperatorTestHarness<>(
                doFnOperator,
                kvWindowedValue -> kvWindowedValue.getValue().getKey(),
                new CoderTypeInformation<>(StringUtf8Coder.of()));

    testHarness.open();

    testHarness.processWatermark(0);

    assertEquals(0, testHarness.numKeyedStateEntries());

    IntervalWindow window1 = new IntervalWindow(new Instant(0), Duration.millis(10));

    testHarness.processElement(
        new StreamRecord<>(
            WindowedValue.of(KV.of("key1", 5), new Instant(1), window1, PaneInfo.NO_FIRING)));

    testHarness.processElement(
        new StreamRecord<>(
            WindowedValue.of(KV.of("key2", 7), new Instant(3), window1, PaneInfo.NO_FIRING)));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(
            WindowedValue.of(
                KV.of("key1", 5 + offset), new Instant(1), window1, PaneInfo.NO_FIRING),
            WindowedValue.of(
                KV.of("key2", 7 + offset), new Instant(3), window1, PaneInfo.NO_FIRING)));

    assertEquals(2, testHarness.numKeyedStateEntries());

    testHarness.getOutput().clear();

    // this should trigger both the window.maxTimestamp() timer and the GC timer
    // this tests that the GC timer fires after the user timer
    // we have to add 1 here because Flink timers fire when watermark >= timestamp while Beam
    // timers fire when watermark > timestamp
    testHarness.processWatermark(
        window1.maxTimestamp()
            .plus(windowingStrategy.getAllowedLateness())
            .plus(StatefulDoFnRunner.TimeInternalsCleanupTimer.GC_DELAY_MS)
            .getMillis() + 1);

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(
            WindowedValue.of(
                KV.of("key1", timerOutput), new Instant(9), window1, PaneInfo.NO_FIRING),
            WindowedValue.of(
                KV.of("key2", timerOutput), new Instant(9), window1, PaneInfo.NO_FIRING)));

    // ensure the state was garbage collected
    assertEquals(0, testHarness.numKeyedStateEntries());

    testHarness.close();
  }

  @Test
  public void testNormalParDoSideInputs() throws Exception {
    testSideInputs(false);
  }

  @Test
  public void testKeyedParDoSideInputs() throws Exception {
    testSideInputs(true);
  }

  void testSideInputs(boolean keyed) throws Exception {

    Coder<WindowedValue<String>> coder = WindowedValue.getValueOnlyCoder(StringUtf8Coder.of());

    TupleTag<String> outputTag = new TupleTag<>("main-output");

    ImmutableMap<Integer, PCollectionView<?>> sideInputMapping =
        ImmutableMap.<Integer, PCollectionView<?>>builder()
            .put(1, view1)
            .put(2, view2)
            .build();

    Coder<String> keyCoder = null;
    if (keyed) {
      keyCoder = StringUtf8Coder.of();
    }

    DoFnOperator<String, String> doFnOperator =
        new DoFnOperator<>(
            new IdentityDoFn<>(),
            "stepName",
            coder,
            outputTag,
            Collections.emptyList(),
            new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
            WindowingStrategy.globalDefault(),
            sideInputMapping, /* side-input mapping */
            ImmutableList.of(view1, view2), /* side inputs */
            PipelineOptionsFactory.as(FlinkPipelineOptions.class),
            keyCoder);

    TwoInputStreamOperatorTestHarness<WindowedValue<String>, RawUnionValue, WindowedValue<String>>
        testHarness = new TwoInputStreamOperatorTestHarness<>(doFnOperator);

    if (keyed) {
      // we use a dummy key for the second input since it is considered to be broadcast
      testHarness = new KeyedTwoInputStreamOperatorTestHarness<>(
          doFnOperator,
          new StringKeySelector(),
          new DummyKeySelector(),
          BasicTypeInfo.STRING_TYPE_INFO);
    }

    testHarness.open();

    IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(100));
    IntervalWindow secondWindow = new IntervalWindow(new Instant(0), new Instant(500));

    // test the keep of sideInputs events
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                1,
                valuesInWindow(ImmutableList.of("hello", "ciao"), new Instant(0), firstWindow))));
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                2,
                valuesInWindow(ImmutableList.of("foo", "bar"), new Instant(0), secondWindow))));

    // push in a regular elements
    WindowedValue<String> helloElement = valueInWindow("Hello", new Instant(0), firstWindow);
    WindowedValue<String> worldElement = valueInWindow("World", new Instant(1000), firstWindow);
    testHarness.processElement1(new StreamRecord<>(helloElement));
    testHarness.processElement1(new StreamRecord<>(worldElement));

    // test the keep of pushed-back events
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                1,
                valuesInWindow(ImmutableList.of("hello", "ciao"),
                    new Instant(1000), firstWindow))));
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                2,
                valuesInWindow(ImmutableList.of("foo", "bar"), new Instant(1000), secondWindow))));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(helloElement, worldElement));

    testHarness.close();

  }

  @Test
  public void testStateRestore() throws Exception {
    DoFn<KV<String, Long>, KV<String, Long>> filterElementsEqualToCountFn =
        new DoFn<KV<String, Long>, KV<String, Long>>() {

          @StateId("counter")
          private final StateSpec<ValueState<Long>> counterSpec = StateSpecs
              .value(VarLongCoder.of());

          @ProcessElement
          public void processElement(ProcessContext context,
              @StateId("counter") ValueState<Long> count) {
            long currentCount = Optional.ofNullable(count.read()).orElse(0L);
            currentCount = currentCount + 1;
            count.write(currentCount);

            KV<String, Long> currentElement = context.element();
            if (currentCount == currentElement.getValue()) {
              context.output(currentElement);
            }
          }
        };

    WindowingStrategy<Object, GlobalWindow> windowingStrategy = WindowingStrategy.globalDefault();

    TupleTag<KV<String, Long>> outputTag = new TupleTag<>("main-output");

    FullWindowedValueCoder<KV<String, Long>> kvCoder = WindowedValue.getFullCoder(
        KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()),
        windowingStrategy.getWindowFn().windowCoder()
    );

    CoderTypeInformation<String> keyCoderInfo = new CoderTypeInformation<>(StringUtf8Coder.of());
    KeySelector<WindowedValue<KV<String, Long>>, String> keySelector = e -> e.getValue().getKey();

    OneInputStreamOperatorTestHarness<WindowedValue<KV<String, Long>>,
        WindowedValue<KV<String, Long>>> testHarness = createTestHarness(windowingStrategy,
        filterElementsEqualToCountFn, kvCoder, kvCoder, outputTag, keyCoderInfo, keySelector);
    testHarness.open();

    testHarness
        .processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow(KV.of("a", 100L))));
    testHarness
        .processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow(KV.of("a", 100L))));

    final OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
    testHarness.close();

    testHarness = createTestHarness(windowingStrategy, filterElementsEqualToCountFn, kvCoder,
        kvCoder, outputTag, keyCoderInfo, keySelector);
    testHarness.initializeState(snapshot);
    testHarness.open();

    // after restore: counter = 2
    testHarness
        .processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow(KV.of("a", 100L))));
    testHarness
        .processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow(KV.of("a", 4L))));
    testHarness
        .processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow(KV.of("a", 5L))));
    testHarness
        .processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow(KV.of("a", 100L))));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(
            WindowedValue.valueInGlobalWindow(KV.of("a", 4L)),
            WindowedValue.valueInGlobalWindow(KV.of("a", 5L))
        )
    );

    testHarness.close();
  }

  @Test
  public void nonKeyedParDoSideInputCheckpointing() throws Exception {
    sideInputCheckpointing(() -> {
      Coder<WindowedValue<String>> coder =
          WindowedValue.getFullCoder(StringUtf8Coder.of(), IntervalWindow.getCoder());
      TupleTag<String> outputTag = new TupleTag<>("main-output");

      ImmutableMap<Integer, PCollectionView<?>> sideInputMapping =
          ImmutableMap.<Integer, PCollectionView<?>>builder()
              .put(1, view1)
              .put(2, view2)
              .build();

      DoFnOperator<String, String> doFnOperator = new DoFnOperator<>(
          new IdentityDoFn<>(),
          "stepName",
          coder,
          outputTag,
          Collections.emptyList(),
          new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
          WindowingStrategy.globalDefault(),
          sideInputMapping, /* side-input mapping */
          ImmutableList.of(view1, view2), /* side inputs */
          PipelineOptionsFactory.as(FlinkPipelineOptions.class),
          null);

      return new TwoInputStreamOperatorTestHarness<>(doFnOperator);
    });
  }

  @Test
  public void keyedParDoSideInputCheckpointing() throws Exception {
    sideInputCheckpointing(() -> {
      Coder<WindowedValue<String>> coder =
          WindowedValue.getFullCoder(StringUtf8Coder.of(), IntervalWindow.getCoder());
      TupleTag<String> outputTag = new TupleTag<>("main-output");

      StringUtf8Coder keyCoder = StringUtf8Coder.of();

      ImmutableMap<Integer, PCollectionView<?>> sideInputMapping =
          ImmutableMap.<Integer, PCollectionView<?>>builder()
              .put(1, view1)
              .put(2, view2)
              .build();

      DoFnOperator<String, String> doFnOperator = new DoFnOperator<>(
          new IdentityDoFn<>(),
          "stepName",
          coder,
          outputTag,
          Collections.emptyList(),
          new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
          WindowingStrategy.globalDefault(),
          sideInputMapping, /* side-input mapping */
          ImmutableList.of(view1, view2), /* side inputs */
          PipelineOptionsFactory.as(FlinkPipelineOptions.class),
          keyCoder);

      return new KeyedTwoInputStreamOperatorTestHarness<>(
          doFnOperator,
          new StringKeySelector(),
          // we use a dummy key for the second input since it is considered to be broadcast
          new DummyKeySelector(),
          BasicTypeInfo.STRING_TYPE_INFO);
    });
  }

  void sideInputCheckpointing(
      TestHarnessFactory<
          TwoInputStreamOperatorTestHarness<
              WindowedValue<String>,
              RawUnionValue,
              WindowedValue<String>>> harnessFactory) throws Exception {

    TwoInputStreamOperatorTestHarness<WindowedValue<String>, RawUnionValue, WindowedValue<String>>
        testHarness = harnessFactory.create();

    testHarness.open();

    IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(100));
    IntervalWindow secondWindow = new IntervalWindow(new Instant(0), new Instant(500));

    // push in some side inputs for both windows
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                1,
                valuesInWindow(
                    ImmutableList.of(KV.of((Void) null, "hello"), KV.of((Void) null, "ciao")),
                    new Instant(0),
                    firstWindow))));
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                2,
                valuesInWindow(
                    ImmutableList.of(KV.of((Void) null, "foo"), KV.of((Void) null, "bar")),
                    new Instant(0),
                    secondWindow))));

    // snapshot state, throw away the operator, then restore and verify that we still match
    // main-input elements to the side-inputs that we sent earlier
    OperatorStateHandles snapshot = testHarness.snapshot(0, 0);

    testHarness = harnessFactory.create();

    testHarness.initializeState(snapshot);
    testHarness.open();

    // push in main-input elements
    WindowedValue<String> helloElement = valueInWindow("Hello", new Instant(0), firstWindow);
    WindowedValue<String> worldElement = valueInWindow("World", new Instant(1000), firstWindow);
    testHarness.processElement1(new StreamRecord<>(helloElement));
    testHarness.processElement1(new StreamRecord<>(worldElement));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(helloElement, worldElement));

    testHarness.close();
  }

  @Test
  public void nonKeyedParDoPushbackDataCheckpointing() throws Exception {
    pushbackDataCheckpointing(() -> {
      Coder<WindowedValue<String>> coder =
          WindowedValue.getFullCoder(StringUtf8Coder.of(), IntervalWindow.getCoder());

      TupleTag<String> outputTag = new TupleTag<>("main-output");

      ImmutableMap<Integer, PCollectionView<?>> sideInputMapping =
          ImmutableMap.<Integer, PCollectionView<?>>builder()
              .put(1, view1)
              .put(2, view2)
              .build();

      DoFnOperator<String, String> doFnOperator = new DoFnOperator<>(
          new IdentityDoFn<>(),
          "stepName",
          coder,
          outputTag,
          Collections.emptyList(),
          new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
          WindowingStrategy.globalDefault(),
          sideInputMapping, /* side-input mapping */
          ImmutableList.of(view1, view2), /* side inputs */
          PipelineOptionsFactory.as(FlinkPipelineOptions.class),
          null);

      return new TwoInputStreamOperatorTestHarness<>(doFnOperator);
    });
  }

  @Test
  public void keyedParDoPushbackDataCheckpointing() throws Exception {
    pushbackDataCheckpointing(() -> {
      Coder<WindowedValue<String>> coder =
          WindowedValue.getFullCoder(StringUtf8Coder.of(), IntervalWindow.getCoder());

      TupleTag<String> outputTag = new TupleTag<>("main-output");

      StringUtf8Coder keyCoder = StringUtf8Coder.of();

      ImmutableMap<Integer, PCollectionView<?>> sideInputMapping =
          ImmutableMap.<Integer, PCollectionView<?>>builder()
              .put(1, view1)
              .put(2, view2)
              .build();

      DoFnOperator<String, String> doFnOperator = new DoFnOperator<>(
          new IdentityDoFn<>(),
          "stepName",
          coder,
          outputTag,
          Collections.emptyList(),
          new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, coder),
          WindowingStrategy.globalDefault(),
          sideInputMapping, /* side-input mapping */
          ImmutableList.of(view1, view2), /* side inputs */
          PipelineOptionsFactory.as(FlinkPipelineOptions.class),
          keyCoder);

      return new KeyedTwoInputStreamOperatorTestHarness<>(
          doFnOperator,
          new StringKeySelector(),
          // we use a dummy key for the second input since it is considered to be broadcast
          new DummyKeySelector(),
          BasicTypeInfo.STRING_TYPE_INFO);
    });
  }

  void pushbackDataCheckpointing(
      TestHarnessFactory<
          TwoInputStreamOperatorTestHarness<
              WindowedValue<String>,
              RawUnionValue,
              WindowedValue<String>>> harnessFactory) throws Exception {

    TwoInputStreamOperatorTestHarness<WindowedValue<String>, RawUnionValue, WindowedValue<String>>
        testHarness = harnessFactory.create();

    testHarness.open();

    IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(100));
    IntervalWindow secondWindow = new IntervalWindow(new Instant(0), new Instant(500));

    // push in main-input elements
    WindowedValue<String> helloElement = valueInWindow("Hello", new Instant(0), firstWindow);
    WindowedValue<String> worldElement = valueInWindow("World", new Instant(1000), firstWindow);
    testHarness.processElement1(new StreamRecord<>(helloElement));
    testHarness.processElement1(new StreamRecord<>(worldElement));

    // snapshot state, throw away the operator, then restore and verify that we still match
    // main-input elements to the side-inputs that we sent earlier
    OperatorStateHandles snapshot = testHarness.snapshot(0, 0);

    testHarness = harnessFactory.create();

    testHarness.initializeState(snapshot);
    testHarness.open();

    // push in some side inputs for both windows
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                1,
                valuesInWindow(
                    ImmutableList.of(KV.of((Void) null, "hello"), KV.of((Void) null, "ciao")),
                    new Instant(0),
                    firstWindow))));
    testHarness.processElement2(
        new StreamRecord<>(
            new RawUnionValue(
                2,
                valuesInWindow(
                    ImmutableList.of(KV.of((Void) null, "foo"), KV.of((Void) null, "bar")),
                    new Instant(0),
                    secondWindow))));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(helloElement, worldElement));

    testHarness.close();
  }

  @Test
  public void testTimersRestore() throws Exception {
    final Instant timerTimestamp = new Instant(1000);
    final String outputMessage = "Timer fired";

    WindowingStrategy<Object, IntervalWindow> windowingStrategy =
        WindowingStrategy.of(FixedWindows.of(new Duration(10_000)));

    DoFn<Integer, String> fn = new DoFn<Integer, String>() {
      private static final String EVENT_TIMER_ID = "eventTimer";

      @TimerId(EVENT_TIMER_ID)
      private final TimerSpec eventTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);

      @ProcessElement
      public void processElement(ProcessContext context, @TimerId(EVENT_TIMER_ID) Timer timer) {
        timer.set(timerTimestamp);
      }

      @OnTimer(EVENT_TIMER_ID)
      public void onEventTime(OnTimerContext context) {
        assertEquals(
            "Timer timestamp must match set timestamp.", timerTimestamp, context.timestamp());
        context.outputWithTimestamp(outputMessage, context.timestamp());
      }
    };

    WindowedValue.FullWindowedValueCoder<Integer> inputCoder =
        WindowedValue.getFullCoder(
            VarIntCoder.of(),
            windowingStrategy.getWindowFn().windowCoder());

    WindowedValue.FullWindowedValueCoder<String> outputCoder =
        WindowedValue.getFullCoder(
            StringUtf8Coder.of(),
            windowingStrategy.getWindowFn().windowCoder());

    TupleTag<String> outputTag = new TupleTag<>("main-output");

    final CoderTypeSerializer<WindowedValue<String>> outputSerializer = new CoderTypeSerializer<>(
        outputCoder);
    CoderTypeInformation<Integer> keyCoderInfo = new CoderTypeInformation<>(VarIntCoder.of());
    KeySelector<WindowedValue<Integer>, Integer> keySelector = WindowedValue::getValue;

    OneInputStreamOperatorTestHarness<WindowedValue<Integer>, WindowedValue<String>> testHarness =
        createTestHarness(windowingStrategy, fn, inputCoder, outputCoder, outputTag, keyCoderInfo,
            keySelector);

    testHarness.setup(outputSerializer);

    testHarness.open();

    testHarness.processWatermark(0);

    IntervalWindow window1 = new IntervalWindow(new Instant(0), Duration.millis(10_000));

    // this should register a timer
    testHarness.processElement(
        new StreamRecord<>(WindowedValue.of(13, new Instant(0), window1, PaneInfo.NO_FIRING)));

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        emptyIterable());

    // snapshot and restore
    final OperatorStateHandles snapshot = testHarness.snapshot(0, 0);
    testHarness.close();

    testHarness = createTestHarness(windowingStrategy, fn, inputCoder, outputCoder, outputTag,
        keyCoderInfo, keySelector);
    testHarness.setup(outputSerializer);
    testHarness.initializeState(snapshot);
    testHarness.open();

    // this must fire the timer
    testHarness.processWatermark(timerTimestamp.getMillis() + 1);

    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(
            WindowedValue.of(
                outputMessage, new Instant(timerTimestamp), window1, PaneInfo.NO_FIRING)));

    testHarness.close();
  }

  private <K, InT, OutT> OneInputStreamOperatorTestHarness<WindowedValue<InT>, WindowedValue<OutT>>
  createTestHarness(WindowingStrategy<Object, ?> windowingStrategy, DoFn<InT, OutT> fn,
      FullWindowedValueCoder<InT> inputCoder, FullWindowedValueCoder<OutT> outputCoder,
      TupleTag<OutT> outputTag, TypeInformation<K> keyCoderInfo,
      KeySelector<WindowedValue<InT>, K> keySelector) throws Exception {
    DoFnOperator<InT, OutT> doFnOperator = new DoFnOperator<>(
        fn,
        "stepName",
        inputCoder,
        outputTag,
        Collections.emptyList(),
        new DoFnOperator.MultiOutputOutputManagerFactory<>(outputTag, outputCoder),
        windowingStrategy,
        new HashMap<>(), /* side-input mapping */
        Collections.emptyList(), /* side inputs */
        PipelineOptionsFactory.as(FlinkPipelineOptions.class),
        VarIntCoder.of() /* key coder */);

    return new KeyedOneInputStreamOperatorTestHarness<>(doFnOperator, keySelector, keyCoderInfo);
  }

  @Test
  @SuppressWarnings("unchecked")
  public void testBundle() throws Exception {

    WindowedValue.ValueOnlyWindowedValueCoder<String> windowedValueCoder =
        WindowedValue.getValueOnlyCoder(StringUtf8Coder.of());

    TupleTag<String> outputTag = new TupleTag<>("main-output");
    FlinkPipelineOptions options = PipelineOptionsFactory.as(FlinkPipelineOptions.class);
    options.setMaxBundleSize(2L);
    options.setMaxBundleTimeMills(10L);

    IdentityDoFn<String> doFn = new IdentityDoFn<String>() {
      @FinishBundle
      public void finishBundle(FinishBundleContext context) {
        context.output(
            "finishBundle", BoundedWindow.TIMESTAMP_MIN_VALUE, GlobalWindow.INSTANCE);
      }
    };

    DoFnOperator.MultiOutputOutputManagerFactory<String> outputManagerFactory =
        new DoFnOperator.MultiOutputOutputManagerFactory(
            outputTag,
            WindowedValue.getFullCoder(StringUtf8Coder.of(), GlobalWindow.Coder.INSTANCE));

    DoFnOperator<String, String> doFnOperator =
        new DoFnOperator<>(
            doFn,
            "stepName",
            windowedValueCoder,
            outputTag,
            Collections.emptyList(),
            outputManagerFactory,
            WindowingStrategy.globalDefault(),
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            options,
            null);

    OneInputStreamOperatorTestHarness<WindowedValue<String>, WindowedValue<String>> testHarness =
        new OneInputStreamOperatorTestHarness<>(doFnOperator);

    testHarness.open();

    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("a")));
    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("b")));
    testHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("c")));

    // draw a snapshot
    OperatorStateHandles snapshot = testHarness.snapshot(0, 0);

    // There is a finishBundle in snapshot()
    // Elements will be buffered as part of finishing a bundle in snapshot()
    assertThat(
        stripStreamRecordFromWindowedValue(testHarness.getOutput()),
        contains(
            WindowedValue.valueInGlobalWindow("a"),
            WindowedValue.valueInGlobalWindow("b"),
            WindowedValue.valueInGlobalWindow("finishBundle"),
            WindowedValue.valueInGlobalWindow("c")));

    testHarness.close();

    DoFnOperator<String, String> newDoFnOperator =
        new DoFnOperator<>(
            doFn,
            "stepName",
            windowedValueCoder,
            outputTag,
            Collections.emptyList(),
            outputManagerFactory,
            WindowingStrategy.globalDefault(),
            new HashMap<>(), /* side-input mapping */
            Collections.emptyList(), /* side inputs */
            options,
            null);

    OneInputStreamOperatorTestHarness<WindowedValue<String>, WindowedValue<String>> newHarness =
        new OneInputStreamOperatorTestHarness<>(newDoFnOperator);

    // restore snapshot
    newHarness.initializeState(snapshot);

    newHarness.open();

    // startBundle will output the buffered elements.
    newHarness.processElement(new StreamRecord<>(WindowedValue.valueInGlobalWindow("d")));

    // check finishBundle by timeout
    newHarness.setProcessingTime(10);

    assertThat(
        stripStreamRecordFromWindowedValue(newHarness.getOutput()),
        contains(
            WindowedValue.valueInGlobalWindow("finishBundle"),
            WindowedValue.valueInGlobalWindow("d"),
            WindowedValue.valueInGlobalWindow("finishBundle")));

    newHarness.close();
  }

  private Iterable<WindowedValue<String>> stripStreamRecord(Iterable<?> input) {
    return FluentIterable.from(input)
        .filter(o -> o instanceof StreamRecord)
        .transform(
            new Function<Object, WindowedValue<String>>() {
              @Nullable
              @Override
              @SuppressWarnings({"unchecked", "rawtypes"})
              public WindowedValue<String> apply(@Nullable Object o) {
                if (o instanceof StreamRecord) {
                  return (WindowedValue<String>) ((StreamRecord) o).getValue();
                }
                throw new RuntimeException("unreachable");
              }
            });
  }

  private static class MultiOutputDoFn extends DoFn<String, String> {
    private TupleTag<String> additionalOutput1;
    private TupleTag<String> additionalOutput2;

    public MultiOutputDoFn(TupleTag<String> additionalOutput1, TupleTag<String> additionalOutput2) {
      this.additionalOutput1 = additionalOutput1;
      this.additionalOutput2 = additionalOutput2;
    }

    @ProcessElement
    public void processElement(ProcessContext c) throws Exception {
      if ("one".equals(c.element())) {
        c.output(additionalOutput1, "extra: one");
      } else if ("two".equals(c.element())) {
        c.output(additionalOutput2, "extra: two");
      } else {
        c.output("got: " + c.element());
        c.output(additionalOutput1, "got: " + c.element());
        c.output(additionalOutput2, "got: " + c.element());
      }
    }
  }

  private static class IdentityDoFn<T> extends DoFn<T, T> {
    @ProcessElement
    public void processElement(ProcessContext c) throws Exception {
      c.output(c.element());
    }
  }

  @SuppressWarnings({"unchecked", "rawtypes"})
  private WindowedValue<Iterable<?>> valuesInWindow(
      Iterable<?> values, Instant timestamp, BoundedWindow window) {
    return (WindowedValue) WindowedValue.of(values, timestamp, window, PaneInfo.NO_FIRING);
  }

  @SuppressWarnings({"unchecked", "rawtypes"})
  private <T> WindowedValue<T> valueInWindow(
      T value, Instant timestamp, BoundedWindow window) {
    return WindowedValue.of(value, timestamp, window, PaneInfo.NO_FIRING);
  }

  private static class DummyKeySelector implements KeySelector<RawUnionValue, String> {
    @Override
    public String getKey(RawUnionValue stringWindowedValue) throws Exception {
      return "dummy_key";
    }
  }

  private static class StringKeySelector implements KeySelector<WindowedValue<String>, String> {
    @Override
    public String getKey(WindowedValue<String> stringWindowedValue) throws Exception {
      return stringWindowedValue.getValue();
    }
  }

  private interface TestHarnessFactory<T> {
    T create() throws Exception;
  }
}


text:
package org.apache.lucene.codecs.perfield;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.ServiceLoader; // javadocs
import java.util.TreeMap;

import org.apache.lucene.codecs.FieldsConsumer;
import org.apache.lucene.codecs.FieldsProducer;
import org.apache.lucene.codecs.PostingsFormat;
import org.apache.lucene.codecs.TermsConsumer;
import org.apache.lucene.index.FieldInfo;
import org.apache.lucene.index.SegmentReadState;
import org.apache.lucene.index.SegmentWriteState;
import org.apache.lucene.index.Terms;
import org.apache.lucene.util.IOUtils;

/**
 * Enables per field postings support.
 *

 * Note, when extending this class, the name ({@link #getName}) is
 * written into the index. In order for the field to be read, the
 * name must resolve to your implementation via {@link #forName(String)}.
 * This method uses Java's
 * {@link ServiceLoader Service Provider Interface} to resolve format names.
 *

 * Files written by each posting format have an additional suffix containing the
 * format name. For example, in a per-field configuration instead of <tt>_1.prx</tt>
 * filenames would look like <tt>_1_Lucene40_0.prx</tt>.
 * @see ServiceLoader
 * @lucene.experimental
 */

public abstract class PerFieldPostingsFormat extends PostingsFormat {
  /** Name of this {@link PostingsFormat}. */
  public static final String PER_FIELD_NAME = "PerField40";

  /** {@link FieldInfo} attribute name used to store the
   *  format name for each field. */
  public static final String PER_FIELD_FORMAT_KEY = PerFieldPostingsFormat.class.getSimpleName() + ".format";

  /** {@link FieldInfo} attribute name used to store the
   *  segment suffix name for each field. */
  public static final String PER_FIELD_SUFFIX_KEY = PerFieldPostingsFormat.class.getSimpleName() + ".suffix";

  /** Sole constructor. */
  public PerFieldPostingsFormat() {
    super(PER_FIELD_NAME);
  }

  @Override
  public final FieldsConsumer fieldsConsumer(SegmentWriteState state)
      throws IOException {
    return new FieldsWriter(state);
  }

  static class FieldsConsumerAndSuffix implements Closeable {
    FieldsConsumer consumer;
    int suffix;

    @Override
    public void close() throws IOException {
      consumer.close();
    }
  }

  private class FieldsWriter extends FieldsConsumer {

    private final Map<PostingsFormat,FieldsConsumerAndSuffix> formats = new HashMap<PostingsFormat,FieldsConsumerAndSuffix>();
    private final Map<String,Integer> suffixes = new HashMap<String,Integer>();

    private final SegmentWriteState segmentWriteState;

    public FieldsWriter(SegmentWriteState state) {
      segmentWriteState = state;
    }

    @Override
    public TermsConsumer addField(FieldInfo field) throws IOException {
      final PostingsFormat format = getPostingsFormatForField(field.name);
      if (format == null) {
        throw new IllegalStateException("invalid null PostingsFormat for field=\"" + field.name + "\"");
      }
      final String formatName = format.getName();

      String previousValue = field.putAttribute(PER_FIELD_FORMAT_KEY, formatName);
      assert previousValue == null;

      Integer suffix;

      FieldsConsumerAndSuffix consumer = formats.get(format);
      if (consumer == null) {
        // First time we are seeing this format; create a new instance

        // bump the suffix
        suffix = suffixes.get(formatName);
        if (suffix == null) {
          suffix = 0;
        } else {
          suffix = suffix + 1;
        }
        suffixes.put(formatName, suffix);

        final String segmentSuffix = getFullSegmentSuffix(field.name,
                                                          segmentWriteState.segmentSuffix,
                                                          getSuffix(formatName, Integer.toString(suffix)));
        consumer = new FieldsConsumerAndSuffix();
        consumer.consumer = format.fieldsConsumer(new SegmentWriteState(segmentWriteState, segmentSuffix));
        consumer.suffix = suffix;
        formats.put(format, consumer);
      } else {
        // we've already seen this format, so just grab its suffix
        assert suffixes.containsKey(formatName);
        suffix = consumer.suffix;
      }

      previousValue = field.putAttribute(PER_FIELD_SUFFIX_KEY, Integer.toString(suffix));
      assert previousValue == null;

      // TODO: we should only provide the "slice" of FIS
      // that this PF actually sees ... then stuff like
      // .hasProx could work correctly?
      // NOTE: .hasProx is already broken in the same way for the non-perfield case,
      // if there is a fieldinfo with prox that has no postings, you get a 0 byte file.
      return consumer.consumer.addField(field);
    }

    @Override
    public void close() throws IOException {
      // Close all subs
      IOUtils.close(formats.values());
    }
  }

  static String getSuffix(String formatName, String suffix) {
    return formatName + "_" + suffix;
  }

  static String getFullSegmentSuffix(String fieldName, String outerSegmentSuffix, String segmentSuffix) {
    if (outerSegmentSuffix.length() == 0) {
      return segmentSuffix;
    } else {
      // TODO: support embedding; I think it should work but
      // we need a test confirm to confirm
      // return outerSegmentSuffix + "_" + segmentSuffix;
      throw new IllegalStateException("cannot embed PerFieldPostingsFormat inside itself (field \"" + fieldName + "\" returned PerFieldPostingsFormat)");
    }
  }

  private class FieldsReader extends FieldsProducer {

    private final Map<String,FieldsProducer> fields = new TreeMap<String,FieldsProducer>();
    private final Map<String,FieldsProducer> formats = new HashMap<String,FieldsProducer>();

    public FieldsReader(final SegmentReadState readState) throws IOException {

      // Read _X.per and init each format:
      boolean success = false;
      try {
        // Read field name -> format name
        for (FieldInfo fi : readState.fieldInfos) {
          if (fi.isIndexed()) {
            final String fieldName = fi.name;
            final String formatName = fi.getAttribute(PER_FIELD_FORMAT_KEY);
            if (formatName != null) {
              // null formatName means the field is in fieldInfos, but has no postings!
              final String suffix = fi.getAttribute(PER_FIELD_SUFFIX_KEY);
              assert suffix != null;
              PostingsFormat format = PostingsFormat.forName(formatName);
              String segmentSuffix = getSuffix(formatName, suffix);
              if (!formats.containsKey(segmentSuffix)) {
                formats.put(segmentSuffix, format.fieldsProducer(new SegmentReadState(readState, segmentSuffix)));
              }
              fields.put(fieldName, formats.get(segmentSuffix));
            }
          }
        }
        success = true;
      } finally {
        if (!success) {
          IOUtils.closeWhileHandlingException(formats.values());
        }
      }
    }

    @Override
    public Iterator<String> iterator() {
      return Collections.unmodifiableSet(fields.keySet()).iterator();
    }

    @Override
    public Terms terms(String field) throws IOException {
      FieldsProducer fieldsProducer = fields.get(field);
      return fieldsProducer == null ? null : fieldsProducer.terms(field);
    }

    @Override
    public int size() {
      return fields.size();
    }

    @Override
    public void close() throws IOException {
      IOUtils.close(formats.values());
    }
  }

  @Override
  public final FieldsProducer fieldsProducer(SegmentReadState state)
      throws IOException {
    return new FieldsReader(state);
  }

  /**
   * Returns the postings format that should be used for writing
   * new segments of field.
   *

   * The field to format mapping is written to the index, so
   * this method is only invoked when writing, not when reading. */
  public abstract PostingsFormat getPostingsFormatForField(String field);
}

redpajama_set_name: RedPajamaGithub


text:
<meta http-equiv="Content-Language" content="en-us">
<link rel="stylesheet" type="text/css" href="stylesheets/style.css">
<title>Writing a Simple Buildfile</title>

Using Apache Ant

Writing a Simple Buildfile

Apache Ant's buildfiles are written in XML. Each buildfile contains one project
and at least one (default) target. Targets contain task elements.
Each task element of the buildfile can have an id attribute and
can later be referred to by the value supplied to this. The value has
to be unique. (For additional information, see the
 Tasks section below.)

Projects

A project has three attributes:

    Attribute
    Description
    Required

    name
    the name of the project.
    No

    default
    the default target to use when no target is supplied.
    No; however, since Ant 1.6.0,
        every project includes an implicit target that contains any and
        all top-level tasks and/or types. This target will always be
        executed as part of the project's initialization, even when Ant is
        run with the -projecthelp option.

    basedir
    the base directory from which all path calculations are
      done. This attribute might be overridden by setting
      the "basedir"
      property beforehand. When this is done, it must be omitted in the
      project tag. If neither the attribute nor the property have
      been set, the parent directory of the buildfile will be used.
    No

Optionally, a description for the project can be provided as a
top-level <description> element (see the description type).

Each project defines one or more targets.
A target is a set of tasks you want
to be executed. When starting Ant, you can select which target(s) you
want to have executed. When no target is given,
the project's default is used.

Targets

A target can depend on other targets. You might have a target for compiling,
for example, and a target for creating a distributable. You can only build a
distributable when you have compiled first, so the distribute target
depends on the compile target. Ant resolves these dependencies.

It should be noted, however, that Ant's depends attribute
only specifies the order in which targets should be executed - it
does not affect whether the target that specifies the dependency(s) gets
executed if the dependent target(s) did not (need to) run.

More information can be found in the
  dedicated manual page.

Tasks

A task is a piece of code that can be executed.

A task can have multiple attributes (or arguments, if you prefer). The value
of an attribute might contain references to a property. These references will be
resolved before the task is executed.

Tasks have a common structure:

<name attribute1="value1" attribute2="value2" ... />

where name is the name of the task,
attributeN is the attribute name, and
valueN is the value for this attribute.

There is a set of built-in tasks, but it is also very
easy to write your own.

All tasks share a task name attribute. The value of
this attribute will be used in the logging messages generated by
Ant.

Tasks can be assigned an id attribute:

<taskname id="taskID" ... />

where taskname is the name of the task, and taskID is
a unique identifier for this task.
You can refer to the
corresponding task object in scripts or other tasks via this name.
For example, in scripts you could do:

<script ... >
  task1.setFoo("bar");
</script>

to set the foo attribute of this particular task instance.
In another task (written in Java), you can access the instance via
project.getReference("task1").

Note1: If "task1" has not been run yet, then
it has not been configured (ie., no attributes have been set), and if it is
going to be configured later, anything you've done to the instance may
be overwritten.

Note2: Future versions of Ant will most likely not
be backward-compatible with this behaviour, since there will likely be no
task instances at all, only proxies.

Properties

Properties are an important way to customize a build process or
  to just provide shortcuts for strings that are used repeatedly
  inside a build file.

In its most simple form properties are defined in the build file
  (for example by the property
  task) or might be set outside Ant.  A property has a name and a
  value; the name is case-sensitive. Properties may be used in the
  value of task attributes or in the nested text of tasks that support
  them.  This is done by placing the property name between
  "${" and "}" in the
  attribute value. For example, if there is a "builddir"
  property with the value "build", then this could be used
  in an attribute like this: ${builddir}/classes.  This
  is resolved at run-time as build/classes.

With Ant 1.8.0 property expansion has become much more powerful
  than simple key value pairs, more details can be
  found in the concepts section of this
  manual.

Example Buildfile

<project name="MyProject" default="dist" basedir=".">
    <description>
        simple example build file
    </description>
  <!-- set global properties for this build -->
  <property name="src" location="src"/>
  <property name="build" location="build"/>
  <property name="dist"  location="dist"/>

  <target name="init">
    <!-- Create the time stamp -->
    <tstamp/>
    <!-- Create the build directory structure used by compile -->
    <mkdir dir="${build}"/>
  </target>

  <target name="compile" depends="init"
        description="compile the source " >
    <!-- Compile the java code from ${src} into ${build} -->
    <javac srcdir="${src}" destdir="${build}"/>
  </target>

  <target name="dist" depends="compile"
        description="generate the distribution" >
    <!-- Create the distribution directory -->
    <mkdir dir="${dist}/lib"/>

    <!-- Put everything in ${build} into the MyProject-${DSTAMP}.jar file -->
    <jar jarfile="${dist}/lib/MyProject-${DSTAMP}.jar" basedir="${build}"/>
  </target>

  <target name="clean"
        description="clean up" >
    <!-- Delete the ${build} and ${dist} directory trees -->
    <delete dir="${build}"/>
    <delete dir="${dist}"/>
  </target>
</project>

Notice that we are declaring properties outside any target.  As of
Ant 1.6 all tasks can be declared outside targets (earlier version
only allowed <tt><property></tt>,<tt><typedef></tt> and
<tt><taskdef></tt>).  When you do this they are evaluated before
any targets are executed.  Some tasks will generate build failures if
they are used outside of targets as they may cause infinite loops
otherwise (<antcall> for example).

We have given some targets descriptions; this causes the <tt>projecthelp</tt>
invocation option to list them as public targets with the descriptions; the
other target is internal and not listed.

Finally, for this target to work the source in the <tt>src</tt> subdirectory
should be stored in a directory tree which matches the package names. Check the
<tt><javac></tt> task for details.

Token Filters

A project can have a set of tokens that might be automatically expanded if
found when a file is copied, when the filtering-copy behavior is selected in the
tasks that support this. These might be set in the buildfile
by the filter task.

Since this can potentially be a very harmful behavior,
the tokens in the files must
be of the form @token@, where
token is the token name that is set
in the <filter> task. This token syntax matches the syntax of other build systems
that perform such filtering and remains sufficiently orthogonal to most
programming and scripting languages, as well as with documentation systems.

Note: If a token with the format @token@
is found in a file, but no
filter is associated with that token, no changes take place;
therefore, no escaping
method is available - but as long as you choose appropriate names for your
tokens, this should not cause problems.

Warning: If you copy binary files with filtering turned on, you can corrupt the
files. This feature should be used with text files only.

Path-like Structures

You can specify PATH- and CLASSPATH-type
references using both
":" and ";" as separator
characters. Ant will
convert the separator to the correct character of the current operating
system.

Wherever path-like values need to be specified, a nested element can
be used. This takes the general form of:

    <classpath>
      <pathelement path="${classpath}"/>
      <pathelement location="lib/helper.jar"/>
    </classpath>

The location attribute specifies a single file or
directory relative to the project's base directory (or an absolute
filename), while the path attribute accepts colon-
or semicolon-separated lists of locations. The path
attribute is intended to be used with predefined paths - in any other
case, multiple elements with location attributes should be
preferred.

Since Ant 1.8.2 the location attribute can also contain a
  wildcard in its last path component (i.e. it can end in a
  "*") in order to support wildcard CLASSPATHs introduced
  with Java6.  Ant will not expand or evaluate the wildcards and the
  resulting path may not work as anything else but a CLASSPATH - or
  even as a CLASSPATH for a Java VM prior to Java6.

As a shortcut, the <classpath> tag
supports path and
location attributes of its own, so:

    <classpath>
      <pathelement path="${classpath}"/>
    </classpath>

can be abbreviated to:

    <classpath path="${classpath}"/>

In addition, one or more
Resource Collections
can be specified as nested elements (these must consist of
file-type resources only).
Additionally, it should be noted that although resource collections are
processed in the order encountered, certain resource collection types
such as fileset,
dirset and
files
are undefined in terms of order.

    <classpath>
      <pathelement path="${classpath}"/>
      <fileset dir="lib">
        <include name="**/*.jar"/>
      </fileset>
      <pathelement location="classes"/>
      <dirset dir="${build.dir}">
        <include name="apps/**/classes"/>
        <exclude name="apps/**/*Test*"/>
      </dirset>
      <filelist refid="third-party_jars"/>
    </classpath>

This builds a path that holds the value of ${classpath},
followed by all jar files in the lib directory,
the classes directory, all directories named
classes under the apps subdirectory of
${build.dir}, except those
that have the text Test in their name, and
the files specified in the referenced FileList.

If you want to use the same path-like structure for several tasks,
you can define them with a <path> element at the
same level as targets, and reference them via their
id attribute--see References for an
example.

By default a path like structure will re-evaluate all nested
  resource collections whenever it is used, which may lead to
  unnecessary re-scanning of the filesystem.  Since Ant 1.8.0 path has
  an optional cache attribute, if it is set to true, the path
  instance will only scan its nested resource collections once and
  assume it doesn't change during the build anymore (the default
  for cache still is false).  Even if you are using the
  path only in a single task it may improve overall performance to set
  cache to true if you are using complex nested
  constructs.

A path-like structure can include a reference to another path-like
structure (a path being itself a resource collection)
via nested <path> elements:

    <path id="base.path">
      <pathelement path="${classpath}"/>
      <fileset dir="lib">
        <include name="**/*.jar"/>
      </fileset>
      <pathelement location="classes"/>
    </path>

    <path id="tests.path" cache="true">
      <path refid="base.path"/>
      <pathelement location="testclasses"/>
    </path>

 The shortcuts previously mentioned for <classpath> are also valid for <path>.For example:

    <path id="base.path">
      <pathelement path="${classpath}"/>
    </path>

can be written as:

    <path id="base.path" path="${classpath}"/>

Path Shortcut

    In Ant 1.6 a shortcut for converting paths to OS specific strings
    in properties has been added. One can use the expression
    ${toString:pathreference} to convert a path element
    reference to a string that can be used for a path argument.
    For example:

  <path id="lib.path.ref">
    <fileset dir="lib" includes="*.jar"/>
  </path>
  <javac srcdir="src" destdir="classes">
    <compilerarg arg="-Xbootclasspath/p:${toString:lib.path.ref}"/>
  </javac>

Command-line Arguments

Several tasks take arguments that will be passed to another
process on the command line. To make it easier to specify arguments
that contain space characters, nested arg elements can be used.

  Attribute
  Description
  Required

    value
    a single command-line argument; can contain space
      characters.
    Exactly one of these.

    file
    The name of a file as a single command-line
      argument; will be replaced with the absolute filename of the file.

    path
    A string that will be treated as a path-like
      string as a single command-line argument; you can use ;
      or : as
      path separators and Ant will convert it to the platform's local
      conventions.

    pathref
    Reference to a path
      defined elsewhere.  Ant will convert it to the platform's local
      conventions.

    line
    a space-delimited list of command-line arguments.

    prefix
    A fixed string to be placed in front of the
      argument. In the case of a line broken into parts, it will be
      placed in front of every part. Since Ant 1.8.
    No

    suffix
    A fixed string to be placed immediately after the
      argument. In the case of a line broken into parts, it will be
      placed after every part. Since Ant 1.8.
    No

It is highly recommended to avoid the line version
when possible.  Ant will try to split the command line in a way
similar to what a (Unix) shell would do, but may create something that
is very different from what you expect under some circumstances.

Examples

  <arg value="-l -a"/>

is a single command-line argument containing a space character,
not separate commands "-l" and "-a".

  <arg line="-l -a"/>

This is a command line with two separate arguments, "-l" and "-a".

  <arg path="/dir;/dir2:\dir3"/>

is a single command-line argument with the value
\dir;\dir2;\dir3 on DOS-based systems and
/dir:/dir2:/dir3 on Unix-like systems.

References

Any project element can be assigned an identifier using its
id attribute. In most cases the element can subsequently
be referenced by specifying the refid attribute on an
element of the same type.  This can be useful if you are going to
replicate the same snippet of XML over and over again--using a
<classpath> structure more than once, for example.

The following example:

<project ... >
  <target ... >
    <rmic ...>
      <classpath>
        <pathelement location="lib/"/>
        <pathelement path="${java.class.path}/"/>
        <pathelement path="${additional.path}"/>
      </classpath>
    </rmic>
  </target>

  <target ... >
    <javac ...>
      <classpath>
        <pathelement location="lib/"/>
        <pathelement path="${java.class.path}/"/>
        <pathelement path="${additional.path}"/>
      </classpath>
    </javac>
  </target>
</project>

could be rewritten as:

<project ... >
  <path id="project.class.path">
    <pathelement location="lib/"/>
    <pathelement path="${java.class.path}/"/>
    <pathelement path="${additional.path}"/>
  </path>

  <target ... >
    <rmic ...>
      <classpath refid="project.class.path"/>
    </rmic>
  </target>

  <target ... >
    <javac ...>
      <classpath refid="project.class.path"/>
    </javac>
  </target>
</project>

All tasks that use nested elements for
PatternSets,
FileSets,
ZipFileSets or
path-like structures accept references to these structures
as shown in the examples. Using refid on a task will ordinarily
have the same effect (referencing a task already declared), but the user
should be aware that the interpretation of this attribute is dependent on the
implementation of the element upon which it is specified. Some tasks (the
property task is a handy example)
deliberately assign a different meaning to refid.

Use of external tasks

Ant supports a plugin mechanism for using third party tasks. For using them you
have to do two steps:

place their implementation somewhere where Ant can find them

declare them.

Don't add anything to the CLASSPATH environment variable - this is often the
reason for very obscure errors. Use Ant's own mechanisms
for adding libraries:

via command line argument -lib

adding to ${user.home}/.ant/lib

adding to ${ant.home}/lib

For the declaration there are several ways:

declare a single task per using instruction using
      <taskdef name="taskname"
      classname="ImplementationClass"/>

      <taskdef name="for" classname="net.sf.antcontrib.logic.For" />
      <for ... />

declare a bundle of tasks using a properties-file holding these
      taskname-ImplementationClass-pairs and <taskdef>

      <taskdef resource="net/sf/antcontrib/antcontrib.properties" />
      <for ... />

declare a bundle of tasks using a xml-file holding these
      taskname-ImplementationClass-pairs and <taskdef>

      <taskdef resource="net/sf/antcontrib/antlib.xml" />
      <for ... />

declare a bundle of tasks using a xml-file named antlib.xml, XML-namespace and
      antlib: protocoll handler

      <project xmlns:ac="antlib:net.sf.antconrib"/>
      <ac:for ... />

If you need a special function, you should

have a look at this manual, because Ant provides lot of tasks

have a look at the external task page in the manual
      (or better online)

have a look at the external task wiki
      page

ask on the Ant user list

implement (and share) your own


text:
load 'map_cordinate4_dotted';
load state.mat % variable S
% rand('state',S);
rng(S);

nSegmentsLD=length(map_x)/300 %300 data points corresponding to about 15meters
segmentLengthLD=round(400+rand(1,nSegmentsLD)*(500-300));
csH = cumsum(segmentLengthLD);
nSegmentsLD = length(find(cumsum(segmentLengthLD)<length(map_x))) ;

% extend length of last segment to complete the road
segmentLengthLD(nSegmentsLD) = length(map_x) - sum(segmentLengthLD(1:nSegmentsLD-1));
segmentLengthLD = segmentLengthLD(1:nSegmentsLD);
segmentLD=zeros(1,nSegmentsLD,);
segmentLD(1) = (simulated_lane_width/2) ; % begin at the center of the left lane
segmentLD(2:end)=simulated_lane_width/2+maximumLateralDisplacement*2*(rand(1,nSegmentsLD-1)-0.5);

%-------------------------------------
% Computation of the camera trajectory
%-------------------------------------
% maximum possible number of segments of constant lateral displacement
nSegmentsLD = roadLength/minSegmentLengthLD ;
segmentLengthLD = round(minSegmentLengthLD + ...
                  rand(1,nSegmentsLD)*(maxSegmentLengthLD - minSegmentLengthLD));
csH = cumsum(segmentLengthLD);
nSegmentsLD = length(find(cumsum(segmentLengthLD)<roadLength)) ;
% extend length of last segment to complete the road
segmentLengthLD(nSegmentsLD) = roadLength - sum(segmentLengthLD(1:nSegmentsLD-1));
segmentLengthLD = segmentLengthLD(1:nSegmentsLD);
segmentLD(1) = (laneWidth/2) ; % begin at the center of the right lane
for i=2:nSegmentsLD
    % keep always in the lane to the right of the road center (two lanes=>right lane)
    segmentLD(i) = (laneWidth/2) + maximumLateralDisplacement*2*(rand(1)-0.5) ;
end
ld = [] ;
for i=1:nSegmentsLD
    ld  = [ld ; segmentLD(i)*ones(segmentLengthLD(i),1)];
end
% heavy smoothing
t = double(-windowSizeLD/2:windowSizeLD/2);
filter = (1/(sqrt(2*pi)*sigmaLD))*exp(-(t.*t)/(2*sigmaLD*sigmaLD));
ld = smoothVector(ld, filter);


text:
import unittest

class Layer(object):

    @classmethod
    def setUp(cls):
        raise RuntimeError('Bad Error in Layer setUp!')

class Test(unittest.TestCase):
    layer = Layer

    def testPass(self):
        pass


text:
<?php

/**
 * @license https://github.com/f500/swiftmailer-sparkpost/blob/master/LICENSE Proprietary
 */

namespace SwiftSparkPost;

/**
 * @copyright Future500 B.V.
 * @author    Jasper N. Brouwer <jasper@future500.nl>
 */
final class Configuration
{
    use OptionsSanitizingCapabilities;

    /**
     * @var string
     */
    private $recipientOverride;

    /**
     * @var bool
     */
    private $overrideGmailStyle;

    /**
     * @var array
     */
    private $options;

    /**
     * @var float
     */
    private $ipPoolProbability;

    /**
     * @return Configuration
     */
    public static function newInstance(): Configuration
    {
        return new self();
    }

    public function __construct()
    {
        $this->recipientOverride  = '';
        $this->overrideGmailStyle = false;
        $this->options            = [Option::TRANSACTIONAL => true];
        $this->ipPoolProbability  = 1.0;
    }

    public function overrideRecipients(): bool
    {
        return $this->recipientOverride !== '';
    }

    /**
     * @return bool
     */
    public function overrideGmailStyle(): bool
    {
        return $this->overrideGmailStyle;
    }

    /**
     * @param bool $overrideGmailStyle
     *
     * @return Configuration
     */
    public function setOverrideGmailStyle(bool $overrideGmailStyle): Configuration
    {
        $this->overrideGmailStyle = (bool) $overrideGmailStyle;

        return $this;
    }

    /**
     * @return string
     */
    public function getRecipientOverride(): string
    {
        return $this->recipientOverride;
    }

    /**
     * @param string|null $recipientOverride
     *
     * @return Configuration
     * @throws Exception
     */
    public function setRecipientOverride(?string $recipientOverride): Configuration
    {
        if (!$recipientOverride) {
            return $this;
        }

        if (!filter_var($recipientOverride, FILTER_VALIDATE_EMAIL)) {
            throw new Exception('Recipient override must be a valid email address');
        }

        $this->recipientOverride = (string) $recipientOverride;

        return $this;
    }

    /**
     * @return array
     */
    public function getOptions(): array
    {
        return $this->options;
    }

    /**
     * @param array $options
     *
     * @return Configuration
     */
    public function setOptions(array $options): Configuration
    {
        $this->options = array_merge(
            $this->options,
            $this->sanitizeOptions($options)
        );

        return $this;
    }

    /**
     * @return float
     */
    public function getIpPoolProbability(): float
    {
        return $this->ipPoolProbability;
    }

    /**
     * @param float $ipPoolProbability
     *
     * @return Configuration
     * @throws Exception
     */
    public function setIpPoolProbability(float $ipPoolProbability): Configuration
    {
        if ($ipPoolProbability < 0 || $ipPoolProbability > 1) {
            throw new Exception('IP pool probability must be between 0 and 1');
        }

        $this->ipPoolProbability = (float) $ipPoolProbability;

        return $this;
    }
}