### Question:
SchemaKStream { @SuppressWarnings("unchecked") public SchemaKStream selectKey(final Field newKeyField, boolean updateRowKey) { if (keyField != null && keyField.name().equals(newKeyField.name())) { return this; } KStream keyedKStream = kstream.filter((key, value) -> value != null && extractColumn(newKeyField, value) != null ).selectKey((key, value) -> extractColumn(newKeyField, value) .toString() ).mapValues((key, row) -> { if (updateRowKey) { row.getColumns().set(SchemaUtil.ROWKEY_NAME_INDEX, key); } return row; }); return new SchemaKStream( schema, keyedKStream, newKeyField, Collections.singletonList(this), Type.REKEY, functionRegistry, schemaRegistryClient ); } SchemaKStream(
      final Schema schema,
      final KStream<String, GenericRow> kstream,
      final Field keyField,
      final List<SchemaKStream> sourceSchemaKStreams,
      final Type type,
      final FunctionRegistry functionRegistry,
      final SchemaRegistryClient schemaRegistryClient
  ); QueuedSchemaKStream toQueue(Optional<Integer> limit); SchemaKStream into(
      final String kafkaTopicName,
      final Serde<GenericRow> topicValueSerDe,
      final Set<Integer> rowkeyIndexes
  ); @SuppressWarnings("unchecked") SchemaKStream filter(final Expression filterExpression); SchemaKStream select(final Schema selectSchema); SchemaKStream select(final List<Pair<String, Expression>> expressionPairList); @SuppressWarnings("unchecked") SchemaKStream leftJoin(
      final SchemaKTable schemaKTable,
      final Schema joinSchema,
      final Field joinKey,
      KsqlTopicSerDe joinSerDe,
      KsqlConfig ksqlConfig
  ); @SuppressWarnings("unchecked") SchemaKStream selectKey(final Field newKeyField, boolean updateRowKey); SchemaKGroupedStream groupBy(
      final Serde<String> keySerde, final Serde<GenericRow> valSerde,
      final List<Expression> groupByExpressions); Field getKeyField(); Schema getSchema(); KStream getKstream(); List<SchemaKStream> getSourceSchemaKStreams(); String getExecutionPlan(String indent); OutputNode outputNode(); void setOutputNode(final OutputNode output); Type getType(); FunctionRegistry getFunctionRegistry(); SchemaRegistryClient getSchemaRegistryClient();  }### Answer:
@Test public void testSelectKey() { String selectQuery = "SELECT col0, col2, col3 FROM test1 WHERE col0 > 100;"; PlanNode logicalPlan = planBuilder.buildLogicalPlan(selectQuery); initialSchemaKStream = new SchemaKStream(logicalPlan.getTheSourceNode().getSchema(), kStream, ksqlStream.getKeyField(), new ArrayList<>(), SchemaKStream.Type.SOURCE, functionRegistry, new MockSchemaRegistryClient()); SchemaKStream rekeyedSchemaKStream = initialSchemaKStream.selectKey(initialSchemaKStream .getSchema().fields() .get(3), true); assertThat(rekeyedSchemaKStream.getKeyField().name().toUpperCase(), equalTo("TEST1.COL1")); }