### Question:
PhysicalPlanBuilder { String getServiceId() { return KsqlConstants.KSQL_INTERNAL_TOPIC_PREFIX + ksqlConfig.get(KsqlConfig.KSQL_SERVICE_ID_CONFIG).toString(); } PhysicalPlanBuilder(
      final StreamsBuilder builder,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient kafkaTopicClient,
      final FunctionRegistry functionRegistry,
      final Map<String, Object> overriddenStreamsProperties,
      final boolean updateMetastore,
      final MetaStore metaStore,
      final SchemaRegistryClient schemaRegistryClient,
      final KafkaStreamsBuilder kafkaStreamsBuilder
  );  PhysicalPlanBuilder(
      final StreamsBuilder builder,
      final KsqlConfig ksqlConfig,
      final KafkaTopicClient kafkaTopicClient,
      final FunctionRegistry functionRegistry,
      final Map<String, Object> overriddenStreamsProperties,
      final boolean updateMetastore,
      final MetaStore metaStore,
      final SchemaRegistryClient schemaRegistryClient
  ); QueryMetadata buildPhysicalPlan(final Pair<String, PlanNode> statementPlanPair);  }### Answer:
@Test public void shouldCreateExpectedServiceId() { String serviceId = physicalPlanBuilder.getServiceId(); assertThat(serviceId, equalTo(KsqlConstants.KSQL_INTERNAL_TOPIC_PREFIX + KsqlConfig.KSQL_SERVICE_ID_DEFAULT)); }