Compare commits
	
		
			3 Commits
		
	
	
		
	
	| Author | SHA1 | Date | |
|---|---|---|---|
| bac736deef | |||
| b89760a2cb | |||
| d830f70157 | 
							
								
								
									
										12
									
								
								go.mod
									
									
									
									
									
								
							
							
						
						
									
										12
									
								
								go.mod
									
									
									
									
									
								
							| @@ -7,17 +7,17 @@ toolchain go1.24.3 | ||||
| require ( | ||||
| 	github.com/google/uuid v1.6.0 | ||||
| 	github.com/stretchr/testify v1.11.1 | ||||
| 	github.com/twmb/franz-go v1.19.5 | ||||
| 	github.com/twmb/franz-go/pkg/kadm v1.16.1 | ||||
| 	github.com/twmb/franz-go/pkg/kmsg v1.11.2 | ||||
| 	github.com/twmb/franz-go v1.20.2 | ||||
| 	github.com/twmb/franz-go/pkg/kadm v1.17.1 | ||||
| 	github.com/twmb/franz-go/pkg/kmsg v1.12.0 | ||||
| 	go.opentelemetry.io/otel v1.38.0 | ||||
| 	go.unistack.org/micro/v3 v3.11.45 | ||||
| 	go.unistack.org/micro/v3 v3.11.48 | ||||
| ) | ||||
|  | ||||
| require ( | ||||
| 	github.com/ash3in/uuidv8 v1.2.0 // indirect | ||||
| 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect | ||||
| 	github.com/klauspost/compress v1.18.0 // indirect | ||||
| 	github.com/klauspost/compress v1.18.1 // indirect | ||||
| 	github.com/kr/pretty v0.3.1 // indirect | ||||
| 	github.com/matoous/go-nanoid v1.5.1 // indirect | ||||
| 	github.com/pierrec/lz4/v4 v4.1.22 // indirect | ||||
| @@ -25,7 +25,7 @@ require ( | ||||
| 	go.unistack.org/micro-proto/v3 v3.4.1 // indirect | ||||
| 	golang.org/x/crypto v0.43.0 // indirect | ||||
| 	golang.org/x/sys v0.37.0 // indirect | ||||
| 	google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff // indirect | ||||
| 	google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8 // indirect | ||||
| 	google.golang.org/grpc v1.76.0 // indirect | ||||
| 	google.golang.org/protobuf v1.36.10 // indirect | ||||
| 	gopkg.in/yaml.v3 v3.0.1 // indirect | ||||
|   | ||||
							
								
								
									
										12
									
								
								go.sum
									
									
									
									
									
								
							
							
						
						
									
										12
									
								
								go.sum
									
									
									
									
									
								
							| @@ -13,6 +13,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= | ||||
| github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= | ||||
| github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= | ||||
| github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= | ||||
| github.com/klauspost/compress v1.18.1 h1:bcSGx7UbpBqMChDtsF28Lw6v/G94LPrrbMbdC3JH2co= | ||||
| github.com/klauspost/compress v1.18.1/go.mod h1:ZQFFVG+MdnR0P+l6wpXgIL4NTtwiKIdBnrBd8Nrxr+0= | ||||
| github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= | ||||
| github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= | ||||
| github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= | ||||
| @@ -31,16 +33,24 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu | ||||
| github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= | ||||
| github.com/twmb/franz-go v1.19.5 h1:W7+o8D0RsQsedqib71OVlLeZ0zI6CbFra7yTYhZTs5Y= | ||||
| github.com/twmb/franz-go v1.19.5/go.mod h1:4kFJ5tmbbl7asgwAGVuyG1ZMx0NNpYk7EqflvWfPCpM= | ||||
| github.com/twmb/franz-go v1.20.2 h1:CiwhyKZHW6vqSHJkh+RTxFAJkio0jBjM/JQhx/HZ72A= | ||||
| github.com/twmb/franz-go v1.20.2/go.mod h1:YCnepDd4gl6vdzG03I5Wa57RnCTIC6DVEyMpDX/J8UA= | ||||
| github.com/twmb/franz-go/pkg/kadm v1.16.1 h1:IEkrhTljgLHJ0/hT/InhXGjPdmWfFvxp7o/MR7vJ8cw= | ||||
| github.com/twmb/franz-go/pkg/kadm v1.16.1/go.mod h1:Ue/ye1cc9ipsQFg7udFbbGiFNzQMqiH73fGC2y0rwyc= | ||||
| github.com/twmb/franz-go/pkg/kadm v1.17.1 h1:Bt02Y/RLgnFO2NP2HVP1kd2TFtGRiJZx+fSArjZDtpw= | ||||
| github.com/twmb/franz-go/pkg/kadm v1.17.1/go.mod h1:s4duQmrDbloVW9QTMXhs6mViTepze7JLG43xwPcAeTg= | ||||
| github.com/twmb/franz-go/pkg/kmsg v1.11.2 h1:hIw75FpwcAjgeyfIGFqivAvwC5uNIOWRGvQgZhH4mhg= | ||||
| github.com/twmb/franz-go/pkg/kmsg v1.11.2/go.mod h1:CFfkkLysDNmukPYhGzuUcDtf46gQSqCZHMW1T4Z+wDE= | ||||
| github.com/twmb/franz-go/pkg/kmsg v1.12.0 h1:CbatD7ers1KzDNgJqPbKOq0Bz/WLBdsTH75wgzeVaPc= | ||||
| github.com/twmb/franz-go/pkg/kmsg v1.12.0/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= | ||||
| go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8= | ||||
| go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM= | ||||
| go.unistack.org/micro-proto/v3 v3.4.1 h1:UTjLSRz2YZuaHk9iSlVqqsA50JQNAEK2ZFboGqtEa9Q= | ||||
| go.unistack.org/micro-proto/v3 v3.4.1/go.mod h1:okx/cnOhzuCX0ggl/vToatbCupi0O44diiiLLsZ93Zo= | ||||
| go.unistack.org/micro/v3 v3.11.45 h1:fjTLZYWgsVf9FIMZBxOg8ios2/tmyimnjZrsrxEUeXU= | ||||
| go.unistack.org/micro/v3 v3.11.45/go.mod h1:fDQ8Mu9wubaFP0L8hNQlpzHiEnWN0wbOlawN9HYo0N4= | ||||
| go.unistack.org/micro/v3 v3.11.48 h1:lHJYSHU2z1TTcuswItGwG7cZXN6n04EFqY7lk/0gA7w= | ||||
| go.unistack.org/micro/v3 v3.11.48/go.mod h1:fDQ8Mu9wubaFP0L8hNQlpzHiEnWN0wbOlawN9HYo0N4= | ||||
| golang.org/x/crypto v0.43.0 h1:dduJYIi3A3KOfdGOHX8AVZ/jGiyPa3IbBozJ5kNuE04= | ||||
| golang.org/x/crypto v0.43.0/go.mod h1:BFbav4mRNlXJL4wNeejLpWxB7wMbc79PdRGhWKncxR0= | ||||
| golang.org/x/net v0.45.0 h1:RLBg5JKixCy82FtLJpeNlVM0nrSqpCRYzVU1n8kj0tM= | ||||
| @@ -51,6 +61,8 @@ golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k= | ||||
| golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM= | ||||
| google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff h1:A90eA31Wq6HOMIQlLfzFwzqGKBTuaVztYu/g8sn+8Zc= | ||||
| google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= | ||||
| google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8 h1:M1rk8KBnUsBDg1oPGHNCxG4vc1f49epmTO7xscSajMk= | ||||
| google.golang.org/genproto/googleapis/rpc v0.0.0-20251022142026-3a174f9686a8/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= | ||||
| google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A= | ||||
| google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c= | ||||
| google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= | ||||
|   | ||||
							
								
								
									
										32
									
								
								meter.go
									
									
									
									
									
								
							
							
						
						
									
										32
									
								
								meter.go
									
									
									
									
									
								
							| @@ -32,27 +32,27 @@ var ( | ||||
| ) | ||||
|  | ||||
| const ( | ||||
| 	metricBrokerConnects    = "broker_connects_total" | ||||
| 	metricBrokerDisconnects = "broker_disconnects_total" | ||||
| 	metricBrokerConnects    = "micro_broker_connects_total" | ||||
| 	metricBrokerDisconnects = "micro_broker_disconnects_total" | ||||
|  | ||||
| 	metricBrokerWriteErrors        = "broker_write_errors_total" | ||||
| 	metricBrokerWriteBytes         = "broker_write_bytes_total" | ||||
| 	metricBrokerWriteWaitLatencies = "broker_write_wait_latencies" | ||||
| 	metricBrokerWriteLatencies     = "broker_write_latencies" | ||||
| 	metricBrokerWriteErrors        = "micro_broker_write_errors_total" | ||||
| 	metricBrokerWriteBytes         = "micro_broker_write_bytes_total" | ||||
| 	metricBrokerWriteWaitLatencies = "micro_broker_write_wait_latencies" | ||||
| 	metricBrokerWriteLatencies     = "micro_broker_write_latencies" | ||||
|  | ||||
| 	metricBrokerReadErrors        = "broker_read_errors_total" | ||||
| 	metricBrokerReadBytes         = "broker_read_bytes_total" | ||||
| 	metricBrokerReadWaitLatencies = "broker_read_wait_latencies" | ||||
| 	metricBrokerReadLatencies     = "broker_read_latencies" | ||||
| 	metricBrokerReadErrors        = "micro_broker_read_errors_total" | ||||
| 	metricBrokerReadBytes         = "micro_broker_read_bytes_total" | ||||
| 	metricBrokerReadWaitLatencies = "micro_broker_read_wait_latencies" | ||||
| 	metricBrokerReadLatencies     = "micro_broker_read_latencies" | ||||
|  | ||||
| 	metricBrokerThrottleLatencies = "broker_throttle_latencies" | ||||
| 	metricBrokerThrottleLatencies = "micro_broker_throttle_latencies" | ||||
|  | ||||
| 	metricBrokerProduceBytesCompressed   = "broker_produce_bytes_compressed_total" | ||||
| 	metricBrokerProduceBytesUncompressed = "broker_produce_bytes_uncompressed_total" | ||||
| 	metricBrokerFetchBytesCompressed     = "broker_consume_bytes_compressed_total" | ||||
| 	metricBrokerFetchBytesUncompressed   = "broker_consume_bytes_uncompressed_total" | ||||
| 	metricBrokerProduceBytesCompressed   = "micro_broker_produce_bytes_compressed_total" | ||||
| 	metricBrokerProduceBytesUncompressed = "micro_broker_produce_bytes_uncompressed_total" | ||||
| 	metricBrokerFetchBytesCompressed     = "micro_broker_consume_bytes_compressed_total" | ||||
| 	metricBrokerFetchBytesUncompressed   = "micro_broker_consume_bytes_uncompressed_total" | ||||
|  | ||||
| 	metricBrokerGroupErrors = "broker_group_errors_total" | ||||
| 	metricBrokerGroupErrors = "micro_broker_group_errors_total" | ||||
|  | ||||
| 	labelNode    = "node_id" | ||||
| 	labelSuccess = "success" | ||||
|   | ||||
| @@ -3,6 +3,7 @@ package kgo | ||||
| import ( | ||||
| 	"context" | ||||
| 	"fmt" | ||||
| 	"strconv" | ||||
| 	"sync" | ||||
| 	"sync/atomic" | ||||
| 	"time" | ||||
| @@ -238,6 +239,11 @@ func (pc *consumer) consume() { | ||||
| 				for _, hdr := range record.Headers { | ||||
| 					p.msg.Header.Set(hdr.Key, string(hdr.Value)) | ||||
| 				} | ||||
| 				p.msg.Header.Set("Micro-Offset", strconv.FormatInt(record.Offset, 10)) | ||||
| 				p.msg.Header.Set("Micro-Partition", strconv.FormatInt(int64(record.Partition), 10)) | ||||
| 				p.msg.Header.Set("Micro-Topic", record.Topic) | ||||
| 				p.msg.Header.Set("Micro-Key", string(record.Key)) | ||||
| 				p.msg.Header.Set("Micro-Timestamp", strconv.FormatInt(record.Timestamp.Unix(), 10)) | ||||
| 				if pc.kopts.Codec.String() == "noop" { | ||||
| 					p.msg.Body = record.Value | ||||
| 				} else if pc.opts.BodyOnly { | ||||
|   | ||||
		Reference in New Issue
	
	Block a user