code: public static KafkaPartition getPartition(State state) {
      Preconditions.checkArgument(state.contains(KafkaSource.TOPIC_NAME),
          "Missing configuration property " + KafkaSource.TOPIC_NAME);
      Preconditions.checkArgument(state.contains(KafkaSource.PARTITION_ID),
          "Missing configuration property " + KafkaSource.PARTITION_ID);
  
      KafkaPartition.Builder builder = new KafkaPartition.Builder().withTopicName(state.getProp(KafkaSource.TOPIC_NAME))
          .withId(state.getPropAsInt(KafkaSource.PARTITION_ID));
      if (state.contains(KafkaSource.LEADER_ID)) {
        builder = builder.withLeaderId(state.getPropAsInt(KafkaSource.LEADER_ID));
      }
      if (state.contains(KafkaSource.LEADER_HOSTANDPORT)) {
        builder = builder.withLeaderHostAndPort(state.getProp(KafkaSource.LEADER_HOSTANDPORT));
      }
      return builder.build();
    }
docs: Get {@link KafkaPartition} from a {@link State} object. The {@link State} should contain properties
  {@link KafkaSource#TOPIC_NAME}, {@link KafkaSource#PARTITION_ID}, and may optionally contain
  {@link KafkaSource#LEADER_ID} and {@link KafkaSource#LEADER_HOSTANDPORT}.