@Override
    public void processElement(I i, ProcessFunction<I, Object>.Context context, Collector<Object> collector)
            throws Exception {
        HoodieRecord<?> record = (HoodieRecord<?>) i;
        final HoodieKey hoodieKey = record.getKey();
        final String partition = hoodieKey.getPartitionPath();
        final HoodieRecordLocation location;

        bootstrapIndexIfNeed(partition);
        Map<Integer, String> bucketToFileId = bucketIndex.computeIfAbsent(partition, p -> new HashMap<>());
        final int bucketNum = BucketIdentifier.getBucketId(hoodieKey, indexKeyFields, this.bucketNum);
        final String bucketId = partition + "/" + bucketNum;

        if (incBucketIndex.contains(bucketId)) {
            location = new HoodieRecordLocation("I", bucketToFileId.get(bucketNum));
        } else if (bucketToFileId.containsKey(bucketNum)) {
            location = new HoodieRecordLocation("U", bucketToFileId.get(bucketNum));
        } else {
            String newFileId = BucketIdentifier.newBucketFileIdPrefix(bucketNum);
            location = new HoodieRecordLocation("I", newFileId);
            bucketToFileId.put(bucketNum, newFileId);
            incBucketIndex.add(bucketId);
        }
        record.unseal();
        record.setCurrentLocation(location);
        record.seal();
        bufferRecord(record);
        if (sinkMetricData != null) {
            sinkMetricData.invokeWithEstimate(record);
        }
    }
Constructs a BucketStreamWriteFunction.

@param config The config options
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/hudi/src/main/java/org/apache/inlong/sort/hudi/sink/bucket/BucketStreamWriteFunction.java