feat: implement kafka topic create/delete api #315
Conversation
vchandela
force-pushed
the
vchandela-kafka-topic-create-delete-api
branch
2 times, most recently
from
493b990
to
2dca2ea
Compare
vchandela
force-pushed
the
vchandela-kafka-topic-create-delete-api
branch
2 times, most recently
from
044853d
to
5660915
Compare
…st; add partial unit test
…CreateDeleteTopics
…d Client(); add tests for duplicate topic creation and deleting non-existent topic
…ontroller unavailability
…ve unused ReplicationFactor field
…; extract error code for setting kafka error codes; set TopicInfosByName lowercase
vchandela
force-pushed
the
vchandela-kafka-topic-create-delete-api
branch
from
5660915
to
bec6a1d
Compare
| var resp kafkaprotocol.CreateTopicsResponse | ||
| r, err := conn.SendRequest(&req, kafkaprotocol.APIKeyCreateTopics, 5, &resp) | ||
| require.NoError(t, err) | ||
| createResp, ok := r.(*kafkaprotocol.CreateTopicsResponse) |
There was a problem hiding this comment.
Should be coordinator not available?
| Responses: make([]kafkaprotocol.DeleteTopicsResponseDeletableTopicResult, len(req.TopicNames)), | ||
| } | ||
| for tidx, topicName := range req.TopicNames { | ||
| var errMsg string |
There was a problem hiding this comment.
should be *string otherwise we are setting err message to pointer to empty string even when there is no error
There was a problem hiding this comment.
I think this comment is outdated as the code has changed a bit. Please let me know if this is still needed.
agent/kafka_handler.go
Outdated
| return common.UnknownError | ||
| } | ||
|
|
||
| func detectInvalidTopic(name string) error { |
There was a problem hiding this comment.
Thinking... it would be better to do this check in topicmeta.CreateTopic(), not here, can you move it?
There was a problem hiding this comment.
I think this is outdated.
…ion; modularise CreateTopics request handler
| GroupCoordinatorConf: group.NewConf(), | ||
| TxCoordinatorConf: tx.NewConf(), | ||
| MaxControllerClients: DefaultMaxControllerClients, | ||
| DefaultDefaultTopicRetentionTime: DefaultDefaultTopicRetentionTime, |
There was a problem hiding this comment.
Why is the member called "DefaultDefaultTopicRetentionTime"?
There was a problem hiding this comment.
ahhh...the member should be DefaultTopicRetentionTime and the constant should be DefaultDefaultTopicRetentionTime.
There was a problem hiding this comment.
Yes, the constant is the default of the default
| {"Invalid special characters", "topic@", int16(kafkaprotocol.ErrorCodeInvalidTopicException)}, | ||
| } | ||
| for _, tc := range testCases { | ||
| t.Run(tc.name, func(t *testing.T) { |
There was a problem hiding this comment.
It's nicer if you put this in a separate method
| return kafkaprotocol.ErrorCodeUnknownServerError | ||
| } | ||
|
|
||
| func isInvalidTopicName(name string) error { |
There was a problem hiding this comment.
As mentioned in a previous comment, this check should be done in topicmeta.Manager
There was a problem hiding this comment.
but Agent doesn't have topicmeta.Manager
type Agent struct {
lock sync.RWMutex
cfg Conf
started bool
transportServer transport.Server
kafkaServer *kafkaserver2.KafkaServer
tablePusher *pusher.TablePusher
batchFetcher *fetcher.BatchFetcher
controller *control.Controller
controlClientCache *control.ClientCache
membership ClusterMembership
compactionWorkersService *lsm.CompactionWorkerService
partitionHashes *parthash.PartitionHashes
fetchCache *fetchcache.Cache
groupCoordinator *group.Coordinator
txCoordinator *tx.Coordinator
topicMetaCache *topicmeta.LocalCache
manifold *membershipChangedManifold
partitionLeaders map[string]map[int]map[int]int32
clusterMembershipFactory ClusterMembershipFactory
}
There was a problem hiding this comment.
I'm not sure I understand what you mean.
I am suggesting the check that the topic name is valid should be in topicmeta.manager.CreateTopic - do you not think that is a better place?
There was a problem hiding this comment.
But don't worry about this - I will do it in a later PR
| Responses: make([]kafkaprotocol.DeleteTopicsResponseDeletableTopicResult, len(req.TopicNames)), | ||
| } | ||
| for tidx, topicName := range req.TopicNames { | ||
| var errMsg string |
Fixes #285