@SuppressWarnings("unchecked")
    @Override
    protected List<OUT> executeOnCollections(
            List<IN1> inputData1,
            List<IN2> inputData2,
            RuntimeContext runtimeContext,
            ExecutionConfig executionConfig)
            throws Exception {
        FlatJoinFunction<IN1, IN2, OUT> function = userFunction.getUserCodeObject();

        FunctionUtils.setFunctionRuntimeContext(function, runtimeContext);
        FunctionUtils.openFunction(function, DefaultOpenContext.INSTANCE);

        TypeInformation<IN1> leftInformation = getOperatorInfo().getFirstInputType();
        TypeInformation<IN2> rightInformation = getOperatorInfo().getSecondInputType();
        TypeInformation<OUT> outInformation = getOperatorInfo().getOutputType();

        TypeSerializer<IN1> leftSerializer =
                leftInformation.createSerializer(executionConfig.getSerializerConfig());
        TypeSerializer<IN2> rightSerializer =
                rightInformation.createSerializer(executionConfig.getSerializerConfig());

        TypeComparator<IN1> leftComparator;
        TypeComparator<IN2> rightComparator;

        if (leftInformation instanceof AtomicType) {
            leftComparator =
                    ((AtomicType<IN1>) leftInformation).createComparator(true, executionConfig);
        } else if (leftInformation instanceof CompositeType) {
            int[] keyPositions = getKeyColumns(0);
            boolean[] orders = new boolean[keyPositions.length];
            Arrays.fill(orders, true);

            leftComparator =
                    ((CompositeType<IN1>) leftInformation)
                            .createComparator(keyPositions, orders, 0, executionConfig);
        } else {
            throw new RuntimeException(
                    "Type information for left input of type "
                            + leftInformation.getClass().getCanonicalName()
                            + " is not supported. Could not generate a comparator.");
        }

        if (rightInformation instanceof AtomicType) {
            rightComparator =
                    ((AtomicType<IN2>) rightInformation).createComparator(true, executionConfig);
        } else if (rightInformation instanceof CompositeType) {
            int[] keyPositions = getKeyColumns(1);
            boolean[] orders = new boolean[keyPositions.length];
            Arrays.fill(orders, true);

            rightComparator =
                    ((CompositeType<IN2>) rightInformation)
                            .createComparator(keyPositions, orders, 0, executionConfig);
        } else {
            throw new RuntimeException(
                    "Type information for right input of type "
                            + rightInformation.getClass().getCanonicalName()
                            + " is not supported. Could not generate a comparator.");
        }

        TypePairComparator<IN1, IN2> pairComparator =
                new GenericPairComparator<IN1, IN2>(leftComparator, rightComparator);

        List<OUT> result = new ArrayList<OUT>();
        Collector<OUT> collector =
                new CopyingListCollector<OUT>(
                        result,
                        outInformation.createSerializer(executionConfig.getSerializerConfig()));

        Map<Integer, List<IN2>> probeTable = new HashMap<Integer, List<IN2>>();

        // Build hash table
        for (IN2 element : inputData2) {
            List<IN2> list = probeTable.get(rightComparator.hash(element));
            if (list == null) {
                list = new ArrayList<IN2>();
                probeTable.put(rightComparator.hash(element), list);
            }

            list.add(element);
        }

        // Probing
        for (IN1 left : inputData1) {
            List<IN2> matchingHashes = probeTable.get(leftComparator.hash(left));

            if (matchingHashes != null) {
                pairComparator.setReference(left);
                for (IN2 right : matchingHashes) {
                    if (pairComparator.equalToReference(right)) {
                        function.join(
                                leftSerializer.copy(left), rightSerializer.copy(right), collector);
                    }
                }
            }
        }

        FunctionUtils.closeFunction(function);

        return result;
    }
@Test
    void testJoinRich() throws Exception {
        final AtomicBoolean opened = new AtomicBoolean(false);
        final AtomicBoolean closed = new AtomicBoolean(false);
        final String taskName = "Test rich join function";

        final RichFlatJoinFunction<String, String, Integer> joiner =
                new RichFlatJoinFunction<String, String, Integer>() {
                    @Override
                    public void open(OpenContext openContext) {
                        opened.compareAndSet(false, true);
                        assertThat(getRuntimeContext().getTaskInfo().getIndexOfThisSubtask())
                                .isZero();
                        assertThat(getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks())
                                .isOne();
                    }

                    @Override
                    public void close() {
                        closed.compareAndSet(false, true);
                    }

                    @Override
                    public void join(String first, String second, Collector<Integer> out) {
                        out.collect(first.length());
                        out.collect(second.length());
                    }
                };

        InnerJoinOperatorBase<
                        String, String, Integer, RichFlatJoinFunction<String, String, Integer>>
                base =
                        new InnerJoinOperatorBase<>(
                                joiner,
                                new BinaryOperatorInformation<>(
                                        BasicTypeInfo.STRING_TYPE_INFO,
                                        BasicTypeInfo.STRING_TYPE_INFO,
                                        BasicTypeInfo.INT_TYPE_INFO),
                                new int[0],
                                new int[0],
                                taskName);

        final List<String> inputData1 = new ArrayList<>(Arrays.asList("foo", "bar", "foobar"));
        final List<String> inputData2 = new ArrayList<>(Arrays.asList("foobar", "foo"));
        final List<Integer> expected = new ArrayList<>(Arrays.asList(3, 3, 6, 6));

        final TaskInfo taskInfo = new TaskInfoImpl(taskName, 1, 0, 1, 0);
        final HashMap<String, Accumulator<?, ?>> accumulatorMap = new HashMap<>();
        final HashMap<String, Future<Path>> cpTasks = new HashMap<>();

        ExecutionConfig executionConfig = new ExecutionConfig();

        executionConfig.disableObjectReuse();
        List<Integer> resultSafe =
                base.executeOnCollections(
                        inputData1,
                        inputData2,
                        new RuntimeUDFContext(
                                taskInfo,
                                null,
                                executionConfig,
                                cpTasks,
                                accumulatorMap,
                                UnregisteredMetricsGroup.createOperatorMetricGroup()),
                        executionConfig);

        executionConfig.enableObjectReuse();
        List<Integer> resultRegular =
                base.executeOnCollections(
                        inputData1,
                        inputData2,
                        new RuntimeUDFContext(
                                taskInfo,
                                null,
                                executionConfig,
                                cpTasks,
                                accumulatorMap,
                                UnregisteredMetricsGroup.createOperatorMetricGroup()),
                        executionConfig);

        assertThat(resultSafe).isEqualTo(expected);
        assertThat(resultRegular).isEqualTo(expected);

        assertThat(opened).isTrue();
        assertThat(closed).isTrue();
    }