@Override
    public DataSink createDataSink(Context context) {
        FactoryHelper.createFactoryHelper(this, context)
                .validateExcept(PREFIX_TABLE_PROPERTIES, PREFIX_CATALOG_PROPERTIES);

        Map<String, String> allOptions = context.getFactoryConfiguration().toMap();
        Map<String, String> catalogOptions = new HashMap<>();
        Map<String, String> tableOptions = new HashMap<>();
        allOptions.forEach(
                (key, value) -> {
                    if (key.startsWith(PREFIX_TABLE_PROPERTIES)) {
                        tableOptions.put(key.substring(PREFIX_TABLE_PROPERTIES.length()), value);
                    } else if (key.startsWith(PaimonDataSinkOptions.PREFIX_CATALOG_PROPERTIES)) {
                        catalogOptions.put(
                                key.substring(
                                        PaimonDataSinkOptions.PREFIX_CATALOG_PROPERTIES.length()),
                                value);
                    }
                });
        Options options = Options.fromMap(catalogOptions);
        try (Catalog catalog = FlinkCatalogFactory.createPaimonCatalog(options)) {
            Preconditions.checkNotNull(
                    catalog.listDatabases(), "catalog option of Paimon is invalid.");
        } catch (Exception e) {
            throw new RuntimeException("failed to create or use paimon catalog", e);
        }
        ZoneId zoneId = ZoneId.systemDefault();
        if (!Objects.equals(
                context.getPipelineConfiguration().get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE),
                PipelineOptions.PIPELINE_LOCAL_TIME_ZONE.defaultValue())) {
            zoneId =
                    ZoneId.of(
                            context.getPipelineConfiguration()
                                    .get(PipelineOptions.PIPELINE_LOCAL_TIME_ZONE));
        }
        String commitUser =
                context.getFactoryConfiguration().get(PaimonDataSinkOptions.COMMIT_USER);
        String partitionKey =
                context.getFactoryConfiguration().get(PaimonDataSinkOptions.PARTITION_KEY);
        Map<TableId, List<String>> partitionMaps = new HashMap<>();
        if (!partitionKey.isEmpty()) {
            for (String tables : partitionKey.split(";")) {
                String[] splits = tables.split(":");
                if (splits.length == 2) {
                    TableId tableId = TableId.parse(splits[0]);
                    List<String> partitions = Arrays.asList(splits[1].split(","));
                    partitionMaps.put(tableId, partitions);
                } else {
                    throw new IllegalArgumentException(
                            PaimonDataSinkOptions.PARTITION_KEY.key()
                                    + " is malformed, please refer to the documents");
                }
            }
        }
        PaimonRecordSerializer<Event> serializer = new PaimonRecordEventSerializer(zoneId);
        String schemaOperatorUid =
                context.getPipelineConfiguration()
                        .get(PipelineOptions.PIPELINE_SCHEMA_OPERATOR_UID);
        return new PaimonDataSink(
                options,
                tableOptions,
                commitUser,
                partitionMaps,
                serializer,
                zoneId,
                schemaOperatorUid);
    }
@Test
    void testUnsupportedOption() {
        DataSinkFactory sinkFactory =
                FactoryDiscoveryUtils.getFactoryByIdentifier("paimon", DataSinkFactory.class);
        Assertions.assertThat(sinkFactory).isInstanceOf(PaimonDataSinkFactory.class);

        Configuration conf =
                Configuration.fromMap(
                        ImmutableMap.<String, String>builder()
                                .put(PaimonDataSinkOptions.METASTORE.key(), "filesystem")
                                .put(
                                        PaimonDataSinkOptions.WAREHOUSE.key(),
                                        new File(
                                                        temporaryFolder.toFile(),
                                                        UUID.randomUUID().toString())
                                                .toString())
                                .put("unsupported_key", "unsupported_value")
                                .build());

        Assertions.assertThatThrownBy(
                        () ->
                                sinkFactory.createDataSink(
                                        new FactoryHelper.DefaultContext(
                                                conf,
                                                conf,
                                                Thread.currentThread().getContextClassLoader())))
                .isInstanceOf(ValidationException.class)
                .hasMessageContaining(
                        "Unsupported options found for 'paimon'.\n\n"
                                + "Unsupported options:\n\n"
                                + "unsupported_key");
    }