| 216 | } |
| 217 | |
| 218 | func (c *Config) checkAndFix() (err error) { |
| 219 | // check topic cfg |
| 220 | if c.MQ.BlobDeleteTopic == "" || c.MQ.ShardRepairTopic == "" || c.MQ.ShardRepairPriorityTopic == "" { |
| 221 | return ErrIllegalTopic |
| 222 | } |
| 223 | |
| 224 | if c.MQ.BlobDeleteTopic == c.MQ.ShardRepairTopic || c.MQ.BlobDeleteTopic == c.MQ.ShardRepairPriorityTopic { |
| 225 | return ErrIllegalTopic |
| 226 | } |
| 227 | defaulter.Equal(&c.HeartbeatIntervalS, defaultHeartbeatIntervalS) |
| 228 | defaulter.Equal(&c.HeartbeatTicks, defaultHeartbeatTicks) |
| 229 | defaulter.Equal(&c.ExpiresTicks, defaultExpiresTicks) |
| 230 | defaulter.LessOrEqual(&c.Clustermgr.Config.ClientTimeoutMs, defaultTimeoutMS) |
| 231 | defaulter.LessOrEqual(&c.MQ.MsgSender.TimeoutMs, defaultTimeoutMS) |
| 232 | if c.MQ.Version != "" { |
| 233 | kafkaVersion, err := sarama.ParseKafkaVersion(c.MQ.Version) |
| 234 | if err != nil { |
| 235 | return ErrIllegalKafka |
| 236 | } |
| 237 | kafka.DefaultKafkaVersion = kafkaVersion |
| 238 | } |
| 239 | return nil |
| 240 | } |