@Override
    public void processElement(StreamRecord<RowData> element) throws Exception {
        RowData rowData = element.getValue();
        int localRowCount = rowData.getInt(0);

        if (globalRowCount == OVER_MAX_ROW_COUNT) {
            // Current global filter is already over-max-row-count, do nothing.
        } else if (localRowCount == OVER_MAX_ROW_COUNT
                || globalRowCount + localRowCount > maxRowCount) {
            // The received local filter is over-max-row-count, mark the global filter as
            // over-max-row-count.
            globalRowCount = OVER_MAX_ROW_COUNT;
            serializedGlobalFilter = null;
        } else {
            // merge the local filter
            byte[] serializedLocalFilter = rowData.getBinary(1);
            if (serializedGlobalFilter == null) {
                serializedGlobalFilter = serializedLocalFilter.clone();
            } else {
                BloomFilter.mergeSerializedBloomFilters(
                        serializedGlobalFilter, serializedLocalFilter);
            }
            globalRowCount += localRowCount;
        }
    }
@Test
    void testNormalInputAndNormalOutput() throws Exception {
        try (StreamTaskMailboxTestHarness<RowData> testHarness =
                createGlobalRuntimeFilterBuilderOperatorHarness(10)) {
            // process elements
            testHarness.processElement(
                    new StreamRecord<RowData>(
                            GenericRowData.of(5, BloomFilter.toBytes(createBloomFilter1()))));
            testHarness.processElement(
                    new StreamRecord<RowData>(
                            GenericRowData.of(5, BloomFilter.toBytes(createBloomFilter2()))));
            testHarness.processEvent(new EndOfData(StopMode.DRAIN), 0);

            // test the output
            Queue<Object> outputs = testHarness.getOutput();
            assertThat(outputs.size()).isEqualTo(1);

            RowData outputRowData = ((StreamRecord<RowData>) outputs.poll()).getValue();
            assertThat(outputRowData.getArity()).isEqualTo(2);

            int globalCount = outputRowData.getInt(0);
            BloomFilter globalBloomFilter = BloomFilter.fromBytes(outputRowData.getBinary(1));
            assertThat(globalCount).isEqualTo(10);
            assertThat(globalBloomFilter.testHash("var1".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var2".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var3".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var4".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var5".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var6".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var7".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var8".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var9".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var10".hashCode())).isTrue();
            assertThat(globalBloomFilter.testHash("var11".hashCode())).isFalse();
            assertThat(globalBloomFilter.testHash("var12".hashCode())).isFalse();
            assertThat(globalBloomFilter.testHash("var13".hashCode())).isFalse();
            assertThat(globalBloomFilter.testHash("var14".hashCode())).isFalse();
            assertThat(globalBloomFilter.testHash("var15".hashCode())).isFalse();
        }
    }