diff --git a/ui/src/containers/configs/NewConfig.js b/ui/src/containers/configs/NewConfig.js
index 70ee49d6..4b524bac 100644
--- a/ui/src/containers/configs/NewConfig.js
+++ b/ui/src/containers/configs/NewConfig.js
@@ -8,6 +8,11 @@ import { getConfigKindOptions } from "../../helpers/getConfigKindOptions";
 import Header from "../../components/layout/Header";
 import { addConfig } from "../../actions/configs";
 import { fetchPipelines } from "../../actions/pipelines";
+import { getStreamConnectionOptions } from "../../helpers/getStreamConnectionOptions";
+import { getStreamTopicOptions } from "../../helpers/getStreamTopicOptions";
+import { getDeserializerOptions } from "../../helpers/getDeserializerOptions";
+import { getSerializerOptions } from "../../helpers/getSerializerOptions";
+import { isStreamHoldingAvroFormat } from "../../helpers/isStreamHoldingAvroFormat";
 import "../../scss/fonts.scss";
 
 class NewConfig extends Component {
@@ -17,6 +22,7 @@ class NewConfig extends Component {
     this.state = {
       creatingConfigFailed: false,
       errorMessages: {},
+      showTopicConfig: false,
       config: {
         name: "",
         kind: "STREAM",
@@ -42,14 +48,39 @@ class NewConfig extends Component {
           },
         },
       },
+      tempStreamConfig: {
+        topicName: "",
+        topicValue: "",
+        connectionName: "",
+        connectionValue: "",
+      },
       configCreated: false,
     };
 
     this.handleCreateConfig = this.handleCreateConfig.bind(this);
     this.handleChange = this.handleChange.bind(this);
+    this.handleEventChange = this.handleEventChange.bind(this);
     this.addLabel = this.addLabel.bind(this);
     this.removeLabel = this.removeLabel.bind(this);
+    this.toggleShowTopicConfig = this.toggleShowTopicConfig.bind(this);
+    this.addStreamConfig = this.addStreamConfig.bind(this);
+    this.removeStreamConfig = this.removeStreamConfig.bind(this);
   }
+    updateTempStreamConfig(field, value) {
+      let tempStreamConfig = this.state.tempStreamConfig;
+
+      tempStreamConfig[field] = value;
+
+      this.setState({ tempStreamConfig: tempStreamConfig });
+    }
+
+    updateConnectionConfig(field, value) {
+      let stream = this.state.stream;
+
+      stream.spec.kafka[field] = value;
+
+      this.setState({ stream: stream });
+    }
 
 
   componentDidMount() {
@@ -88,11 +119,49 @@ class NewConfig extends Component {
     });
   }
 
+  handleEventChange(event) {
+    let stream = this.state.stream;
+
+    const newValue =
+          event.target.type === "checkbox"
+            ? event.target.checked
+            : event.target.value;
+
+    switch (event.target.dataset.prefix) {
+      case "stream.spec":
+        stream["spec"][event.target.name] = newValue;
+        break;
+      case "stream.spec.kafka":
+        stream["spec"]["kafka"][event.target.name] = newValue;
+        break;
+      case "stream.spec.kafka.topic":
+        stream["spec"]["kafka"]["topic"][event.target.name] = newValue;
+        break;
+      case "stream.spec.kafka.topic.config":
+        stream["spec"]["kafka"]["topic"]["config"][event.target.name] =
+          newValue;
+        break;
+      default:
+        stream[event.target.name] = newValue;
+        break;
+    }
+
+    this.setState({
+      creatingConfigFailed: false,
+      errorMessage: "",
+      stream: stream,
+    });
+  }
+
   handleChange(name, value, prefix) {
     let config = this.state.config;
     let deployment = this.state.deployment;
     let stream = this.state.stream;
 
+    if(prefix === undefined){
+        prefix = "";
+    }
+
     if(prefix === "deployment.spec"){
         deployment.spec[name] = value;
     } else{
@@ -105,10 +174,34 @@ class NewConfig extends Component {
         config.spec = stream.spec;
     }
 
+    if(prefix.includes("stream")){
+        switch (event.target.dataset.prefix) {
+          case "stream.spec":
+            stream["spec"][event.target.name] = newValue;
+            break;
+          case "stream.spec.kafka":
+            stream["spec"]["kafka"][event.target.name] = newValue;
+            break;
+          case "stream.spec.kafka.topic":
+            stream["spec"]["kafka"]["topic"][event.target.name] = newValue;
+            break;
+          case "stream.spec.kafka.topic.config":
+            stream["spec"]["kafka"]["topic"]["config"][event.target.name] =
+              newValue;
+            break;
+          default:
+            stream[event.target.name] = newValue;
+            break;
+        }
+    }
+
+
     this.setState({
       creatingConfigFailed: false,
       errorMessage: "",
       config: config,
+      deployment: deployment,
+      stream: stream,
     });
   }
 
@@ -128,15 +221,55 @@ class NewConfig extends Component {
       this.setState({ config: config });
     }
 
-        setDefaultKind(field, value) {
-          let config = this.state.config;
+    setDefaultKind(field, value) {
+      let config = this.state.config;
+
+      config.kind = value;
+
+      this.setState({ config: config });
+      return this.state.config.kind;
+    }
+
+  toggleShowTopicConfig(event) {
+    event.preventDefault();
 
-          config.kind = value;
+    this.setState({
+      showTopicConfig: !this.state.showTopicConfig,
+    });
+  }
+
+    addStreamConfig(event) {
+      event.preventDefault();
+      const tempStreamConfig = this.state.tempStreamConfig;
+      let stream = this.state.stream;
+
+      if (event.target.dataset.prefix === "stream.spec.kafka") {
+        stream.spec.kafka[tempStreamConfig.connectionName] = tempStreamConfig.connectionValue;
+        this.setState({ stream: stream, tempStreamConfig: tempStreamConfig });
+      } else if (event.target.dataset.prefix === "stream.spec.kafka.topic.config") {
+        stream.spec.kafka.topic.config[tempStreamConfig.topicName] =
+          tempStreamConfig.topicValue;
+        this.setState({ stream: stream, tempStreamConfig: tempStreamConfig });
+      }
+    }
 
-          this.setState({ config: config });
-          return this.state.config.kind;
+      removeStreamConfig(event) {
+        event.preventDefault();
+        let stream = this.state.stream;
+        const prefix = event.target.dataset.prefix;
+        const streamConfig = event.target.dataset.config;
+
+        if (prefix === "stream.spec.kafka") {
+          delete stream.spec.kafka[streamConfig];
+        } else if (prefix === "stream.spec.kafka.topic.config") {
+          delete stream.spec.kafka.topic.config[streamConfig];
         }
 
+        this.setState({ stream: stream });
+      }
+
+
+
   render() {
     if (this.state.configCreated) {
       return (
@@ -156,6 +289,32 @@ class NewConfig extends Component {
           return { value: pipeline.uuid, label: name };
         });
 
+    const stream = this.state.stream;
+
+    const topicOptions = getStreamTopicOptions();
+    const connectionOptions = getStreamConnectionOptions();
+    const deserializerOptions = getDeserializerOptions(stream);
+    const serializerOptions = getSerializerOptions(stream);
+
+    const defaultDeserializer = "io.datacater.core.serde.JsonDeserializer";
+    const defaultSerializer = "io.datacater.core.serde.JsonSerializer";
+
+    const addedTopicConfigs = Object.keys(stream.spec.kafka.topic.config);
+    const addedConnectionConfigs = Object.keys(stream.spec.kafka).filter(
+      (item) =>
+        ![
+          "bootstrap.servers",
+          "key.deserializer",
+          "value.deserializer",
+          "key.serializer",
+          "value.serializer",
+          "schema.registry.url",
+          "topic",
+        ].includes(item)
+    );
+
+    const streamHoldsAvroFormat = isStreamHoldingAvroFormat(stream);
+
     return (
       <div className="container">
         <div className="row">
@@ -279,14 +438,163 @@ class NewConfig extends Component {
                 </div>
               </div>
             </div>
-
-
-
             {[undefined, defaultKind].includes(this.state.config["kind"]) && (
               <>
                 <div className="col-12 mt-4">
-
+                  <label htmlFor="name" className="form-label">
+                    <h5 className="fw-semibold">
+                      Stream definition
+                    </h5>
+                    <a
+                      className="fs-7"
+                      href="/configs/new"
+                      onClick={this.toggleShowTopicConfig}
+                    >
+                      {!this.state.showTopicConfig && "Edit topic config"}
+                      {this.state.showTopicConfig && "Hide topic config"}
+                    </a>
+                  </label>
                 </div>
+                {this.state.showTopicConfig && (
+                  <>
+                    <div className="card mt-4">
+                      <div className="card-body">
+                        <div className="col-12 d-flex align-items-center">
+                          <h5 className="fw-semibold d-inline mb-0">
+                            Topic configuration
+                          </h5>
+                          <a
+                            className="fs-7 ms-2"
+                            href="/configs/new"
+                            onClick={this.toggleShowTopicConfig}
+                          >
+                            Hide
+                          </a>
+                        </div>
+                        <div className="col-12 mt-2">
+                          <label htmlFor="num.partitions" className="form-label">
+                            num.partitions
+                          </label>
+                          <input
+                            type="text"
+                            className="form-control"
+                            id="num.partitions"
+                            data-prefix="stream.spec.kafka.topic"
+                            name="num.partitions"
+                            onChange={this.handleEventChange}
+                            placeholder="1"
+                            value={
+                              this.state.stream.spec.kafka["topic"]["num.partitions"] || ""
+                            }
+                          />
+                        </div>
+                        <div className="col-12 mt-2">
+                          <label
+                            htmlFor="replication.factor"
+                            className="form-label"
+                          >
+                            replication.factor
+                          </label>
+                          <input
+                            type="text"
+                            className="form-control"
+                            id="replication.factor"
+                            data-prefix="stream.spec.kafka.topic"
+                            name="replication.factor"
+                            onChange={this.handleEventChange}
+                            placeholder="3"
+                            value={
+                              this.state.stream.spec.kafka["topic"][
+                                "replication.factor"
+                              ] || ""
+                            }
+                          />
+                        </div>
+                        {addedTopicConfigs.map((topicConfig) => (
+                          <div className="col-12 mt-2" key={topicConfig}>
+                            <label htmlFor={topicConfig} className="form-label">
+                              {topicConfig}
+                              <a
+                                className="ms-2 fs-7"
+                                data-config={topicConfig}
+                                data-prefix="stream.spec.kafka.topic.config"
+                                href="/configs/new"
+                                onClick={this.removeStreamConfig}
+                              >
+                                Remove
+                              </a>
+                            </label>
+                            <input
+                              type="text"
+                              className="form-control"
+                              id={topicConfig}
+                              data-prefix="stream.spec.kafka.topic.config"
+                              name={topicConfig}
+                              onChange={this.handleEventChange}
+                              value={
+                                this.state.stream.spec.kafka.topic.config[
+                                  topicConfig
+                                ] || ""
+                              }
+                            />
+                          </div>
+                        ))}
+                        <div className="col-12 mt-3">
+                          <h6 className="d-inline me-2">Add config</h6>
+                          <span className="text-muted fs-7">
+                            You can here use{" "}
+                            <a
+                              href="https://kafka.apache.org/documentation/#topicconfigs"
+                              target="_blank"
+                            >
+                              topic-level
+                            </a>{" "}
+                            configuration options.
+                          </span>
+                        </div>
+                        <div className="col-12 mt-2">
+                          <div className="row">
+                              <div className="col-md-3">
+                                <label className="form-label">Name</label>
+                                <Creatable
+                                  isSearchable
+                                  options={topicOptions}
+                                  onChange={(value) => {
+                                    this.updateTempStreamConfig("topicName", value.value);
+                                  }}
+                                />
+                              </div>
+                            <div className="col-md-3">
+                              <label className="form-label">Value</label>
+                              <input
+                                type="text"
+                                className="form-control"
+                                name="topicValue"
+                                onChange={(event) => {
+                                  this.updateTempStreamConfig(
+                                    "topicValue",
+                                    event.target.value
+                                  );
+                                }}
+                                value={this.state.tempStreamConfig.topicValue || ""}
+                              />
+                            </div>
+                            <div className="col-md-3 d-flex align-items-end">
+                              <a
+                                href="/configs/new"
+                                className="btn btn-outline-primary"
+                                data-prefix="stream.spec.kafka.topic.config"
+                                onClick={this.addStreamConfig}
+                              >
+                                Add
+                              </a>
+                            </div>
+                          </div>
+                        </div>
+                      </div>
+                    </div>
+                  </>
+                )}
               </>
             )}
 
@@ -337,7 +645,7 @@ const mapStateToProps = function (state) {
 };
 
 const mapDispatchToProps = {
-  addonfig: addConfig,
+  addConfig: addConfig,
   fetchPipelines: fetchPipelines,
 };