mockresponses.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860
  1. package sarama
  2. import (
  3. "fmt"
  4. )
  5. // TestReporter has methods matching go's testing.T to avoid importing
  6. // `testing` in the main part of the library.
  7. type TestReporter interface {
  8. Error(...interface{})
  9. Errorf(string, ...interface{})
  10. Fatal(...interface{})
  11. Fatalf(string, ...interface{})
  12. }
  13. // MockResponse is a response builder interface it defines one method that
  14. // allows generating a response based on a request body. MockResponses are used
  15. // to program behavior of MockBroker in tests.
  16. type MockResponse interface {
  17. For(reqBody versionedDecoder) (res encoder)
  18. }
  19. // MockWrapper is a mock response builder that returns a particular concrete
  20. // response regardless of the actual request passed to the `For` method.
  21. type MockWrapper struct {
  22. res encoder
  23. }
  24. func (mw *MockWrapper) For(reqBody versionedDecoder) (res encoder) {
  25. return mw.res
  26. }
  27. func NewMockWrapper(res encoder) *MockWrapper {
  28. return &MockWrapper{res: res}
  29. }
  30. // MockSequence is a mock response builder that is created from a sequence of
  31. // concrete responses. Every time when a `MockBroker` calls its `For` method
  32. // the next response from the sequence is returned. When the end of the
  33. // sequence is reached the last element from the sequence is returned.
  34. type MockSequence struct {
  35. responses []MockResponse
  36. }
  37. func NewMockSequence(responses ...interface{}) *MockSequence {
  38. ms := &MockSequence{}
  39. ms.responses = make([]MockResponse, len(responses))
  40. for i, res := range responses {
  41. switch res := res.(type) {
  42. case MockResponse:
  43. ms.responses[i] = res
  44. case encoder:
  45. ms.responses[i] = NewMockWrapper(res)
  46. default:
  47. panic(fmt.Sprintf("Unexpected response type: %T", res))
  48. }
  49. }
  50. return ms
  51. }
  52. func (mc *MockSequence) For(reqBody versionedDecoder) (res encoder) {
  53. res = mc.responses[0].For(reqBody)
  54. if len(mc.responses) > 1 {
  55. mc.responses = mc.responses[1:]
  56. }
  57. return res
  58. }
  59. type MockListGroupsResponse struct {
  60. groups map[string]string
  61. t TestReporter
  62. }
  63. func NewMockListGroupsResponse(t TestReporter) *MockListGroupsResponse {
  64. return &MockListGroupsResponse{
  65. groups: make(map[string]string),
  66. t: t,
  67. }
  68. }
  69. func (m *MockListGroupsResponse) For(reqBody versionedDecoder) encoder {
  70. request := reqBody.(*ListGroupsRequest)
  71. _ = request
  72. response := &ListGroupsResponse{
  73. Groups: m.groups,
  74. }
  75. return response
  76. }
  77. func (m *MockListGroupsResponse) AddGroup(groupID, protocolType string) *MockListGroupsResponse {
  78. m.groups[groupID] = protocolType
  79. return m
  80. }
  81. type MockDescribeGroupsResponse struct {
  82. groups map[string]*GroupDescription
  83. t TestReporter
  84. }
  85. func NewMockDescribeGroupsResponse(t TestReporter) *MockDescribeGroupsResponse {
  86. return &MockDescribeGroupsResponse{
  87. t: t,
  88. groups: make(map[string]*GroupDescription),
  89. }
  90. }
  91. func (m *MockDescribeGroupsResponse) AddGroupDescription(groupID string, description *GroupDescription) *MockDescribeGroupsResponse {
  92. m.groups[groupID] = description
  93. return m
  94. }
  95. func (m *MockDescribeGroupsResponse) For(reqBody versionedDecoder) encoder {
  96. request := reqBody.(*DescribeGroupsRequest)
  97. response := &DescribeGroupsResponse{}
  98. for _, requestedGroup := range request.Groups {
  99. if group, ok := m.groups[requestedGroup]; ok {
  100. response.Groups = append(response.Groups, group)
  101. } else {
  102. // Mimic real kafka - if a group doesn't exist, return
  103. // an entry with state "Dead"
  104. response.Groups = append(response.Groups, &GroupDescription{
  105. GroupId: requestedGroup,
  106. State: "Dead",
  107. })
  108. }
  109. }
  110. return response
  111. }
  112. // MockMetadataResponse is a `MetadataResponse` builder.
  113. type MockMetadataResponse struct {
  114. controllerID int32
  115. leaders map[string]map[int32]int32
  116. brokers map[string]int32
  117. t TestReporter
  118. }
  119. func NewMockMetadataResponse(t TestReporter) *MockMetadataResponse {
  120. return &MockMetadataResponse{
  121. leaders: make(map[string]map[int32]int32),
  122. brokers: make(map[string]int32),
  123. t: t,
  124. }
  125. }
  126. func (mmr *MockMetadataResponse) SetLeader(topic string, partition, brokerID int32) *MockMetadataResponse {
  127. partitions := mmr.leaders[topic]
  128. if partitions == nil {
  129. partitions = make(map[int32]int32)
  130. mmr.leaders[topic] = partitions
  131. }
  132. partitions[partition] = brokerID
  133. return mmr
  134. }
  135. func (mmr *MockMetadataResponse) SetBroker(addr string, brokerID int32) *MockMetadataResponse {
  136. mmr.brokers[addr] = brokerID
  137. return mmr
  138. }
  139. func (mmr *MockMetadataResponse) SetController(brokerID int32) *MockMetadataResponse {
  140. mmr.controllerID = brokerID
  141. return mmr
  142. }
  143. func (mmr *MockMetadataResponse) For(reqBody versionedDecoder) encoder {
  144. metadataRequest := reqBody.(*MetadataRequest)
  145. metadataResponse := &MetadataResponse{
  146. Version: metadataRequest.version(),
  147. ControllerID: mmr.controllerID,
  148. }
  149. for addr, brokerID := range mmr.brokers {
  150. metadataResponse.AddBroker(addr, brokerID)
  151. }
  152. if len(metadataRequest.Topics) == 0 {
  153. for topic, partitions := range mmr.leaders {
  154. for partition, brokerID := range partitions {
  155. metadataResponse.AddTopicPartition(topic, partition, brokerID, nil, nil, ErrNoError)
  156. }
  157. }
  158. return metadataResponse
  159. }
  160. for _, topic := range metadataRequest.Topics {
  161. for partition, brokerID := range mmr.leaders[topic] {
  162. metadataResponse.AddTopicPartition(topic, partition, brokerID, nil, nil, ErrNoError)
  163. }
  164. }
  165. return metadataResponse
  166. }
  167. // MockOffsetResponse is an `OffsetResponse` builder.
  168. type MockOffsetResponse struct {
  169. offsets map[string]map[int32]map[int64]int64
  170. t TestReporter
  171. version int16
  172. }
  173. func NewMockOffsetResponse(t TestReporter) *MockOffsetResponse {
  174. return &MockOffsetResponse{
  175. offsets: make(map[string]map[int32]map[int64]int64),
  176. t: t,
  177. }
  178. }
  179. func (mor *MockOffsetResponse) SetVersion(version int16) *MockOffsetResponse {
  180. mor.version = version
  181. return mor
  182. }
  183. func (mor *MockOffsetResponse) SetOffset(topic string, partition int32, time, offset int64) *MockOffsetResponse {
  184. partitions := mor.offsets[topic]
  185. if partitions == nil {
  186. partitions = make(map[int32]map[int64]int64)
  187. mor.offsets[topic] = partitions
  188. }
  189. times := partitions[partition]
  190. if times == nil {
  191. times = make(map[int64]int64)
  192. partitions[partition] = times
  193. }
  194. times[time] = offset
  195. return mor
  196. }
  197. func (mor *MockOffsetResponse) For(reqBody versionedDecoder) encoder {
  198. offsetRequest := reqBody.(*OffsetRequest)
  199. offsetResponse := &OffsetResponse{Version: mor.version}
  200. for topic, partitions := range offsetRequest.blocks {
  201. for partition, block := range partitions {
  202. offset := mor.getOffset(topic, partition, block.time)
  203. offsetResponse.AddTopicPartition(topic, partition, offset)
  204. }
  205. }
  206. return offsetResponse
  207. }
  208. func (mor *MockOffsetResponse) getOffset(topic string, partition int32, time int64) int64 {
  209. partitions := mor.offsets[topic]
  210. if partitions == nil {
  211. mor.t.Errorf("missing topic: %s", topic)
  212. }
  213. times := partitions[partition]
  214. if times == nil {
  215. mor.t.Errorf("missing partition: %d", partition)
  216. }
  217. offset, ok := times[time]
  218. if !ok {
  219. mor.t.Errorf("missing time: %d", time)
  220. }
  221. return offset
  222. }
  223. // MockFetchResponse is a `FetchResponse` builder.
  224. type MockFetchResponse struct {
  225. messages map[string]map[int32]map[int64]Encoder
  226. highWaterMarks map[string]map[int32]int64
  227. t TestReporter
  228. batchSize int
  229. version int16
  230. }
  231. func NewMockFetchResponse(t TestReporter, batchSize int) *MockFetchResponse {
  232. return &MockFetchResponse{
  233. messages: make(map[string]map[int32]map[int64]Encoder),
  234. highWaterMarks: make(map[string]map[int32]int64),
  235. t: t,
  236. batchSize: batchSize,
  237. }
  238. }
  239. func (mfr *MockFetchResponse) SetVersion(version int16) *MockFetchResponse {
  240. mfr.version = version
  241. return mfr
  242. }
  243. func (mfr *MockFetchResponse) SetMessage(topic string, partition int32, offset int64, msg Encoder) *MockFetchResponse {
  244. partitions := mfr.messages[topic]
  245. if partitions == nil {
  246. partitions = make(map[int32]map[int64]Encoder)
  247. mfr.messages[topic] = partitions
  248. }
  249. messages := partitions[partition]
  250. if messages == nil {
  251. messages = make(map[int64]Encoder)
  252. partitions[partition] = messages
  253. }
  254. messages[offset] = msg
  255. return mfr
  256. }
  257. func (mfr *MockFetchResponse) SetHighWaterMark(topic string, partition int32, offset int64) *MockFetchResponse {
  258. partitions := mfr.highWaterMarks[topic]
  259. if partitions == nil {
  260. partitions = make(map[int32]int64)
  261. mfr.highWaterMarks[topic] = partitions
  262. }
  263. partitions[partition] = offset
  264. return mfr
  265. }
  266. func (mfr *MockFetchResponse) For(reqBody versionedDecoder) encoder {
  267. fetchRequest := reqBody.(*FetchRequest)
  268. res := &FetchResponse{
  269. Version: mfr.version,
  270. }
  271. for topic, partitions := range fetchRequest.blocks {
  272. for partition, block := range partitions {
  273. initialOffset := block.fetchOffset
  274. offset := initialOffset
  275. maxOffset := initialOffset + int64(mfr.getMessageCount(topic, partition))
  276. for i := 0; i < mfr.batchSize && offset < maxOffset; {
  277. msg := mfr.getMessage(topic, partition, offset)
  278. if msg != nil {
  279. res.AddMessage(topic, partition, nil, msg, offset)
  280. i++
  281. }
  282. offset++
  283. }
  284. fb := res.GetBlock(topic, partition)
  285. if fb == nil {
  286. res.AddError(topic, partition, ErrNoError)
  287. fb = res.GetBlock(topic, partition)
  288. }
  289. fb.HighWaterMarkOffset = mfr.getHighWaterMark(topic, partition)
  290. }
  291. }
  292. return res
  293. }
  294. func (mfr *MockFetchResponse) getMessage(topic string, partition int32, offset int64) Encoder {
  295. partitions := mfr.messages[topic]
  296. if partitions == nil {
  297. return nil
  298. }
  299. messages := partitions[partition]
  300. if messages == nil {
  301. return nil
  302. }
  303. return messages[offset]
  304. }
  305. func (mfr *MockFetchResponse) getMessageCount(topic string, partition int32) int {
  306. partitions := mfr.messages[topic]
  307. if partitions == nil {
  308. return 0
  309. }
  310. messages := partitions[partition]
  311. if messages == nil {
  312. return 0
  313. }
  314. return len(messages)
  315. }
  316. func (mfr *MockFetchResponse) getHighWaterMark(topic string, partition int32) int64 {
  317. partitions := mfr.highWaterMarks[topic]
  318. if partitions == nil {
  319. return 0
  320. }
  321. return partitions[partition]
  322. }
  323. // MockConsumerMetadataResponse is a `ConsumerMetadataResponse` builder.
  324. type MockConsumerMetadataResponse struct {
  325. coordinators map[string]interface{}
  326. t TestReporter
  327. }
  328. func NewMockConsumerMetadataResponse(t TestReporter) *MockConsumerMetadataResponse {
  329. return &MockConsumerMetadataResponse{
  330. coordinators: make(map[string]interface{}),
  331. t: t,
  332. }
  333. }
  334. func (mr *MockConsumerMetadataResponse) SetCoordinator(group string, broker *MockBroker) *MockConsumerMetadataResponse {
  335. mr.coordinators[group] = broker
  336. return mr
  337. }
  338. func (mr *MockConsumerMetadataResponse) SetError(group string, kerror KError) *MockConsumerMetadataResponse {
  339. mr.coordinators[group] = kerror
  340. return mr
  341. }
  342. func (mr *MockConsumerMetadataResponse) For(reqBody versionedDecoder) encoder {
  343. req := reqBody.(*ConsumerMetadataRequest)
  344. group := req.ConsumerGroup
  345. res := &ConsumerMetadataResponse{}
  346. v := mr.coordinators[group]
  347. switch v := v.(type) {
  348. case *MockBroker:
  349. res.Coordinator = &Broker{id: v.BrokerID(), addr: v.Addr()}
  350. case KError:
  351. res.Err = v
  352. }
  353. return res
  354. }
  355. // MockFindCoordinatorResponse is a `FindCoordinatorResponse` builder.
  356. type MockFindCoordinatorResponse struct {
  357. groupCoordinators map[string]interface{}
  358. transCoordinators map[string]interface{}
  359. t TestReporter
  360. }
  361. func NewMockFindCoordinatorResponse(t TestReporter) *MockFindCoordinatorResponse {
  362. return &MockFindCoordinatorResponse{
  363. groupCoordinators: make(map[string]interface{}),
  364. transCoordinators: make(map[string]interface{}),
  365. t: t,
  366. }
  367. }
  368. func (mr *MockFindCoordinatorResponse) SetCoordinator(coordinatorType CoordinatorType, group string, broker *MockBroker) *MockFindCoordinatorResponse {
  369. switch coordinatorType {
  370. case CoordinatorGroup:
  371. mr.groupCoordinators[group] = broker
  372. case CoordinatorTransaction:
  373. mr.transCoordinators[group] = broker
  374. }
  375. return mr
  376. }
  377. func (mr *MockFindCoordinatorResponse) SetError(coordinatorType CoordinatorType, group string, kerror KError) *MockFindCoordinatorResponse {
  378. switch coordinatorType {
  379. case CoordinatorGroup:
  380. mr.groupCoordinators[group] = kerror
  381. case CoordinatorTransaction:
  382. mr.transCoordinators[group] = kerror
  383. }
  384. return mr
  385. }
  386. func (mr *MockFindCoordinatorResponse) For(reqBody versionedDecoder) encoder {
  387. req := reqBody.(*FindCoordinatorRequest)
  388. res := &FindCoordinatorResponse{}
  389. var v interface{}
  390. switch req.CoordinatorType {
  391. case CoordinatorGroup:
  392. v = mr.groupCoordinators[req.CoordinatorKey]
  393. case CoordinatorTransaction:
  394. v = mr.transCoordinators[req.CoordinatorKey]
  395. }
  396. switch v := v.(type) {
  397. case *MockBroker:
  398. res.Coordinator = &Broker{id: v.BrokerID(), addr: v.Addr()}
  399. case KError:
  400. res.Err = v
  401. }
  402. return res
  403. }
  404. // MockOffsetCommitResponse is a `OffsetCommitResponse` builder.
  405. type MockOffsetCommitResponse struct {
  406. errors map[string]map[string]map[int32]KError
  407. t TestReporter
  408. }
  409. func NewMockOffsetCommitResponse(t TestReporter) *MockOffsetCommitResponse {
  410. return &MockOffsetCommitResponse{t: t}
  411. }
  412. func (mr *MockOffsetCommitResponse) SetError(group, topic string, partition int32, kerror KError) *MockOffsetCommitResponse {
  413. if mr.errors == nil {
  414. mr.errors = make(map[string]map[string]map[int32]KError)
  415. }
  416. topics := mr.errors[group]
  417. if topics == nil {
  418. topics = make(map[string]map[int32]KError)
  419. mr.errors[group] = topics
  420. }
  421. partitions := topics[topic]
  422. if partitions == nil {
  423. partitions = make(map[int32]KError)
  424. topics[topic] = partitions
  425. }
  426. partitions[partition] = kerror
  427. return mr
  428. }
  429. func (mr *MockOffsetCommitResponse) For(reqBody versionedDecoder) encoder {
  430. req := reqBody.(*OffsetCommitRequest)
  431. group := req.ConsumerGroup
  432. res := &OffsetCommitResponse{}
  433. for topic, partitions := range req.blocks {
  434. for partition := range partitions {
  435. res.AddError(topic, partition, mr.getError(group, topic, partition))
  436. }
  437. }
  438. return res
  439. }
  440. func (mr *MockOffsetCommitResponse) getError(group, topic string, partition int32) KError {
  441. topics := mr.errors[group]
  442. if topics == nil {
  443. return ErrNoError
  444. }
  445. partitions := topics[topic]
  446. if partitions == nil {
  447. return ErrNoError
  448. }
  449. kerror, ok := partitions[partition]
  450. if !ok {
  451. return ErrNoError
  452. }
  453. return kerror
  454. }
  455. // MockProduceResponse is a `ProduceResponse` builder.
  456. type MockProduceResponse struct {
  457. version int16
  458. errors map[string]map[int32]KError
  459. t TestReporter
  460. }
  461. func NewMockProduceResponse(t TestReporter) *MockProduceResponse {
  462. return &MockProduceResponse{t: t}
  463. }
  464. func (mr *MockProduceResponse) SetVersion(version int16) *MockProduceResponse {
  465. mr.version = version
  466. return mr
  467. }
  468. func (mr *MockProduceResponse) SetError(topic string, partition int32, kerror KError) *MockProduceResponse {
  469. if mr.errors == nil {
  470. mr.errors = make(map[string]map[int32]KError)
  471. }
  472. partitions := mr.errors[topic]
  473. if partitions == nil {
  474. partitions = make(map[int32]KError)
  475. mr.errors[topic] = partitions
  476. }
  477. partitions[partition] = kerror
  478. return mr
  479. }
  480. func (mr *MockProduceResponse) For(reqBody versionedDecoder) encoder {
  481. req := reqBody.(*ProduceRequest)
  482. res := &ProduceResponse{
  483. Version: mr.version,
  484. }
  485. for topic, partitions := range req.records {
  486. for partition := range partitions {
  487. res.AddTopicPartition(topic, partition, mr.getError(topic, partition))
  488. }
  489. }
  490. return res
  491. }
  492. func (mr *MockProduceResponse) getError(topic string, partition int32) KError {
  493. partitions := mr.errors[topic]
  494. if partitions == nil {
  495. return ErrNoError
  496. }
  497. kerror, ok := partitions[partition]
  498. if !ok {
  499. return ErrNoError
  500. }
  501. return kerror
  502. }
  503. // MockOffsetFetchResponse is a `OffsetFetchResponse` builder.
  504. type MockOffsetFetchResponse struct {
  505. offsets map[string]map[string]map[int32]*OffsetFetchResponseBlock
  506. t TestReporter
  507. }
  508. func NewMockOffsetFetchResponse(t TestReporter) *MockOffsetFetchResponse {
  509. return &MockOffsetFetchResponse{t: t}
  510. }
  511. func (mr *MockOffsetFetchResponse) SetOffset(group, topic string, partition int32, offset int64, metadata string, kerror KError) *MockOffsetFetchResponse {
  512. if mr.offsets == nil {
  513. mr.offsets = make(map[string]map[string]map[int32]*OffsetFetchResponseBlock)
  514. }
  515. topics := mr.offsets[group]
  516. if topics == nil {
  517. topics = make(map[string]map[int32]*OffsetFetchResponseBlock)
  518. mr.offsets[group] = topics
  519. }
  520. partitions := topics[topic]
  521. if partitions == nil {
  522. partitions = make(map[int32]*OffsetFetchResponseBlock)
  523. topics[topic] = partitions
  524. }
  525. partitions[partition] = &OffsetFetchResponseBlock{offset, 0, metadata, kerror}
  526. return mr
  527. }
  528. func (mr *MockOffsetFetchResponse) For(reqBody versionedDecoder) encoder {
  529. req := reqBody.(*OffsetFetchRequest)
  530. group := req.ConsumerGroup
  531. res := &OffsetFetchResponse{}
  532. for topic, partitions := range mr.offsets[group] {
  533. for partition, block := range partitions {
  534. res.AddBlock(topic, partition, block)
  535. }
  536. }
  537. return res
  538. }
  539. type MockCreateTopicsResponse struct {
  540. t TestReporter
  541. }
  542. func NewMockCreateTopicsResponse(t TestReporter) *MockCreateTopicsResponse {
  543. return &MockCreateTopicsResponse{t: t}
  544. }
  545. func (mr *MockCreateTopicsResponse) For(reqBody versionedDecoder) encoder {
  546. req := reqBody.(*CreateTopicsRequest)
  547. res := &CreateTopicsResponse{}
  548. res.TopicErrors = make(map[string]*TopicError)
  549. for topic, _ := range req.TopicDetails {
  550. res.TopicErrors[topic] = &TopicError{Err: ErrNoError}
  551. }
  552. return res
  553. }
  554. type MockDeleteTopicsResponse struct {
  555. t TestReporter
  556. }
  557. func NewMockDeleteTopicsResponse(t TestReporter) *MockDeleteTopicsResponse {
  558. return &MockDeleteTopicsResponse{t: t}
  559. }
  560. func (mr *MockDeleteTopicsResponse) For(reqBody versionedDecoder) encoder {
  561. req := reqBody.(*DeleteTopicsRequest)
  562. res := &DeleteTopicsResponse{}
  563. res.TopicErrorCodes = make(map[string]KError)
  564. for _, topic := range req.Topics {
  565. res.TopicErrorCodes[topic] = ErrNoError
  566. }
  567. return res
  568. }
  569. type MockCreatePartitionsResponse struct {
  570. t TestReporter
  571. }
  572. func NewMockCreatePartitionsResponse(t TestReporter) *MockCreatePartitionsResponse {
  573. return &MockCreatePartitionsResponse{t: t}
  574. }
  575. func (mr *MockCreatePartitionsResponse) For(reqBody versionedDecoder) encoder {
  576. req := reqBody.(*CreatePartitionsRequest)
  577. res := &CreatePartitionsResponse{}
  578. res.TopicPartitionErrors = make(map[string]*TopicPartitionError)
  579. for topic, _ := range req.TopicPartitions {
  580. res.TopicPartitionErrors[topic] = &TopicPartitionError{Err: ErrNoError}
  581. }
  582. return res
  583. }
  584. type MockDeleteRecordsResponse struct {
  585. t TestReporter
  586. }
  587. func NewMockDeleteRecordsResponse(t TestReporter) *MockDeleteRecordsResponse {
  588. return &MockDeleteRecordsResponse{t: t}
  589. }
  590. func (mr *MockDeleteRecordsResponse) For(reqBody versionedDecoder) encoder {
  591. req := reqBody.(*DeleteRecordsRequest)
  592. res := &DeleteRecordsResponse{}
  593. res.Topics = make(map[string]*DeleteRecordsResponseTopic)
  594. for topic, deleteRecordRequestTopic := range req.Topics {
  595. partitions := make(map[int32]*DeleteRecordsResponsePartition)
  596. for partition, _ := range deleteRecordRequestTopic.PartitionOffsets {
  597. partitions[partition] = &DeleteRecordsResponsePartition{Err: ErrNoError}
  598. }
  599. res.Topics[topic] = &DeleteRecordsResponseTopic{Partitions: partitions}
  600. }
  601. return res
  602. }
  603. type MockDescribeConfigsResponse struct {
  604. t TestReporter
  605. }
  606. func NewMockDescribeConfigsResponse(t TestReporter) *MockDescribeConfigsResponse {
  607. return &MockDescribeConfigsResponse{t: t}
  608. }
  609. func (mr *MockDescribeConfigsResponse) For(reqBody versionedDecoder) encoder {
  610. req := reqBody.(*DescribeConfigsRequest)
  611. res := &DescribeConfigsResponse{}
  612. for _, r := range req.Resources {
  613. var configEntries []*ConfigEntry
  614. switch r.Type {
  615. case TopicResource:
  616. configEntries = append(configEntries,
  617. &ConfigEntry{Name: "max.message.bytes",
  618. Value: "1000000",
  619. ReadOnly: false,
  620. Default: true,
  621. Sensitive: false,
  622. }, &ConfigEntry{Name: "retention.ms",
  623. Value: "5000",
  624. ReadOnly: false,
  625. Default: false,
  626. Sensitive: false,
  627. }, &ConfigEntry{Name: "password",
  628. Value: "12345",
  629. ReadOnly: false,
  630. Default: false,
  631. Sensitive: true,
  632. })
  633. res.Resources = append(res.Resources, &ResourceResponse{
  634. Name: r.Name,
  635. Configs: configEntries,
  636. })
  637. }
  638. }
  639. return res
  640. }
  641. type MockAlterConfigsResponse struct {
  642. t TestReporter
  643. }
  644. func NewMockAlterConfigsResponse(t TestReporter) *MockAlterConfigsResponse {
  645. return &MockAlterConfigsResponse{t: t}
  646. }
  647. func (mr *MockAlterConfigsResponse) For(reqBody versionedDecoder) encoder {
  648. req := reqBody.(*AlterConfigsRequest)
  649. res := &AlterConfigsResponse{}
  650. for _, r := range req.Resources {
  651. res.Resources = append(res.Resources, &AlterConfigsResourceResponse{Name: r.Name,
  652. Type: TopicResource,
  653. ErrorMsg: "",
  654. })
  655. }
  656. return res
  657. }
  658. type MockCreateAclsResponse struct {
  659. t TestReporter
  660. }
  661. func NewMockCreateAclsResponse(t TestReporter) *MockCreateAclsResponse {
  662. return &MockCreateAclsResponse{t: t}
  663. }
  664. func (mr *MockCreateAclsResponse) For(reqBody versionedDecoder) encoder {
  665. req := reqBody.(*CreateAclsRequest)
  666. res := &CreateAclsResponse{}
  667. for range req.AclCreations {
  668. res.AclCreationResponses = append(res.AclCreationResponses, &AclCreationResponse{Err: ErrNoError})
  669. }
  670. return res
  671. }
  672. type MockListAclsResponse struct {
  673. t TestReporter
  674. }
  675. func NewMockListAclsResponse(t TestReporter) *MockListAclsResponse {
  676. return &MockListAclsResponse{t: t}
  677. }
  678. func (mr *MockListAclsResponse) For(reqBody versionedDecoder) encoder {
  679. req := reqBody.(*DescribeAclsRequest)
  680. res := &DescribeAclsResponse{}
  681. res.Err = ErrNoError
  682. acl := &ResourceAcls{}
  683. acl.Resource.ResourceName = *req.ResourceName
  684. acl.Resource.ResourceType = req.ResourceType
  685. acl.Acls = append(acl.Acls, &Acl{})
  686. res.ResourceAcls = append(res.ResourceAcls, acl)
  687. return res
  688. }
  689. type MockSaslAuthenticateResponse struct {
  690. t TestReporter
  691. kerror KError
  692. saslAuthBytes []byte
  693. }
  694. func NewMockSaslAuthenticateResponse(t TestReporter) *MockSaslAuthenticateResponse {
  695. return &MockSaslAuthenticateResponse{t: t}
  696. }
  697. func (msar *MockSaslAuthenticateResponse) For(reqBody versionedDecoder) encoder {
  698. res := &SaslAuthenticateResponse{}
  699. res.Err = msar.kerror
  700. res.SaslAuthBytes = msar.saslAuthBytes
  701. return res
  702. }
  703. func (msar *MockSaslAuthenticateResponse) SetError(kerror KError) *MockSaslAuthenticateResponse {
  704. msar.kerror = kerror
  705. return msar
  706. }
  707. func (msar *MockSaslAuthenticateResponse) SetAuthBytes(saslAuthBytes []byte) *MockSaslAuthenticateResponse {
  708. msar.saslAuthBytes = saslAuthBytes
  709. return msar
  710. }
  711. type MockDeleteAclsResponse struct {
  712. t TestReporter
  713. }
  714. type MockSaslHandshakeResponse struct {
  715. enabledMechanisms []string
  716. kerror KError
  717. t TestReporter
  718. }
  719. func NewMockSaslHandshakeResponse(t TestReporter) *MockSaslHandshakeResponse {
  720. return &MockSaslHandshakeResponse{t: t}
  721. }
  722. func (mshr *MockSaslHandshakeResponse) For(reqBody versionedDecoder) encoder {
  723. res := &SaslHandshakeResponse{}
  724. res.Err = mshr.kerror
  725. res.EnabledMechanisms = mshr.enabledMechanisms
  726. return res
  727. }
  728. func (mshr *MockSaslHandshakeResponse) SetError(kerror KError) *MockSaslHandshakeResponse {
  729. mshr.kerror = kerror
  730. return mshr
  731. }
  732. func (mshr *MockSaslHandshakeResponse) SetEnabledMechanisms(enabledMechanisms []string) *MockSaslHandshakeResponse {
  733. mshr.enabledMechanisms = enabledMechanisms
  734. return mshr
  735. }
  736. func NewMockDeleteAclsResponse(t TestReporter) *MockDeleteAclsResponse {
  737. return &MockDeleteAclsResponse{t: t}
  738. }
  739. func (mr *MockDeleteAclsResponse) For(reqBody versionedDecoder) encoder {
  740. req := reqBody.(*DeleteAclsRequest)
  741. res := &DeleteAclsResponse{}
  742. for range req.Filters {
  743. response := &FilterResponse{Err: ErrNoError}
  744. response.MatchingAcls = append(response.MatchingAcls, &MatchingAcl{Err: ErrNoError})
  745. res.FilterResponses = append(res.FilterResponses, response)
  746. }
  747. return res
  748. }