Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix deleteRecord #1425

Merged
merged 1 commit into from
Jul 20, 2019
Merged

fix deleteRecord #1425

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 40 additions & 19 deletions admin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,29 +374,50 @@ func (ca *clusterAdmin) DeleteRecords(topic string, partitionOffsets map[int32]i
if topic == "" {
return ErrInvalidTopic
}

topics := make(map[string]*DeleteRecordsRequestTopic)
topics[topic] = &DeleteRecordsRequestTopic{PartitionOffsets: partitionOffsets}
request := &DeleteRecordsRequest{
Topics: topics,
Timeout: ca.conf.Admin.Timeout,
}

b, err := ca.Controller()
if err != nil {
return err
partitionPerBroker := make(map[*Broker][]int32)
for partition := range partitionOffsets {
broker, err := ca.client.Leader(topic, partition)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

formatting looks strange, can you please run gofmt again?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

gofmt again but nothing changed. Can you please tell me what's the format problem. I'll fix it manually.

if err != nil {
return err
}
if _, ok := partitionPerBroker[broker]; ok {
partitionPerBroker[broker] = append(partitionPerBroker[broker], partition)
} else {
partitionPerBroker[broker] = []int32{partition}
}
}
errs := make([]error, 0)
for broker, partitions := range partitionPerBroker {
topics := make(map[string]*DeleteRecordsRequestTopic)
recordsToDelete := make(map[int32]int64)
for _, p := range partitions {
recordsToDelete[p] = partitionOffsets[p]
}
topics[topic] = &DeleteRecordsRequestTopic{PartitionOffsets: recordsToDelete}
request := &DeleteRecordsRequest{
Topics: topics,
Timeout: ca.conf.Admin.Timeout,
}

rsp, err := b.DeleteRecords(request)
if err != nil {
return err
rsp, err := broker.DeleteRecords(request)
if err != nil {
errs = append(errs, err)
} else {
deleteRecordsResponseTopic, ok := rsp.Topics[topic]
if !ok {
errs = append(errs, ErrIncompleteResponse)
} else {
for _, deleteRecordsResponsePartition := range deleteRecordsResponseTopic.Partitions {
if deleteRecordsResponsePartition.Err != ErrNoError {
errs = append(errs, errors.New(deleteRecordsResponsePartition.Err.Error()))
}
}
}
}
}

_, ok := rsp.Topics[topic]
if !ok {
return ErrIncompleteResponse
if len(errs) > 0 {
return ErrDeleteRecords{MultiError{&errs}}
}

//todo since we are dealing with couple of partitions it would be good if we return slice of errors
//for each partition instead of one error
return nil
Expand Down
87 changes: 82 additions & 5 deletions admin_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,13 +333,17 @@ func TestClusterAdminCreatePartitionsWithoutAuthorization(t *testing.T) {
}

func TestClusterAdminDeleteRecords(t *testing.T) {
topicName := "my_topic"
seedBroker := NewMockBroker(t, 1)
defer seedBroker.Close()

seedBroker.SetHandlerByMap(map[string]MockResponse{
"MetadataRequest": NewMockMetadataResponse(t).
SetController(seedBroker.BrokerID()).
SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
SetLeader(topicName, 1, 1).
SetLeader(topicName, 2, 1).
SetLeader(topicName, 3, 1),
"DeleteRecordsRequest": NewMockDeleteRecordsResponse(t),
})

Expand All @@ -350,12 +354,70 @@ func TestClusterAdminDeleteRecords(t *testing.T) {
t.Fatal(err)
}

partitionOffsetFake := make(map[int32]int64)
partitionOffsetFake[4] = 1000
errFake := admin.DeleteRecords(topicName, partitionOffsetFake)
if errFake == nil {
t.Fatal(err)
}

partitionOffset := make(map[int32]int64)
partitionOffset[1] = 1000
partitionOffset[2] = 1000
partitionOffset[3] = 1000

err = admin.DeleteRecords(topicName, partitionOffset)
if err != nil {
t.Fatal(err)
}

err = admin.Close()
if err != nil {
t.Fatal(err)
}
}

func TestClusterAdminDeleteRecordsWithInCorrectBroker(t *testing.T) {
topicName := "my_topic"
seedBroker := NewMockBroker(t, 1)
secondBroker := NewMockBroker(t, 2)
defer seedBroker.Close()
defer secondBroker.Close()

seedBroker.SetHandlerByMap(map[string]MockResponse{
"MetadataRequest": NewMockMetadataResponse(t).
SetController(seedBroker.BrokerID()).
SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
SetBroker(secondBroker.Addr(), secondBroker.brokerID).
SetLeader(topicName, 1, 1).
SetLeader(topicName, 2, 1).
SetLeader(topicName, 3, 2),
"DeleteRecordsRequest": NewMockDeleteRecordsResponse(t),
})

secondBroker.SetHandlerByMap(map[string]MockResponse{
"MetadataRequest": NewMockMetadataResponse(t).
SetController(seedBroker.BrokerID()).
SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
SetBroker(secondBroker.Addr(), secondBroker.brokerID).
SetLeader(topicName, 1, 1).
SetLeader(topicName, 2, 1).
SetLeader(topicName, 3, 2),
"DeleteRecordsRequest": NewMockDeleteRecordsResponse(t),
})

config := NewConfig()
config.Version = V1_0_0_0
admin, err := NewClusterAdmin([]string{seedBroker.Addr()}, config)
if err != nil {
t.Fatal(err)
}
partitionOffset := make(map[int32]int64)
partitionOffset[1] = 1000
partitionOffset[2] = 1000
partitionOffset[3] = 1000

err = admin.DeleteRecords("my_topic", partitionOffset)
err = admin.DeleteRecords(topicName, partitionOffset)
if err != nil {
t.Fatal(err)
}
Expand All @@ -367,13 +429,17 @@ func TestClusterAdminDeleteRecords(t *testing.T) {
}

func TestClusterAdminDeleteRecordsWithDiffVersion(t *testing.T) {
topicName := "my_topic"
seedBroker := NewMockBroker(t, 1)
defer seedBroker.Close()

seedBroker.SetHandlerByMap(map[string]MockResponse{
"MetadataRequest": NewMockMetadataResponse(t).
SetController(seedBroker.BrokerID()).
SetBroker(seedBroker.Addr(), seedBroker.BrokerID()),
SetBroker(seedBroker.Addr(), seedBroker.BrokerID()).
SetLeader(topicName, 1, 1).
SetLeader(topicName, 2, 1).
SetLeader(topicName, 3, 1),
"DeleteRecordsRequest": NewMockDeleteRecordsResponse(t),
})

Expand All @@ -389,10 +455,21 @@ func TestClusterAdminDeleteRecordsWithDiffVersion(t *testing.T) {
partitionOffset[2] = 1000
partitionOffset[3] = 1000

err = admin.DeleteRecords("my_topic", partitionOffset)
if err != ErrUnsupportedVersion {
err = admin.DeleteRecords(topicName, partitionOffset)
if !strings.HasPrefix(err.Error(), "kafka server: failed to delete records") {
t.Fatal(err)
}
deleteRecordsError, ok := err.(ErrDeleteRecords)

if !ok {
t.Fatal(err)
}

for _, err := range *deleteRecordsError.Errors {
if err != ErrUnsupportedVersion {
t.Fatal(err)
}
}

err = admin.Close()
if err != nil {
Expand Down
22 changes: 22 additions & 0 deletions errors.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,28 @@ func (err ConfigurationError) Error() string {
// See https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-ErrorCodes
type KError int16

// MultiError is used to contain multi error.
type MultiError struct {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's the reason to add this?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently the delete record request is sent to leader by leader. Thus we may receive multi error response and probably the errors are not the same. I use MultiError to contain the array of error.

Errors *[]error
}

func (mErr MultiError) Error() string {
var errString = ""
for _, err := range *mErr.Errors {
errString += err.Error() + ","
}
return errString
}

// ErrDeleteRecords is the type of error returned when fail to delete the required records
type ErrDeleteRecords struct {
MultiError
}

func (err ErrDeleteRecords) Error() string {
return "kafka server: failed to delete records " + err.MultiError.Error()
}

// Numeric error codes returned by the Kafka server.
const (
ErrNoError KError = 0
Expand Down