feat: add extra streams

goto · Sep 30, 2024 · 9df3622 · 9df3622
1 parent 2eb0f56
commit 9df3622
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 5 deletions.
diff --git a/modules/flink/config.go b/modules/flink/config.go
@@ -23,11 +23,12 @@ type Influx struct {
 }
 
 type Config struct {
-	KubeNamespace   string `json:"kube_namespace,omitempty"`
-	Influx          Influx `json:"influx,omitempty"`
-	SinkKafkaStream string `json:"sink_kafka_stream,omitempty"`
-	PrometheusURL   string `json:"prometheus_url,omitempty"`
-	FlinkName       string `json:"flink_name,omitempty"`
+	KubeNamespace   string   `json:"kube_namespace,omitempty"`
+	Influx          Influx   `json:"influx,omitempty"`
+	SinkKafkaStream string   `json:"sink_kafka_stream,omitempty"`
+	PrometheusURL   string   `json:"prometheus_url,omitempty"`
+	FlinkName       string   `json:"flink_name,omitempty"`
+	ExtraStreams    []string `json:"extra_streams,omitempty"`
 }
 
 func readConfig(_ resource.Resource, confJSON json.RawMessage, dc driverConf) (*Config, error) {

diff --git a/modules/flink/driver.go b/modules/flink/driver.go
@@ -27,6 +27,7 @@ type Output struct {
 	SinkKafkaStream string            `json:"sink_kafka_stream,omitempty"`
 	PrometheusURL   string            `json:"prometheus_url,omitempty"`
 	FlinkName       string            `json:"flink_name,omitempty"`
+	ExtraStreams    []string          `json:"extra_streams,omitempty"`
 }
 
 func readOutputData(exr module.ExpandedResource) (*Output, error) {

diff --git a/modules/flink/driver_output.go b/modules/flink/driver_output.go
@@ -35,6 +35,7 @@ func (fd *flinkDriver) Output(ctx context.Context, exr module.ExpandedResource)
 	output.SinkKafkaStream = conf.SinkKafkaStream
 	output.PrometheusURL = conf.PrometheusURL
 	output.FlinkName = conf.FlinkName
+	output.ExtraStreams = conf.ExtraStreams
 
 	return modules.MustJSON(output), nil
 }