You can use the "github.com/jmoiron/sqlx" library to execute dynamic queries with parameters in Golang. Here's an example code snippet that demonstrates how you can achieve this:

import (
    "database/sql"
    "encoding/json"
    "fmt"

    "github.com/jmoiron/sqlx"
)

type InputParam struct {
    Key   string `json:"key"`
    Value string `json:"value"`
}

type Message struct {
    SqlTable    string       `json:"sql_table"`
    InputParams []InputParam `json:"input_params"`
}

type KafkaMessage struct {
    Message []Message `json:"message"`
}

func main() {
    db, err := sqlx.Connect("database_type", "connection_string")
    if err != nil {
        fmt.Println(err)
        return
    }

    kafkaMsg := KafkaMessage{}
    err = json.Unmarshal([]byte(jsonStr), &kafkaMsg)
    if err != nil {
        fmt.Println(err)
        return
    }

    for _, msg := range kafkaMsg.Message {
        queryParams := make(map[string]interface{})
        query := fmt.Sprintf("SELECT * FROM %s WHERE ", msg.SqlTable)

        for index, param := range msg.InputParams {
            query += fmt.Sprintf("%s = :%d", param.Key, index)
            queryParams[param.Key] = param.Value

            if index+1 != len(msg.InputParams) {
                query += " AND "
            }
        }

        rows, err := db.NamedQuery(query, queryParams)
        if err != nil {
            fmt.Println(err)
            continue
        }

        // Iterate over rows
    }
}

In this code, we first define structs to represent the Kafka message and its contents. We parse the input JSON string into a KafkaMessage object using the encoding/json package.

Next, we loop over each message in the Kafka message and build a parameterized SQL query using the query += fmt.Sprintf("%s = :%d", param.Key, index) statement. The query variable will be replaced with the correct table name and input parameters for each iteration.

We create a map of parameter values called queryParams, which is then passed to the db.NamedQuery() function along with the constructed SQL query string. This function returns an iterator over the matching rows, allowing you to iterate over the results and do whatever you need to do with them.

By using this approach, you can dynamically construct parameterized SQL queries with any number of parameters and execute them safely against your database.