@Override
    public String getSinkTableName(Table table) {
        String tableName = table.getName();
        Map<String, String> sink = config.getSink();

        // Add table name mapping logic
        String mappingRoute = sink.get(FlinkCDCConfig.TABLE_MAPPING_ROUTES);
        if (mappingRoute != null) {
            Map<String, String> mappingRules = parseMappingRoute(mappingRoute);
            if (mappingRules.containsKey(tableName)) {
                tableName = mappingRules.get(tableName);
            }
        }

        tableName = sink.getOrDefault(FlinkCDCConfig.TABLE_PREFIX, "")
                + tableName
                + sink.getOrDefault(FlinkCDCConfig.TABLE_SUFFIX, "");
        // table.lower and table.upper can not be true at the same time
        if (Boolean.parseBoolean(sink.get(FlinkCDCConfig.TABLE_LOWER))
                && Boolean.parseBoolean(sink.get(FlinkCDCConfig.TABLE_UPPER))) {
            throw new IllegalArgumentException("table.lower and table.upper can not be true at the same time");
        }

        if (Boolean.parseBoolean(sink.get(FlinkCDCConfig.TABLE_UPPER))) {
            tableName = tableName.toUpperCase();
        }

        if (Boolean.parseBoolean(sink.get(FlinkCDCConfig.TABLE_LOWER))) {
            tableName = tableName.toLowerCase();
        }
        // Implement regular expressions to replace table names through
        // sink.table.replace.pattern and table.replace.with
        String replacePattern = sink.get(FlinkCDCConfig.TABLE_REPLACE_PATTERN);
        String replaceWith = sink.get(FlinkCDCConfig.TABLE_REPLACE_WITH);
        if (replacePattern != null && replaceWith != null) {
            Pattern pattern = Pattern.compile(replacePattern);
            Matcher matcher = pattern.matcher(tableName);
            tableName = matcher.replaceAll(replaceWith);
        }

        // add schema
        if (Boolean.parseBoolean(sink.get("table.prefix.schema"))) {
            tableName = table.getSchema() + "_" + tableName;
        }

        return tableName;
    }
@Test
    public void testGetSinkTableNameWithConversionUpperAndLowerCase() {
        Map<String, String> sinkConfig = new HashMap<>();
        sinkConfig.put("table.prefix", "");
        sinkConfig.put("table.suffix", "");
        sinkConfig.put("table.lower", "true");
        sinkConfig.put("table.upper", "true");
        when(config.getSink()).thenReturn(sinkConfig);

        Table table = new Table("TestTable", "TestSchema", null);
        Assert.assertThrows(IllegalArgumentException.class, () -> sinkBuilder.getSinkTableName(table));
    }