Kafka Source outputs as key_prefix in GCP sink

[zedgepetkepau]

Question

Hi,
Looking at documentation it looks like its possible to use kafka source output keys on transform and sink resources.
My goal is to offload kafka messages to GCS(google cloud storage). And file location should be key_prefix = "test/{{ .topic }}/dt=%F/hr=%H/"
which will translate to test/events.my.topic.test/dt=2025-10-21/hr=10/uid.log.gz

but when i try to do so, im not able to get topic name.
in transforms im able to add to event and check if it works:
.partition_topic = .topic
but in sink key_prefix = "test/{{ .topic }}/dt=%F/hr=%H/" does not work it just resolves to test//dt=2025-10-21/hr=10/uid.log.gz

So how correctly i should access source output? im using helm deployment version = "0.45.0"

Vector Config

   customConfig = {
        acknowledgements = {
          enabled = true
        }
        data_dir = "/vector-data-dir"
        sources = {
          kafka_in = {
            type              = "kafka"
            bootstrap_servers = "10.10.0.98:9094"
            group_id          = "vector_offloader"                 
            topics            = ["^events\\.my\\.topic\\.test$"] 
            auto_offset_reset = "earliest"                             
            tls = {
              enabled         = true
              ca_file         = "/vector/tls/ca.crt"   
              crt_file        = "/vector/tls/user.crt" 
              key_file        = "/vector/tls/user.key" 
              verify_hostname = false                  
            }

            librdkafka_options = {
              "security.protocol"        = "ssl"                 
              "ssl.ca.location"          = "/vector/tls/ca.crt"   
              "ssl.certificate.location" = "/vector/tls/user.crt"
              "ssl.key.location"         = "/vector/tls/user.key"
              "session.timeout.ms"       = "30000"                
              "heartbeat.interval.ms"    = "10000"                
            }
          }
        }

        sinks = {
          console = {
            type   = "console"
            inputs = ["kafka_in"]
            encoding = {
              codec = "text"
            }
          }
          gcs_out = {
            type   = "gcp_cloud_storage"
            inputs = ["kafka_in"]
            bucket = "bucket_name"
            encoding = {
              codec = "text"
            }
            compression = "gzip"
            batch = {
              max_events   = 1000000 
              timeout_secs = 60
            }
            key_prefix           = "test/{{ .topic }}/dt=%F/hr=%H/"
            filename_time_format = "%s"
            filename_append_uuid = true
          }
        }
      }

Vector Logs

no errors

[zedgepetkepau]

Found problem. Problem was HCL, helm and vector template syntax interprets variable.
solution:

key_prefix           = "test/{{ `{{ topic }}` }}/dt=%F/hr=%H/"

[pront]

Thank you for taking the time to share the fix with the community!