PUT<\/code> method when making our HTTP Requests. As for transforming the data, Kafka already provided a method of doing this with something they call SMT (Single Message Transform).<\/p>\nAnd then finally, the last challenge we needed to overcome is being able to securely push our data through HTTPS as GridDB cloud’s endpoint is protected by SSL.<\/p>\n
Following Along<\/h2>\n
All source code for this project are available on our GitHub page.<\/p>\n
$ git clone https:\/\/github.com\/griddbnet\/Blogs.git --branch kafka_http<\/code><\/p>\nWithin that repo you will find the source code, the docker compose file, and the SSL certificates.<\/p>\n
As this entire project is dockerized, to run the project yourself, you will simply need docker installed. From there, you can run the project: docker compose up -d<\/code>. We have already included the .jar<\/code> file in the library dir so you won’t need to build the custom SMT code to push data to GridDB Cloud.<\/p>\nImplementation<\/h2>\n
To connect to push data to GridDB Cloud via the Web API, you must make an HTTP Request with a data structure that the Web API expects. If you look at the docs<\/a>, you will see that to push data into a container we need to ensure a couple of things: first we need to ensure we make a PUT<\/code> HTTP Request. Second, we need to ensure the data is set up as an array of arrays in the order of the schema. For example:<\/p>\n\n[\n [\"2025-01-16T10:25:00.253Z\", 100.5, \"normal\"],\n [\"2025-01-16T10:35:00.691Z\", 173.9, \"normal\"],\n [\"2025-01-16T10:45:00.032Z\", 173.9, null]\n]<\/code><\/pre>\n<\/div>\nIn order to get our Kafka messages to output messages like this, we will need to write a custom SMT<\/code>. Here’s an excellent article on how flexible and useful these can be: Single Message Transformations – The Swiss Army Knife of Kafka Connect<\/a>.<\/p>\nOnce we have the SMT<\/code> finished, we can set up our SSL rules and certs and then make our connectors and topics via Confluent’s UI or through JSON files.<\/p>\nSingle Message Transformations<\/h3>\n
The code to get this working is not very complicated, essentially we want to take an objject structure coming in from a typical Kafka message and transform into an array of arrays with all of the values parsed out. We will ensure that the index positions match our schema outside of the context of the SMT<\/code>.<\/p>\nAs mentioned earlier, the .jar<\/code> file is included within this project so you don’t need to do anything else, but if you would like to build it yourself or make changes, you can use mvn<\/code> to build it. Here is the full Java code (it’s also available in this repo in the smt<\/code> directory).<\/p>\n\n@Override\n public R apply(R record) {\n final Schema schema = operatingSchema(record);\n \n if (schema == null) {\n final Map<String, Object> value = requireMapOrNull(operatingValue(record), PURPOSE);\n return newRecord(record, null, value == null ? null : fieldPath.valueFrom(value));\n } else {\n final Struct value = requireStructOrNull(operatingValue(record), PURPOSE);\n fieldNames = schema.fields(); \n\n List<List<Object>> nestedArray = new ArrayList<>();\n List<Object> row = new ArrayList<>();\n for (Field f : fieldNames) {\n String fName = f.name();\n SingleFieldPath fPath = new SingleFieldPath(fName, FieldSyntaxVersion.V2);\n row.add(fPath.valueFrom(value));\n }\n nestedArray.add(row);\n \n return newRecord(record, schema, value == null ? null : nestedArray);\n }\n \n }<\/code><\/pre>\n<\/div>\nThe main method we will be using is this apply<\/code> function. We extract all of the values from the incoming messages, remove the field names, and make a new array of arrays and return that new array. That’s it! Of course there’s more to it, but this is the important bit.<\/p>\nNow that we’ve got the structure we need, let’s set up our connectors and SSL information.<\/p>\n
Docker SSL Parameters<\/h3>\n
Because GridDB Cloud’s endpoint is SSL protected, we need to ensure that our Kafka broker and HTTP Sink have the proper SSL Certs in place to securely communicate with the endpoint. If we miss any part of the process, the connection will fail with various errors, including the dreaded Handshake failed<\/code>.<\/p>\nBased on the docker-compose<\/code> file I used as the base for this project, to get SSL working, we will need to add a ton SSL environment values for our broker and kafka-connect.<\/p>\nHere are some of the values I added to the broker<\/code> in order for it to get SSL working<\/p>\n\n KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT, PLAINTEXT:PLAINTEXT, PLAINTEXT_HOST:PLAINTEXT, SSL:SSL'\n KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT:\/\/broker:29092, PLAINTEXT_HOST:\/\/localhost:9092, SSL:\/\/broker:9093'\n KAFKA_SSL_KEYSTORE_FILENAME: kafka.kafka-1.keystore.pkcs12\n KAFKA_SSL_KEYSTORE_CREDENTIALS: kafka-1_keystore_creds\n KAFKA_SSL_KEY_CREDENTIALS: kafka-1_sslkey_creds\n KAFKA_SSL_TRUSTSTORE_FILENAME: kafka.client.truststore.jks\n KAFKA_SSL_TRUSTSTORE_CREDENTIALS: kafka-1_trustore_creds\n KAFKA_SECURITY_PROTOCOL: 'SSL'\n KAFKA_SASL_MECHANISM: 'plain'\n KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: \n KAFKA_LISTENERS: 'PLAINTEXT:\/\/broker:29092, CONTROLLER:\/\/broker:29093, PLAINTEXT_HOST:\/\/0.0.0.0:9092, SSL:\/\/broker:9093'<\/code><\/pre>\n<\/div>\nOn top of adding these values, we also needed to generate these certificate files and copy them to the docker containers using a mounted volume.<\/p>\n
Generating SSL Certificates<\/h4>\n
First, let’s take a look at the .pkcs12<\/code> file, which is the SSL_KEYSTORE_FILE<\/code>. This is a file you can generate on your local working machine, to do so, I followed a guide which gave me the following instructions:<\/p>\n\n$ openssl req -new -nodes \n -x509 \n -days 365 \n -newkey rsa:2048 \n -keyout ca.key \n -out ca.crt \n\n$ openssl req -new \n -newkey rsa:2048 \n -keyout kafka-1.key \n -out kafka-1.csr \n -nodes\n\n$ openssl x509 -req \n -days 3650 \n -in kafka-1.csr \n -CA ca.crt \n -CAkey ca.key \n -CAcreateserial \n -out kafka-1.crt \n -extensions v3_req\n\n$ openssl pkcs12 -export \n -in kafka-1.crt \n -inkey afka-1.key \n -chain \n -CAfile ca.pem \n -name kafka-1 \n -out kafka-1.p12 \n -password pass:confluent\n\n$ keytool -importkeystore \n -deststorepass confluent \n -destkeystore kafka.kafka-1.keystore.pkcs12 \n -srckeystore kafka-1.p12 \n -deststoretype PKCS12 \n -srcstoretype PKCS12 \n -noprompt \n -srcstorepass confluent<\/code><\/pre>\n<\/div>\nWith that out of the way, we will also need to tell our server that the GridDB Cloud is safe by grabbing its certs and then generating some certs and including them into our broker and connect. From the GridDB Cloud web dashboard, if you click on the lock icon from the browser, you can view\/manage the SSL Certificates. From that menu, you can download the .pem<\/code> files. Alternatively, you can use the CLI: openssl s_client -showcerts -connect cloud5197.griddb.com:443<\/code>.<\/p>\nWith the output, you can save the portions that say BEGIN CERTIFICATE<\/code> to END CERTIFICATE<\/code> into a separate file. Armed with this file, you can generate a truststore file to let your server know it’s a trusted location.<\/p>\n\n$ keytool -import -trustcacerts -alias griddb-cloud-cert -file ca.pem -keystore kafka.client.truststore.jks -storepass confluent -v<\/code><\/pre>\n<\/div>\nNow we have the two key files (kafka.kafka-1.keystore.pkcs12<\/code> && kafka.client.truststore.jks<\/code>) needed for secure communication with GridDB Cloud — cool!<\/p>\nConnector Clients<\/h3>\n
This next step is where we actually tell our kafka cluster which data we want streaming to where. So in this case, we will make a test topic with a simple schema of just three values:<\/p>\n
\n{\n \"connect.name\": \"net.griddb.webapi.griddb\",\n \"connect.parameters\": {\n \"io.confluent.connect.avro.field.doc.data\": \"The string is a unicode character sequence.\",\n \"io.confluent.connect.avro.field.doc.temp\": \"The double type is a double precision (64-bit) IEEE 754 floating-point number.\",\n \"io.confluent.connect.avro.field.doc.ts\": \"The int type is a 32-bit signed integer.\",\n \"io.confluent.connect.avro.record.doc\": \"Sample schema to help you get started.\"\n },\n \"doc\": \"Sample schema to help you get started.\",\n \"fields\": [\n {\n \"doc\": \"The int type is a 32-bit signed integer.\",\n \"name\": \"ts\",\n \"type\": \"int\"\n },\n {\n \"doc\": \"The double type is a double precision (64-bit) IEEE 754 floating-point number.\",\n \"name\": \"temp\",\n \"type\": \"double\"\n },\n {\n \"doc\": \"The string is a unicode character sequence.\",\n \"name\": \"data\",\n \"type\": \"double\"\n }\n ],\n \"name\": \"griddb\",\n \"namespace\": \"net.griddb.webapi\",\n \"type\": \"record\"\n}<\/code><\/pre>\n<\/div>\nBefore we try pushing our data to GridDB Cloud, we will need to create our container inside of our DB. You can use the Dashboard or simply send a CURL request using Postman or the CLI to create the container to match that schema. For me, I’m calling it kafka<\/code>. In this case, I’m not going to make a Time Series container and will settle for a Collection container for educational purposes.<\/p>\nWe will then make a source connector provided by Confluent to generate mock data in the style of that schema.<\/p>\n
Once you have it set up, it looks like this in the dashboard:<\/p>\n
![\"\"](\"https:\/\/griddb.net\/wp-content\/uploads\/2025\/04\/topic-messages.png\")
<\/a><\/p>\nNext, we make a connector for the HTTP Sink which takes that source connector’s mock data and streams it out to the HTTP we set it to (hint: it’s GridDB Cloud!). But as the data moves through from the source to the sink, we will of course apply our SMT<\/code> to change the data into an array of arrays to push to GridDB Cloud. And if we configured our SSL correctly, we should see our data inside of our GridDB Cloud container.<\/p>\nConnector Client Values and Rules<\/h4>\n
To send the connectors to your Kafka cluster, you can either manually enter in the values using the Kafka Control Center, which provides a nice UI for editing connectors, or simply take the .json<\/code> files included with this repo and pushing them using CURL.<\/p>\nHere are the values for the datagen which creates mock data for our GridDB Cloud to ingest:<\/p>\n
\n{\n \"name\": \"web_api_datagen\",\n \"config\": {\n \"connector.class\": \"io.confluent.kafka.connect.datagen.DatagenConnector\",\n \"kafka.topic\": \"griddb_test\",\n \"schema.string\": \"{ \"connect.name\": \"net.griddb.webapi.griddb\", \"connect.parameters\": { \"io.confluent.connect.avro.field.doc.data\": \"The string is a unicode character sequence.\", \"io.confluent.connect.avro.field.doc.temp\": \"The double type is a double precision (64-bit) IEEE 754 floating-point number.\", \"io.confluent.connect.avro.field.doc.ts\": \"The int type is a 32-bit signed integer.\", \"io.confluent.connect.avro.record.doc\": \"Sample schema to help you get started.\" }, \"doc\": \"Sample schema to help you get started.\", \"fields\": [ { \"doc\": \"The int type is a 32-bit signed integer.\", \"name\": \"ts\", \"type\": \"int\" }, { \"doc\": \"The double type is a double precision (64-bit) IEEE 754 floating-point number.\", \"name\": \"temp\", \"type\": \"double\" }, { \"doc\": \"The string is a unicode character sequence.\", \"name\": \"data\", \"type\": \"double\" } ], \"name\": \"griddb\", \"namespace\": \"net.griddb.webapi\", \"type\": \"record\" }\"\n }\n}<\/code><\/pre>\n<\/div>\nIt is messy, but that’s because the schema string includes the raw string of the schema I shared earlier (up above).<\/p>\n
And here are the values of the HTTP Sink itself:<\/p>\n
\n{\n \"name\": \"griddb_web_api_sink\",\n \"config\": {\n \"connector.class\": \"io.confluent.connect.http.HttpSinkConnector\",\n \"transforms\": \"nestedList\",\n \"topics\": \"griddb\",\n \"transforms.nestedList.type\": \"net.griddb.GridDBWebAPITransform$Value\",\n \"transforms.nestedList.fields\": \"ts\",\n \"http.api.url\": \"https:\/\/cloud5197.griddb.com\/griddb\/v2\/gs_clustermfcloud97\/dbs\/ZUlQ8\/containers\/kafka\/rows\",\n \"request.method\": \"put\",\n \"headers\": \"Content-Type: application\/json\",\n \"auth.type\": \"basic\",\n \"connection.user\": \"user\",\n \"connection.password\": \"password\",\n \"https.ssl.key.password\": \"confluent\",\n \"https.ssl.keystore.key\": \"\",\n \"https.ssl.keystore.location\": \"\/etc\/kafka\/secrets\/kafka.kafka-1.keystore.pkcs12\",\n \"https.ssl.keystore.password\": \"confluent\",\n \"https.ssl.truststore.location\": \"\/etc\/kafka\/secrets\/kafka.client.truststore.jks\",\n \"https.ssl.truststore.password\": \"confluent\",\n \"https.ssl.enabled.protocols\": \"\",\n \"https.ssl.keystore.type\": \"PKCS12\",\n \"https.ssl.protocol\": \"TLSv1.2\",\n \"https.ssl.truststore.type\": \"JKS\",\n \"reporter.result.topic.replication.factor\": \"1\",\n \"reporter.error.topic.replication.factor\": \"1\",\n \"reporter.bootstrap.servers\": \"broker:29092\"\n }\n}<\/code><\/pre>\n<\/div>\nSome important values here: of course the SSL values and certs, as well as the URL as this contains the container name (kafka in our case). We also have our BASIC AUTHENICATION values in here as well as our SMT<\/code>. All of this information is crucial to ensure that our Kafka cluster streams our mock data to the proper place with zero errors.<\/p>\nYou can push these connectors using HTTP Requests:<\/p>\n
\n$ #!\/bin\/sh\n\ncurl -s \n -X \"POST\" \"http:\/\/localhost:8083\/connectors\/\" \n -H \"Content-Type: application\/json\" \n -d '{\n \"name\": \"griddb_web_api_sink\",\n \"config\": {\n \"connector.class\": \"io.confluent.connect.http.HttpSinkConnector\",\n \"transforms\": \"nestedList\",\n \"topics\": \"griddb_test\",\n \"transforms.nestedList.type\": \"net.griddb.GridDBWebAPITransform$Value\",\n \"transforms.nestedList.fields\": \"ts\",\n \"http.api.url\": \"https:\/\/cloud5197.griddb.com\/griddb\/v2\/gs_clustermfcloud97\/dbs\/ZUlQ8\/containers\/kafka\/rows\",\n \"request.method\": \"put\",\n \"headers\": \"Content-Type: application\/json\",\n \"auth.type\": \"basic\",\n \"connection.user\": \"user\",\n \"connection.password\": \"password\",\n \"https.ssl.key.password\": \"confluent\",\n \"https.ssl.keystore.key\": \"\",\n \"https.ssl.keystore.location\": \"\/etc\/kafka\/secrets\/kafka.kafka-1.keystore.pkcs12\",\n \"https.ssl.keystore.password\": \"confluent\",\n \"https.ssl.truststore.location\": \"\/etc\/kafka\/secrets\/kafka.client.truststore.jks\",\n \"https.ssl.truststore.password\": \"confluent\",\n \"https.ssl.enabled.protocols\": \"\",\n \"https.ssl.keystore.type\": \"PKCS12\",\n \"https.ssl.protocol\": \"TLSv1.2\",\n \"https.ssl.truststore.type\": \"JKS\",\n \"reporter.result.topic.replication.factor\": \"1\",\n \"reporter.error.topic.replication.factor\": \"1\",\n \"reporter.bootstrap.servers\": \"broker:29092\"\n }\n }'<\/code><\/pre>\n<\/div>\nAnd then the same thing for the source connector. The main thing to take away from this section is the values you need to enter to successfully push your data from Kafka to GridDB Cloud. For example, you can see in the transforms section that we are using the SMT<\/code> we wrote and built earlier.<\/p>\nResults<\/h2>\n
First, let’s take a look at our logs to see if our data is going through<\/p>\n
\n$ docker logs -f connect<\/code><\/pre>\n<\/div>\nHere you should see some sort of output. You can also check your Control Center and ensure that the GridDB Web API Sink doesn’t have any errors. For me, this is what it looks like:<\/p>\n
![\"\"](\"https:\/\/griddb.net\/wp-content\/uploads\/2025\/04\/no-error-control-center.png\")
<\/a><\/p>\n![\"\"](\"https:\/\/griddb.net\/wp-content\/uploads\/2025\/04\/griddb-query.png\")
<\/a><\/p>\n