How to create a mock for JoinGroupResponse ?

[gunturaf]

Versions

Sarama	Kafka	Go
v1.26.1	2.0.0	1.14.0

Configuration

What configuration values are you using for Sarama and Kafka?

config := sarama.NewConfig()
config.ClientID = viper.GetString(cfg.ServiceName)
config.Version = sarama.V2_0_0_0
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
config.Consumer.Group.Rebalance.Retry.Max = 2

Logs

logs: CLICK ME

Initializing new client
client/metadata fetching metadata for all topics from broker 127.0.0.1:62364
Connected to broker at 127.0.0.1:62364 (unregistered)
*** mockbroker/0/0: connection opened*** mockbroker/0/0: served &{0 testkafka 0xc0003a00f0} -> &{1 0 [0xc0003b8000] <nil> 0 [0xc0003b6040]}client/brokers registered new broker #0 at 127.0.0.1:62364Successfully initialized new client
loop
client/metadata fetching metadata for [test_topic] from broker 127.0.0.1:62364
*** mockbroker/0/0: served &{1 testkafka 0xc0003a0330} -> &{1 0 [0xc0003b82c0] <nil> 0 [0xc0003b6240]}client/coordinator requesting coordinator for consumergroup test from 127.0.0.1:62364
*** mockbroker/0/0: served &{2 testkafka 0xc000412000} -> &{0 0s kafka server: Not an error, why are you printing me? <nil> 0xc000414000}client/coordinator coordinator for consumergroup test is #0 (127.0.0.1:62364)
*** mockbroker/0/1: connection openedConnected to broker at 127.0.0.1:62364 (registered as #0)
*** mockbroker/0/1: ignored (*sarama.request)(0xc00040c1e0)({
 correlationID: (int32) 0,
 clientID: (string) (len=9) "testkafka",
 body: (*sarama.JoinGroupRequest)(0xc000408540)({
  Version: (int16) 1,
  GroupId: (string) (len=4) "test",
  SessionTimeout: (int32) 10000,
  RebalanceTimeout: (int32) 60000,
  MemberId: (string) "",
  ProtocolType: (string) (len=8) "consumer",
  GroupProtocols: (map[string][]uint8) (len=1) {
   (string) (len=5) "range": ([]uint8) (len=22 cap=22) {
    00000000  00 00 00 00 00 01 00 0a  74 65 73 74 5f 74 6f 70  |........test_top|
    00000010  69 63 ff ff ff ff                                 |ic....|
   }
  },
  OrderedGroupProtocols: ([]*sarama.GroupProtocol) (len=1 cap=1) {
   (*sarama.GroupProtocol)(0xc00040c240)({
    Name: (string) (len=5) "range",
    Metadata: ([]uint8) (len=22 cap=22) {
     00000000  00 00 00 00 00 01 00 0a  74 65 73 74 5f 74 6f 70  |........test_top|
     00000010  69 63 ff ff ff ff                                 |ic....|
    }
   })
  }
 })
})

Problem Description

I am using sarama.NewClient and then sarama.NewConsumerGroupFromClient to create a consumer group, it works smoothly when compiled.

Now the question is on the testing side, I want to be able to test my function safely using mocks. I've write the test like so:

func_test.go: CLICK

broker0 := sarama.NewMockBroker(t, 0)
defer broker0.Close()

var testMsg = sarama.StringEncoder("Foo")

mockFetchResponse := sarama.NewMockFetchResponse(t, 1)
for i := 0; i < 10; i++ {
	mockFetchResponse.SetMessage("test_topic", 0, int64(i+1234), testMsg)
}

broker0.SetHandlerByMap(map[string]sarama.MockResponse{
	"MetadataRequest": sarama.NewMockMetadataResponse(t).
		SetBroker(broker0.Addr(), broker0.BrokerID()).
		SetLeader("test_topic", 0, broker0.BrokerID()).
		SetController(broker0.BrokerID()),
	"OffsetRequest": sarama.NewMockOffsetResponse(t).
		SetOffset("test_topic", 0, sarama.OffsetOldest, 0).
		SetOffset("test_topic", 0, sarama.OffsetNewest, 1),
	"FetchRequest": mockFetchResponse,
	"FindCoordinatorRequest": sarama.NewMockFindCoordinatorResponse(t).
		SetCoordinator(sarama.CoordinatorGroup, "test", broker0),
	// "JoinGroupRequest": ----,
})

but when I run my test, my sarama.ConsumerGroupHandler won't invoke the Setup function. I suppose this is caused by the incomplete mocked handlers in SetHandlerByMap, this log suggested that:

JoinGroupRequest is ignored: CLICK

*** mockbroker/0/1: ignored (*sarama.request)(0xc00040c1e0)({
 correlationID: (int32) 0,
 clientID: (string) (len=9) "testkafka",
 body: (*sarama.JoinGroupRequest)(0xc000408540)({
  Version: (int16) 1,
  GroupId: (string) (len=4) "test",
  SessionTimeout: (int32) 10000,
  RebalanceTimeout: (int32) 60000,
  MemberId: (string) "",
  ProtocolType: (string) (len=8) "consumer",
  GroupProtocols: (map[string][]uint8) (len=1) {
   (string) (len=5) "range": ([]uint8) (len=22 cap=22) {
    00000000  00 00 00 00 00 01 00 0a  74 65 73 74 5f 74 6f 70  |........test_top|
    00000010  69 63 ff ff ff ff                                 |ic....|
   }
  },
  OrderedGroupProtocols: ([]*sarama.GroupProtocol) (len=1 cap=1) {
   (*sarama.GroupProtocol)(0xc00040c240)({
    Name: (string) (len=5) "range",
    Metadata: ([]uint8) (len=22 cap=22) {
     00000000  00 00 00 00 00 01 00 0a  74 65 73 74 5f 74 6f 70  |........test_top|
     00000010  69 63 ff ff ff ff                                 |ic....|
    }
   })
  }
 })
})

I tried to create the mock by myself by trying to implement sarama.MockResponse interface but it is cannot be done because func For( ) needs a sarama.versionedDecoder parameter which is not exported, also the return value is expected to be a sarama.encoder which is also not exported.

Should we create a new factory function to mock JoinGroupResponse ? or is there any other suggestion?
Thanks

[MichaHoffmann]

I would be interessted in creating a NewJoinGroupResponse factory function. Is this still available?

[gunturaf]

@MichaHoffmann yes a factory function can also resolve this. I just looked at #1631 it seems it tries to solve this similar issue #1577 , maybe you can extend that PR because it got no activity for some months now. I might should close my issue then?

[gunturaf]

I'm closing this issue since #1750 had addressed this issue and #1577 cc @MichaHoffmann