Added kafka consumer logic #296
Conversation
| var lambdaMgr *lambda.LambdaMgr | ||
| var once sync.Once | ||
|
|
||
| // LambdaManager is now a singleton, one per worker. This is because lambda manager |
There was a problem hiding this comment.
I agree that we can't create multiple instances in the same process right now because they would collide in terms of directories created. However, is it better to make it a singleton, or to eventually modify the manager so that we can have multiple ones? I think this is a longer discussion, and it shouldn't be part of this PR because it will slow things down.
There was a problem hiding this comment.
For this PR, what would you recommend? If the worker is supposed to support both Kafka and HTTP/Cron triggers at the same time, this change or something similar would be needed for even bare minimum functionality
| @@ -149,3 +149,20 @@ Invoke your lambda with `curl` (the result should be the same as the POST body): | |||
| ``` | |||
| curl -X POST localhost:5000/run/echo -d '{"hello": "world"}' | |||
| ``` | |||
|
|
|||
There was a problem hiding this comment.
Thanks for documenting it! Kafka isn't really a quick-start kind of topic though -- this is better as its own markdown file.
| "github.com/twmb/franz-go/pkg/kgo" | ||
| ) | ||
|
|
||
| type KafkaServer struct { |
There was a problem hiding this comment.
I think things will get messy if we eventually have a bunch of different Server types wrapping the same lambdaMgr. Better to have a top-level struct (maybe this? https://github.com/open-lambda/open-lambda/blob/main/src/worker/event/lambdaServer.go#L16) that has the lambdaMgr and references to other types, like a KafkaServer and HTTPServer for now.
| } | ||
|
|
||
| var data map[string]string | ||
| if err := json.NewDecoder(r.Body).Decode(&data); err != nil { |
There was a problem hiding this comment.
Let's not do a string => string map for the body type. We might eventually want non-string types. Let's have a JSON-serializable struct.
There was a problem hiding this comment.
On the boss side, Mandy is going to be parsing lambda config, from here: https://github.com/open-lambda/open-lambda/blob/main/src/common/lambdaConfig.go. So it probably makes sense if the information the boss passes to the worker uses the same structs that the boss reads from the lambda config.
| panic(err) | ||
| } | ||
| w := httptest.NewRecorder() | ||
| kafkaServer.lambdaMgr.Get(functionName).Invoke(w, r) |
There was a problem hiding this comment.
To minimize overhead, can we send the lambda function a whole batch at once?
| return req, nil | ||
| } | ||
|
|
||
| func KafkaInit(w http.ResponseWriter, r *http.Request) { |
There was a problem hiding this comment.
The boss might ask a worker to start or stop consuming. So maybe we can make the API like that? I think it's fine to have a TODO comment about what you will do for stopping in this PR (and leave the implementation to another one).
| if errs := fetches.Errors(); len(errs) > 0 { | ||
| // All errors are retried internally when fetching, but non-retriable errors are | ||
| // returned from polls so that users can notice and take action. | ||
| panic(fmt.Sprint(errs)) |
There was a problem hiding this comment.
I don't think we want to panic. If one user's function has a typo in the topic name, perhaps they would get an error, and then the whole server would go down, with the functions belonging to all other users too.
I do agree it's hard to know how to surface this. Perhaps we should make a call to a user's lambda function with error info so they can do what they want about it?
| return req, nil | ||
| } | ||
|
|
||
| func KafkaInit(w http.ResponseWriter, r *http.Request) { |
There was a problem hiding this comment.
Can we refactor this function into two? One for the HTTP side of things, and one for actually starting the goroutine and consuming messages.
Also, it probably makes sense as methods on KafkaServer, no functions that use a global.
No description provided.