int preferredLocalParallelism() {
        if (options.containsKey(SqlConnector.OPTION_PREFERRED_LOCAL_PARALLELISM)) {
            return Integer.parseInt(options.get(SqlConnector.OPTION_PREFERRED_LOCAL_PARALLELISM));
        }
        return StreamKafkaP.PREFERRED_LOCAL_PARALLELISM;
    }
@Test
    @Parameters(method = "preferredLocalParallelisms")
    public void when_preferredLocalParallelism_isDefined_then_parseInt(String plp, Integer expected, boolean shouldThrow) {
        KafkaTable table = new KafkaTable(
                null, null, null, null, null, null, null,
                Map.of(OPTION_PREFERRED_LOCAL_PARALLELISM, plp),
                null, null, null, null, null
        );

        if (shouldThrow) {
            assertThatThrownBy(() -> table.preferredLocalParallelism())
                    .isInstanceOf(NumberFormatException.class);
        } else {
            assertThat(table.preferredLocalParallelism()).isEqualTo(expected);
        }
    }