Kafka trigger #302
Kafka trigger #302
Conversation
src/boss/lambdastore/store.go
Outdated
|
|
||
| err = s.eventManager.Register(funcName, cfg.Triggers) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| entry.Lock.Unlock() |
There was a problem hiding this comment.
If it covers to the end, why not use defer? If you return err above, the lock is never released.
src/common/lambdaConfig.go
Outdated
| @@ -33,7 +33,12 @@ type CronTrigger struct { | |||
| Schedule string `yaml:"schedule"` // Cron schedule (e.g., "*/5 * * * *") | |||
| } | |||
|
|
|||
| // TODO: add KafkaTrigger struct | |||
| type KafkaTrigger struct { | |||
| Bootstrap_servers []string `yaml:"bootstrap_servers" json:"bootstrap_servers"` // e.g., ["localhost:9092"] | |||
There was a problem hiding this comment.
We normally use camel case, like BootstrapServers
src/boss/event/kafka_manager.go
Outdated
|
|
||
| type KafkaManager struct { | ||
| workerPool *cloudvm.WorkerPool // to forward the req to worker | ||
| mapLock sync.Mutex |
There was a problem hiding this comment.
should we just call it lock? It doesn't seem like we're just using it to protect changes to the map, but to the entries as well given we hold it for the entire time the functions run.
src/boss/event/kafka_manager.go
Outdated
|
|
||
| data, err := json.Marshal(trigger) | ||
| if err != nil { | ||
| log.Printf("[KafkaManager] Failed to marshal Kafka trigger for %s: %v", functionName, err) |
src/boss/event/kafka_manager.go
Outdated
| continue | ||
| } | ||
|
|
||
| req := httptest.NewRequest(http.MethodPost, "/kafka-init/"+functionName, bytes.NewBuffer(data)) |
There was a problem hiding this comment.
I don't know what httptest does, but is what we're doing related to testing? Why not a regular HTTP request, like here: https://github.com/open-lambda/open-lambda/blob/main/src/worker/lambda/lambdaInstance.go#L145?
src/boss/event/kafka_manager.go
Outdated
|
|
||
| data, err := json.Marshal(trigger) | ||
| if err != nil { | ||
| log.Printf("[KafkaManager] Failed to marshal Kafka trigger for %s: %v", functionName, err) |
There was a problem hiding this comment.
Is it right that we have errors here but the function returns no error?
No description provided.