code: public static String getTopicName(State state) {
      Preconditions.checkArgument(state.contains(KafkaSource.TOPIC_NAME),
          "Missing configuration property " + KafkaSource.TOPIC_NAME);
  
      return state.getProp(KafkaSource.TOPIC_NAME);
    }
docs: Get topic name from a {@link State} object. The {@link State} should contain property
  {@link KafkaSource#TOPIC_NAME}.