### Question:
StructuredDataSourceNode extends PlanNode { @Override public Field getKeyField() { return structuredDataSource.getKeyField(); } @JsonCreator  StructuredDataSourceNode(
      @JsonProperty("id") final PlanNodeId id,
      @JsonProperty("structuredDataSource") final StructuredDataSource structuredDataSource,
      @JsonProperty("schema") Schema schema
  ); String getTopicName(); @Override Schema getSchema(); @Override Field getKeyField(); StructuredDataSource getStructuredDataSource(); @Override int getPartitions(KafkaTopicClient kafkaTopicClient); @Override List<PlanNode> getSources(); @Override R accept(PlanVisitor<C, R> visitor, C context); @Override SchemaKStream buildStream(
      final StreamsBuilder builder,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient kafkaTopicClient,
      final FunctionRegistry functionRegistry,
      final Map<String, Object> props,
      final SchemaRegistryClient schemaRegistryClient
  ); StructuredDataSource.DataSourceType getDataSourceType(); TimestampExtractionPolicy getTimestampExtractionPolicy();  }### Answer:
@Test public void shouldExtracKeyField() { assertThat(stream.getKeyField(), equalTo(new Field("key", 4, Schema.STRING_SCHEMA))); }