Takahiro Suzuki | d7bf820 | 2020-12-17 20:21:59 +0900 | [diff] [blame] | 1 | package sarama |
| 2 | |
| 3 | import "github.com/rcrowley/go-metrics" |
| 4 | |
| 5 | // RequiredAcks is used in Produce Requests to tell the broker how many replica acknowledgements |
| 6 | // it must see before responding. Any of the constants defined here are valid. On broker versions |
| 7 | // prior to 0.8.2.0 any other positive int16 is also valid (the broker will wait for that many |
| 8 | // acknowledgements) but in 0.8.2.0 and later this will raise an exception (it has been replaced |
| 9 | // by setting the `min.isr` value in the brokers configuration). |
| 10 | type RequiredAcks int16 |
| 11 | |
| 12 | const ( |
| 13 | // NoResponse doesn't send any response, the TCP ACK is all you get. |
| 14 | NoResponse RequiredAcks = 0 |
| 15 | // WaitForLocal waits for only the local commit to succeed before responding. |
| 16 | WaitForLocal RequiredAcks = 1 |
| 17 | // WaitForAll waits for all in-sync replicas to commit before responding. |
| 18 | // The minimum number of in-sync replicas is configured on the broker via |
| 19 | // the `min.insync.replicas` configuration key. |
| 20 | WaitForAll RequiredAcks = -1 |
| 21 | ) |
| 22 | |
| 23 | type ProduceRequest struct { |
| 24 | TransactionalID *string |
| 25 | RequiredAcks RequiredAcks |
| 26 | Timeout int32 |
| 27 | Version int16 // v1 requires Kafka 0.9, v2 requires Kafka 0.10, v3 requires Kafka 0.11 |
| 28 | records map[string]map[int32]Records |
| 29 | } |
| 30 | |
| 31 | func updateMsgSetMetrics(msgSet *MessageSet, compressionRatioMetric metrics.Histogram, |
| 32 | topicCompressionRatioMetric metrics.Histogram) int64 { |
| 33 | var topicRecordCount int64 |
| 34 | for _, messageBlock := range msgSet.Messages { |
| 35 | // Is this a fake "message" wrapping real messages? |
| 36 | if messageBlock.Msg.Set != nil { |
| 37 | topicRecordCount += int64(len(messageBlock.Msg.Set.Messages)) |
| 38 | } else { |
| 39 | // A single uncompressed message |
| 40 | topicRecordCount++ |
| 41 | } |
| 42 | // Better be safe than sorry when computing the compression ratio |
| 43 | if messageBlock.Msg.compressedSize != 0 { |
| 44 | compressionRatio := float64(len(messageBlock.Msg.Value)) / |
| 45 | float64(messageBlock.Msg.compressedSize) |
| 46 | // Histogram do not support decimal values, let's multiple it by 100 for better precision |
| 47 | intCompressionRatio := int64(100 * compressionRatio) |
| 48 | compressionRatioMetric.Update(intCompressionRatio) |
| 49 | topicCompressionRatioMetric.Update(intCompressionRatio) |
| 50 | } |
| 51 | } |
| 52 | return topicRecordCount |
| 53 | } |
| 54 | |
| 55 | func updateBatchMetrics(recordBatch *RecordBatch, compressionRatioMetric metrics.Histogram, |
| 56 | topicCompressionRatioMetric metrics.Histogram) int64 { |
| 57 | if recordBatch.compressedRecords != nil { |
| 58 | compressionRatio := int64(float64(recordBatch.recordsLen) / float64(len(recordBatch.compressedRecords)) * 100) |
| 59 | compressionRatioMetric.Update(compressionRatio) |
| 60 | topicCompressionRatioMetric.Update(compressionRatio) |
| 61 | } |
| 62 | |
| 63 | return int64(len(recordBatch.Records)) |
| 64 | } |
| 65 | |
| 66 | func (r *ProduceRequest) encode(pe packetEncoder) error { |
| 67 | if r.Version >= 3 { |
| 68 | if err := pe.putNullableString(r.TransactionalID); err != nil { |
| 69 | return err |
| 70 | } |
| 71 | } |
| 72 | pe.putInt16(int16(r.RequiredAcks)) |
| 73 | pe.putInt32(r.Timeout) |
| 74 | metricRegistry := pe.metricRegistry() |
| 75 | var batchSizeMetric metrics.Histogram |
| 76 | var compressionRatioMetric metrics.Histogram |
| 77 | if metricRegistry != nil { |
| 78 | batchSizeMetric = getOrRegisterHistogram("batch-size", metricRegistry) |
| 79 | compressionRatioMetric = getOrRegisterHistogram("compression-ratio", metricRegistry) |
| 80 | } |
| 81 | totalRecordCount := int64(0) |
| 82 | |
| 83 | err := pe.putArrayLength(len(r.records)) |
| 84 | if err != nil { |
| 85 | return err |
| 86 | } |
| 87 | |
| 88 | for topic, partitions := range r.records { |
| 89 | err = pe.putString(topic) |
| 90 | if err != nil { |
| 91 | return err |
| 92 | } |
| 93 | err = pe.putArrayLength(len(partitions)) |
| 94 | if err != nil { |
| 95 | return err |
| 96 | } |
| 97 | topicRecordCount := int64(0) |
| 98 | var topicCompressionRatioMetric metrics.Histogram |
| 99 | if metricRegistry != nil { |
| 100 | topicCompressionRatioMetric = getOrRegisterTopicHistogram("compression-ratio", topic, metricRegistry) |
| 101 | } |
| 102 | for id, records := range partitions { |
| 103 | startOffset := pe.offset() |
| 104 | pe.putInt32(id) |
| 105 | pe.push(&lengthField{}) |
| 106 | err = records.encode(pe) |
| 107 | if err != nil { |
| 108 | return err |
| 109 | } |
| 110 | err = pe.pop() |
| 111 | if err != nil { |
| 112 | return err |
| 113 | } |
| 114 | if metricRegistry != nil { |
| 115 | if r.Version >= 3 { |
| 116 | topicRecordCount += updateBatchMetrics(records.RecordBatch, compressionRatioMetric, topicCompressionRatioMetric) |
| 117 | } else { |
| 118 | topicRecordCount += updateMsgSetMetrics(records.MsgSet, compressionRatioMetric, topicCompressionRatioMetric) |
| 119 | } |
| 120 | batchSize := int64(pe.offset() - startOffset) |
| 121 | batchSizeMetric.Update(batchSize) |
| 122 | getOrRegisterTopicHistogram("batch-size", topic, metricRegistry).Update(batchSize) |
| 123 | } |
| 124 | } |
| 125 | if topicRecordCount > 0 { |
| 126 | getOrRegisterTopicMeter("record-send-rate", topic, metricRegistry).Mark(topicRecordCount) |
| 127 | getOrRegisterTopicHistogram("records-per-request", topic, metricRegistry).Update(topicRecordCount) |
| 128 | totalRecordCount += topicRecordCount |
| 129 | } |
| 130 | } |
| 131 | if totalRecordCount > 0 { |
| 132 | metrics.GetOrRegisterMeter("record-send-rate", metricRegistry).Mark(totalRecordCount) |
| 133 | getOrRegisterHistogram("records-per-request", metricRegistry).Update(totalRecordCount) |
| 134 | } |
| 135 | |
| 136 | return nil |
| 137 | } |
| 138 | |
| 139 | func (r *ProduceRequest) decode(pd packetDecoder, version int16) error { |
| 140 | r.Version = version |
| 141 | |
| 142 | if version >= 3 { |
| 143 | id, err := pd.getNullableString() |
| 144 | if err != nil { |
| 145 | return err |
| 146 | } |
| 147 | r.TransactionalID = id |
| 148 | } |
| 149 | requiredAcks, err := pd.getInt16() |
| 150 | if err != nil { |
| 151 | return err |
| 152 | } |
| 153 | r.RequiredAcks = RequiredAcks(requiredAcks) |
| 154 | if r.Timeout, err = pd.getInt32(); err != nil { |
| 155 | return err |
| 156 | } |
| 157 | topicCount, err := pd.getArrayLength() |
| 158 | if err != nil { |
| 159 | return err |
| 160 | } |
| 161 | if topicCount == 0 { |
| 162 | return nil |
| 163 | } |
| 164 | |
| 165 | r.records = make(map[string]map[int32]Records) |
| 166 | for i := 0; i < topicCount; i++ { |
| 167 | topic, err := pd.getString() |
| 168 | if err != nil { |
| 169 | return err |
| 170 | } |
| 171 | partitionCount, err := pd.getArrayLength() |
| 172 | if err != nil { |
| 173 | return err |
| 174 | } |
| 175 | r.records[topic] = make(map[int32]Records) |
| 176 | |
| 177 | for j := 0; j < partitionCount; j++ { |
| 178 | partition, err := pd.getInt32() |
| 179 | if err != nil { |
| 180 | return err |
| 181 | } |
| 182 | size, err := pd.getInt32() |
| 183 | if err != nil { |
| 184 | return err |
| 185 | } |
| 186 | recordsDecoder, err := pd.getSubset(int(size)) |
| 187 | if err != nil { |
| 188 | return err |
| 189 | } |
| 190 | var records Records |
| 191 | if err := records.decode(recordsDecoder); err != nil { |
| 192 | return err |
| 193 | } |
| 194 | r.records[topic][partition] = records |
| 195 | } |
| 196 | } |
| 197 | |
| 198 | return nil |
| 199 | } |
| 200 | |
| 201 | func (r *ProduceRequest) key() int16 { |
| 202 | return 0 |
| 203 | } |
| 204 | |
| 205 | func (r *ProduceRequest) version() int16 { |
| 206 | return r.Version |
| 207 | } |
| 208 | |
| 209 | func (r *ProduceRequest) requiredVersion() KafkaVersion { |
| 210 | switch r.Version { |
| 211 | case 1: |
| 212 | return V0_9_0_0 |
| 213 | case 2: |
| 214 | return V0_10_0_0 |
| 215 | case 3: |
| 216 | return V0_11_0_0 |
| 217 | default: |
| 218 | return MinVersion |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | func (r *ProduceRequest) ensureRecords(topic string, partition int32) { |
| 223 | if r.records == nil { |
| 224 | r.records = make(map[string]map[int32]Records) |
| 225 | } |
| 226 | |
| 227 | if r.records[topic] == nil { |
| 228 | r.records[topic] = make(map[int32]Records) |
| 229 | } |
| 230 | } |
| 231 | |
| 232 | func (r *ProduceRequest) AddMessage(topic string, partition int32, msg *Message) { |
| 233 | r.ensureRecords(topic, partition) |
| 234 | set := r.records[topic][partition].MsgSet |
| 235 | |
| 236 | if set == nil { |
| 237 | set = new(MessageSet) |
| 238 | r.records[topic][partition] = newLegacyRecords(set) |
| 239 | } |
| 240 | |
| 241 | set.addMessage(msg) |
| 242 | } |
| 243 | |
| 244 | func (r *ProduceRequest) AddSet(topic string, partition int32, set *MessageSet) { |
| 245 | r.ensureRecords(topic, partition) |
| 246 | r.records[topic][partition] = newLegacyRecords(set) |
| 247 | } |
| 248 | |
| 249 | func (r *ProduceRequest) AddBatch(topic string, partition int32, batch *RecordBatch) { |
| 250 | r.ensureRecords(topic, partition) |
| 251 | r.records[topic][partition] = newDefaultRecords(batch) |
| 252 | } |