graceful shutdown porting

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

go.mod

@@ -4,11 +4,11 @@ go 1.19
 
 require (
 	github.com/twmb/franz-go v1.11.5
-	go.unistack.org/micro/v4 v4.0.1
+	github.com/twmb/franz-go/pkg/kmsg v1.3.0
+	go.unistack.org/micro/v4 v4.0.2
 )
 
 require (
 	github.com/klauspost/compress v1.15.9 // indirect
 	github.com/pierrec/lz4/v4 v4.1.15 // indirect
-	github.com/twmb/franz-go/pkg/kmsg v1.3.0 // indirect
 )

go.sum

@@ -6,8 +6,8 @@ github.com/twmb/franz-go v1.11.5 h1:TTv5lVJd+87XkmP9dWN9Jgpf7IUUr7a7jee+byR8LBE=
 github.com/twmb/franz-go v1.11.5/go.mod h1:FvaHNlpT6woVYIl6LAuIeL7yHol1Fp6Gv2Dn21AvH78=
 github.com/twmb/franz-go/pkg/kmsg v1.3.0 h1:ouBETB7nTqRxiO5E8/pySoFZtVEW2VWw55z3/bsUzTw=
 github.com/twmb/franz-go/pkg/kmsg v1.3.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=
-go.unistack.org/micro/v4 v4.0.1 h1:xo1IxbVfgh8i0eY0VeYa3cbb13u5n/Mxnp3FOgWD4Jo=
-go.unistack.org/micro/v4 v4.0.1/go.mod h1:p/J5UcSJjfHsWGT31uKoghQ5rUQZzQJBAFy+Z4+ZVMs=
+go.unistack.org/micro/v4 v4.0.2 h1:2LeG6jslE50c72f1XwJhfTiidy67xklIC3saptLoUys=
+go.unistack.org/micro/v4 v4.0.2/go.mod h1:+wBa98rSf+mRXb/MuSVFPXtDrqN0k8rzPQiC8wRCwCo=
 golang.org/x/crypto v0.0.0-20220817201139-bc19a97f63c8/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

subscriber.go

@@ -22,7 +22,7 @@ type consumer struct {
 	partition int32
 	opts      broker.SubscribeOptions
 	kopts     broker.Options
-	handler   broker.Handler
+	handler   interface{}
 	quit      chan struct{}
 	done      chan struct{}
 	recs      chan kgo.FetchTopicPartition
@@ -33,7 +33,7 @@ type subscriber struct {
 	topic     string
 	opts      broker.SubscribeOptions
 	kopts     broker.Options
-	handler   broker.Handler
+	handler   interface{}
 	closed    bool
 	done      chan struct{}
 	consumers map[tp]*consumer