GH-1400: Expose topology from StreamsBuilder

Resolves #1400

Currently, StreamsBuilderFacotryBean does not give direct access to the toplogy object.
It will be beneficial for downstream clients such as the Kafka Strems binder in
Spring Cloud Stream to get access to the topology object directly.

Author: sobychacko

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

@@ -92,6 +92,8 @@ public class StreamsBuilderFactoryBean extends AbstractFactoryBean<StreamsBuilde
 
 	private volatile boolean running;
 
+	private Topology topology;
+
 	/**
 	 * Default constructor that creates the factory without configuration
 	 * {@link Properties}. It is the factory user's responsibility to properly set
@@ -254,6 +256,15 @@ public void setCloseTimeout(int closeTimeout) {
 		this.closeTimeout = Duration.ofSeconds(closeTimeout); // NOSONAR (sync)
 	}
 
+	/**
+	 * Providing access to the associated {@link Topology} of this {@link StreamsBuilderFactoryBean}.
+	 * @return {@link Topology} object
+	 * @since 2.3.7
+	 */
+	public Topology getTopology() {
+		return this.topology;
+	}
+
 	@Override
 	public Class<?> getObjectType() {
 		return StreamsBuilder.class;
@@ -311,6 +322,7 @@ public synchronized void start() {
 				Assert.state(this.properties != null,
 						"streams configuration properties must not be null");
 				Topology topology = getObject().build(this.properties); // NOSONAR: getObject() cannot return null
+				this.topology = topology;
 				LOGGER.debug(() -> topology.describe().toString());
 				this.kafkaStreams = new KafkaStreams(topology, this.properties, this.clientSupplier);
 				this.kafkaStreams.setStateListener(this.stateListener);

spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -51,6 +51,7 @@
  * @author Artem Bilan
  * @author Gary Russell
  * @author Denis Washington
+ * @author Soby Chacko
  */
 @SpringJUnitConfig
 @DirtiesContext
@@ -99,6 +100,7 @@ protected StreamsBuilder createInstance() {
 		streamsBuilderFactoryBean.start();
 		StreamsBuilder streamsBuilder = streamsBuilderFactoryBean.getObject();
 		verify(streamsBuilder).build(kafkaStreamsConfiguration.asProperties());
+		assertThat(streamsBuilderFactoryBean.getTopology()).isNotNull();
 	}
 
 	@Test