Using the Streams for Apache Kafka Bridge Red Hat Streams for Apache Kafka 2.7 Use the Streams for Apache Kafka Bridge to connect with a Kafka cluster

Commands:
Content-Type: application/vnd.kafka.v2+json
{ "name": "my-consumer", "format": "binary", 1 # }
{ "records": [ { "key": "my-key", "value": "ZWR3YXJkdGhldGhyZWVsZWdnZWRjYXQ=" }, ] }
curl -X POST http://localhost:8080/topics/my-topic -H 'content-type: application/vnd.kafka.json.v2+json' -d '{ "records": [ { "key": "my-key", "value": "sales-lead-0001", "partition": 2, "headers": [ { "key": "key1", "value": "QXBhY2hlIEthZmthIGlzIHRoZSBib21iIQ==" 1 } ] } ] }'
Accept: application/vnd.kafka. EMBEDDED-DATA-FORMAT .v2+json
Accept: application/vnd.kafka.json.v2+json
http.cors.enabled=true http.cors.allowedOrigins=http://my-web-application.io http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH
Origin: http://my-web-application.io
curl -v -X GET HTTP-BRIDGE-ADDRESS /consumers/my-group/instances/my-consumer/records -H 'Origin: http://my-web-application.io' -H 'content-type: application/vnd.kafka.v2+json'
HTTP/1.1 200 OK Access-Control-Allow-Origin: * 1
OPTIONS /my-group/instances/my-consumer/subscription HTTP/1.1 Origin: http://my-web-application.io Access-Control-Request-Method: POST 1 Access-Control-Request-Headers: Content-Type 2
curl -v -X OPTIONS -H 'Origin: http://my-web-application.io' -H 'Access-Control-Request-Method: POST' -H 'content-type: application/vnd.kafka.v2+json'
HTTP/1.1 200 OK Access-Control-Allow-Origin: http://my-web-application.io Access-Control-Allow-Methods: GET,POST,PUT,DELETE,OPTIONS,PATCH Access-Control-Allow-Headers: content-type
curl -v -X POST HTTP-BRIDGE-ADDRESS /topics/bridge-topic -H 'Origin: http://my-web-application.io' -H 'content-type: application/vnd.kafka.v2+json'
HTTP/1.1 200 OK Access-Control-Allow-Origin: http://my-web-application.io
logger.healthy.name = http.openapi.operation.healthy logger.healthy.level = WARN logger.ready.name = http.openapi.operation.ready logger.ready.level = WARN
logger. <operation_id> .name = http.openapi.operation. <operation_id> logger. <operation_id>_level = _<LOG_LEVEL>
http.host=0.0.0.0 http.port=8080
./bin/kafka_bridge_run.sh --config-file= <path> /application.properties
HTTP-Kafka Bridge started and listening on port 8080 HTTP-Kafka Bridge bootstrap servers localhost:9092
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic bridge-quickstart-topic --partitions 3 --replication-factor 1
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic bridge-quickstart-topic
curl -X POST http://localhost:8080/topics/bridge-quickstart-topic -H 'content-type: application/vnd.kafka.json.v2+json' -d '{ "records": [ { "key": "my-key", "value": "sales-lead-0001" }, { "value": "sales-lead-0002", "partition": 2 }, { "value": "sales-lead-0003" } ] }'
# { "offsets":[ { "partition":0, "offset":0 }, { "partition":2, "offset":0 }, { "partition":0, "offset":1 } ] }
curl -X GET http://localhost:8080/topics
__strimzi_store_topic, __strimzi-topic-operator-kstreams-topic-store-changelog, bridge-quickstart-topic, my-topic
curl -X GET http://localhost:8080/topics/bridge-quickstart-topic
{ "name": "bridge-quickstart-topic", "configs": { "compression.type": "producer", "leader.replication.throttled.replicas": "", "min.insync.replicas": "1", "message.downconversion.enable": "true", "segment.jitter.ms": "0", "cleanup.policy": "delete", "flush.ms": "9223372036854775807", "follower.replication.throttled.replicas": "", "segment.bytes": "1073741824", "retention.ms": "604800000", "flush.messages": "9223372036854775807", "message.format.version": "2.8-IV1", "max.compaction.lag.ms": "9223372036854775807", "file.delete.delay.ms": "60000", "max.message.bytes": "1048588", "min.compaction.lag.ms": "0", "message.timestamp.type": "CreateTime", "preallocate": "false", "index.interval.bytes": "4096", "min.cleanable.dirty.ratio": "0.5", "unclean.leader.election.enable": "false", "retention.bytes": "-1", "delete.retention.ms": "86400000", "segment.ms": "604800000", "message.timestamp.difference.max.ms": "9223372036854775807", "segment.index.bytes": "10485760" }, "partitions": [ { "partition": 0, "leader": 0, "replicas": [ { "broker": 0, "leader": true, "in_sync": true }, { "broker": 1, "leader": false, "in_sync": true }, { "broker": 2, "leader": false, "in_sync": true } ] }, { "partition": 1, "leader": 2, "replicas": [ { "broker": 2, "leader": true, "in_sync": true }, { "broker": 0, "leader": false, "in_sync": true }, { "broker": 1, "leader": false, "in_sync": true } ] }, { "partition": 2, "leader": 1, "replicas": [ { "broker": 1, "leader": true, "in_sync": true }, { "broker": 2, "leader": false, "in_sync": true }, { "broker": 0, "leader": false, "in_sync": true } ] } ] }
curl -X GET http://localhost:8080/topics/bridge-quickstart-topic/partitions
[ { "partition": 0, "leader": 0, "replicas": [ { "broker": 0, "leader": true, "in_sync": true }, { "broker": 1, "leader": false, "in_sync": true }, { "broker": 2, "leader": false, "in_sync": true } ] }, { "partition": 1, "leader": 2, "replicas": [ { "broker": 2, "leader": true, "in_sync": true }, { "broker": 0, "leader": false, "in_sync": true }, { "broker": 1, "leader": false, "in_sync": true } ] }, { "partition": 2, "leader": 1, "replicas": [ { "broker": 1, "leader": true, "in_sync": true }, { "broker": 2, "leader": false, "in_sync": true }, { "broker": 0, "leader": false, "in_sync": true } ] } ]
curl -X GET http://localhost:8080/topics/bridge-quickstart-topic/partitions/0
{ "partition": 0, "leader": 0, "replicas": [ { "broker": 0, "leader": true, "in_sync": true }, { "broker": 1, "leader": false, "in_sync": true }, { "broker": 2, "leader": false, "in_sync": true } ] }
curl -X GET http://localhost:8080/topics/bridge-quickstart-topic/partitions/0/offsets
beginning_offset: 0
end_offset: 1
curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group -H 'content-type: application/vnd.kafka.v2+json' -d '{ "name": "bridge-quickstart-consumer", "auto.offset.reset": "earliest", "format": "json", "enable.auto.commit": false, "fetch.min.bytes": 512, "consumer.request.timeout.ms": 30000 }'
# { "instance_id": "bridge-quickstart-consumer", "base_uri":"http:// <bridge_id> -bridge-service:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer" }
curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/subscription -H 'content-type: application/vnd.kafka.v2+json' -d '{ "topics": [ "bridge-quickstart-topic" ] }'
curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records -H 'accept: application/vnd.kafka.json.v2+json'
HTTP/1.1 200 OK content-type: application/vnd.kafka.json.v2+json # [ { "topic":"bridge-quickstart-topic", "key":"my-key", "value":"sales-lead-0001", "partition":0, "offset":0 }, { "topic":"bridge-quickstart-topic", "key":null, "value":"sales-lead-0003", "partition":0, "offset":1 }, #
curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/offsets
curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions -H 'content-type: application/vnd.kafka.v2+json' -d '{ "offsets": [ { "topic": "bridge-quickstart-topic", "partition": 0, "offset": 2 } ] }'
curl -X GET http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/records -H 'accept: application/vnd.kafka.json.v2+json'
curl -X POST http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer/positions/end -H 'content-type: application/vnd.kafka.v2+json' -d '{ "partitions": [ { "topic": "bridge-quickstart-topic", "partition": 0 } ] }'
curl -X DELETE http://localhost:8080/consumers/bridge-quickstart-consumer-group/instances/bridge-quickstart-consumer
bridge.id=my-bridge http.host=0.0.0.0 http.port=8080 1 http.cors.enabled=true 2 http.cors.allowedOrigins=https://strimzi.io 3 http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH 4
KAFKA_BRIDGE_METRICS_ENABLED=true
./bin/kafka_bridge_run.sh --config-file=<path>/application.properties
bridge.tracing=opentelemetry 1
OTEL_SERVICE_NAME=my-tracing-service 1 OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 2
./bin/kafka_bridge_run.sh --config-file= <path> /application.properties
OTEL_SERVICE_NAME=my-tracing-service OTEL_TRACES_EXPORTER=zipkin 1 OTEL_EXPORTER_ZIPKIN_ENDPOINT=http://localhost:9411/api/v2/spans 2
bridge_version: 0.16.0
name: consumer1
format: binary
auto.offset.reset: earliest
enable.auto.commit: no
fetch.min.bytes: 512
consumer.request.timeout.ms: 30000
isolation.level: read_committed
instance_id: consumer1
base_uri: http://localhost:8080/consumers/my-group/instances/consumer1
error_code: 409
message: A consumer instance with the specified name already exists in the Kafka Bridge.
error_code: 422
message: One or more consumer configuration options have invalid values.
error_code: 404
message: The specified consumer instance was not found.
{ "partitions" : [ { "topic" : "topic", "partition" : 0 }, { "topic" : "topic", "partition" : 1 } ] }
error_code: 404
message: The specified consumer instance was not found.
error_code: 409
message: Subscriptions to topics, partitions, and patterns are mutually exclusive.
{ "offsets" : [ { "topic" : "topic", "partition" : 0, "offset" : 15 }, { "topic" : "topic", "partition" : 1, "offset" : 42 } ] }
error_code: 404
message: The specified consumer instance was not found.
{ "offsets" : [ { "topic" : "topic", "partition" : 0, "offset" : 15 }, { "topic" : "topic", "partition" : 1, "offset" : 42 } ] }
error_code: 404
message: The specified consumer instance was not found.
{ "partitions" : [ { "topic" : "topic", "partition" : 0 }, { "topic" : "topic", "partition" : 1 } ] }
error_code: 404
message: The specified consumer instance was not found.
{ "partitions" : [ { "topic" : "topic", "partition" : 0 }, { "topic" : "topic", "partition" : 1 } ] }
error_code: 404
message: The specified consumer instance was not found.
[ { "topic" : "topic", "key" : "key1", "value" : { "foo" : "bar" }, "partition" : 0, "offset" : 2 }, { "topic" : "topic", "key" : "key2", "value" : [ "foo2", "bar2" ], "partition" : 1, "offset" : 3 } ]
topic: test
key: a2V5
value: Y29uZmx1ZW50
partition: 1
offset: 100
topic: test
key: a2V5
value: a2Fma2E=
partition: 2
offset: 101
error_code: 404
message: The specified consumer instance was not found.
error_code: 406
message: The `format` used in the consumer creation request does not match the embedded format in the Accept header of this request.
error_code: 422
message: Response exceeds the maximum number of bytes the consumer can receive
{ "topics" : [ "topic1", "topic2" ] }
error_code: 404
message: The specified consumer instance was not found.
error_code: 409
message: Subscriptions to topics, partitions, and patterns are mutually exclusive.
error_code: 422
message: A list (of Topics type) or a topic_pattern must be specified.
{ "topics" : [ "my-topic1", "my-topic2" ], "partitions" : [ { "my-topic1" : [ 1, 2, 3 ] }, { "my-topic2" : [ 1 ] } ] }
error_code: 404
message: The specified consumer instance was not found.
error_code: 404
message: The specified consumer instance was not found.
topic1, topic2
{ "records" : [ { "key" : "key1", "value" : "value1" }, { "value" : "value2", "partition" : 1 }, { "value" : "value3" } ] }
{ "offsets" : [ { "partition" : 2, "offset" : 0 }, { "partition" : 1, "offset" : 1 }, { "partition" : 2, "offset" : 2 } ] }
error_code: 404
message: The specified topic was not found.
error_code: 422
message: The record list contains invalid records.
{ "name" : "topic", "offset" : 2, "configs" : { "cleanup.policy" : "compact" }, "partitions" : [ { "partition" : 1, "leader" : 1, "replicas" : [ { "broker" : 1, "leader" : true, "in_sync" : true }, { "broker" : 2, "leader" : false, "in_sync" : true } ] }, { "partition" : 2, "leader" : 2, "replicas" : [ { "broker" : 1, "leader" : false, "in_sync" : true }, { "broker" : 2, "leader" : true, "in_sync" : true } ] } ] }
[ { "partition" : 1, "leader" : 1, "replicas" : [ { "broker" : 1, "leader" : true, "in_sync" : true }, { "broker" : 2, "leader" : false, "in_sync" : true } ] }, { "partition" : 2, "leader" : 2, "replicas" : [ { "broker" : 1, "leader" : false, "in_sync" : true }, { "broker" : 2, "leader" : true, "in_sync" : true } ] } ]
error_code: 404
message: The specified topic was not found.
{ "records" : [ { "key" : "key1", "value" : "value1" }, { "value" : "value2" } ] }
{ "offsets" : [ { "partition" : 2, "offset" : 0 }, { "partition" : 1, "offset" : 1 }, { "partition" : 2, "offset" : 2 } ] }
error_code: 404
message: The specified topic partition was not found.
error_code: 422
message: The record is not valid.
{ "partition" : 1, "leader" : 1, "replicas" : [ { "broker" : 1, "leader" : true, "in_sync" : true }, { "broker" : 2, "leader" : false, "in_sync" : true } ] }
error_code: 404
message: The specified topic partition was not found.
beginning_offset: 10
end_offset: 50
error_code: 404
message: The specified topic partition was not found.
dnf install <package_name>
dnf install <path_to_download_package>