KafkaIO SDF does not commit offsets but KafkaIO UnboundedSource does

        final Pipeline p = Pipeline.create(options);

        p.apply(

            KafkaIO.<Long, String>read()

                .withBootstrapServers(options.getKafkaBroker())

                .withTopic(options.getTopic())

                .withConsumerConfigUpdates(

                    Map.of(

                        ConsumerConfig.GROUP_ID_CONFIG, options.getConsumerGroup(),

                        CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL",

                        SaslConfigs.SASL_KERBEROS_SERVICE_NAME, "kafka",

                        SaslConfigs.SASL_JAAS_CONFIG, "com.sun.security.auth.module.GssLoginModule required initiate=true;"))

                .withKeyDeserializer(LongDeserializer.class)

                .withValueDeserializer(StringDeserializer.class)

                .commitOffsetsInFinalize()

                .withoutMetadata());