public static TopicRouter<RowData> getTopicRouter(
            ReadableConfig readableConfig, ClassLoader classLoader) {
        if (!readableConfig.getOptional(SINK_CUSTOM_TOPIC_ROUTER).isPresent()) {
            return null;
        }

        String className = readableConfig.get(SINK_CUSTOM_TOPIC_ROUTER);
        try {
            Class<?> clazz = Class.forName(className, true, classLoader);
            if (!TopicRouter.class.isAssignableFrom(clazz)) {
                throw new ValidationException(
                        String.format(
                                "Sink TopicRouter class '%s' should extend from the required class %s",
                                className, TopicRouter.class.getName()));
            }
            @SuppressWarnings("unchecked")
            final TopicRouter<RowData> topicRouter =
                    InstantiationUtil.instantiate(className, TopicRouter.class, classLoader);

            return topicRouter;
        } catch (ClassNotFoundException | FlinkException e) {
            throw new ValidationException(
                    String.format(
                            "Could not find and instantiate TopicRouter class '%s'", className),
                    e);
        }
    }
Creates an array of indices that determine which physical fields of the table schema to
include in the key format and the order that those fields have in the key format.
inlong-sort/sort-flink/sort-flink-v1.18/sort-connectors/pulsar/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableOptionUtils.java