@Override
    public int partition(T record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
        try {
            for (Map.Entry<String, String> entry : patternPartitionMap.entrySet()) {
                if (DEFAULT_PARTITION.equals(entry.getKey())) {
                    continue;
                }
                String databaseName = dynamicSchemaFormat.parse(value, databasePattern);
                String tableName = dynamicSchemaFormat.parse(value, tablePattern);
                List<String> regexList = Arrays.asList(entry.getKey().split(DELIMITER1));
                String databaseNameRegex = regexList.get(0);
                String tableNameRegex = regexList.get(1);

                if (match(databaseName, databaseNameRegex) && match(tableName, tableNameRegex)) {
                    return Integer.parseInt(entry.getValue());
                }
            }
        } catch (Exception e) {
            LOG.warn("Extract partition failed", e);
        }
        return defaultPartitionId;
    }
Partition mapping is performed according to the library name and table name rules,
and no matching is written to the default partition.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kafka/src/main/java/org/apache/inlong/sort/kafka/partitioner/InLongFixedPartitionPartitioner.java