Encryption support ext consumer #612
Conversation
- use base crypto package for encryption
- move it to Consumer MR
…Fanatics/pulsar-client-go into encryption-support-ext-producer
pulsar/impl_message.go
Outdated
| @@ -215,6 +233,7 @@ type message struct { | |||
| replicatedFrom string | |||
| redeliveryCount uint32 | |||
| schema Schema | |||
| encryptionContext EncryptionContext | |||
There was a problem hiding this comment.
Let's make this a pointer
pulsar/internal/crypto/decryptor.go
Outdated
| // Decryptor support decrypting of message | ||
| type Decryptor interface { | ||
| Decrypt(payload []byte, msgID *pb.MessageIdData, msgMetadata *pb.MessageMetadata) ([]byte, error) | ||
| CryptoFailureAction() int |
There was a problem hiding this comment.
Why is this needed? I think the Decryptor should just have Decrypt and return an error. The caller of this can then figure out what to do with the error.
pulsar/message.go
Outdated
|
|
||
| // GetEncryptionContext get the ecryption context of message | ||
| // It will be used by the application to parse undecrypted message | ||
| GetEncryptionContext() EncryptionContext |
There was a problem hiding this comment.
Let's return a pointer or interface
| pc.log.Error(err) | ||
| switch pc.decryptor.CryptoFailureAction() { | ||
| case crypto.ConsumerCryptoFailureActionFail: | ||
| pc.log.Errorf("consuming message failed due to decryption err :%v", err) |
There was a problem hiding this comment.
The java clients add this to the unacked message tracker do we need to do the same?
There was a problem hiding this comment.
Let me take a look at this..
There was a problem hiding this comment.
I don't think we have this so may we just need to ack the message so it's not resent?
There was a problem hiding this comment.
yeah, but i don't think it is good idea to do ack.
Let's say user wants to consume this message again by providing proper crypto configuration, if we do ack, then he may not be able to consume.
There was a problem hiding this comment.
So, I think it's better to resend the message than doing ack.
Any other thoughts here ??
There was a problem hiding this comment.
Yes, we can do nack. I'll push changes with nack on failure.
There was a problem hiding this comment.
@cckellogg I've made the changes
There was a problem hiding this comment.
Looks good to me will you please fix the CI issue.
There was a problem hiding this comment.
thanks @cckellogg
I don't think CI is failing due to last commit.
I see most of the open PR's are failing with the same issue as below.
TestNamespaceTopicsNamespaceDoesNotExit (56.25s) client_impl_test.go:387: Error Trace: client_impl_test.go:387 Error: Expected nil, but got: &errors.errorString{s:"server error: AuthorizationError: Exception occurred while trying to authorize GetTopicsOfNamespace"} Test: TestNamespaceTopicsNamespaceDoesNotExit
And last commit do not have any changes that effects client_impl.go. More likely this has something to do with broker config that CI uses for running test cases.
please let me know If I'm wrong here.
There was a problem hiding this comment.
@cckellogg looks like the CI is passed now.
Please do suggest if there are any other improvements on this PR.
pulsar/consumer_partition.go
Outdated
|
|
||
| // error decrypting the payload | ||
| if err != nil { | ||
| pc.log.Error(err) |
There was a problem hiding this comment.
remove this and add a log under ConsumerCryptoFailureActionDiscard so there is more context.
pulsar/consumer_partition.go
Outdated
| replicationClusters: msgMeta.GetReplicateTo(), | ||
| replicatedFrom: msgMeta.GetReplicatedFrom(), | ||
| redeliveryCount: response.GetRedeliveryCount(), | ||
| encryptionContext: createEncryptionContext(msgMeta), |
There was a problem hiding this comment.
Does the encryptionContext need to be added to messages that don't fail?
There was a problem hiding this comment.
I don't think there is a need to add this.
There was a problem hiding this comment.
So the encryption context is only needed for failed messages?
pulsar/consumer_partition.go
Outdated
| case crypto.ConsumerCryptoFailureActionConsume: | ||
| pc.log.Warnf("consuming encrypted message due to error in decryption :%v", err) | ||
| messages = append(messages, &message{ | ||
| publishTime: timeFromUnixTimestampMillis(msgMeta.GetPublishTime()), |
There was a problem hiding this comment.
Nit. we can use create a new variable here and move messages := make([]*message, 0) back to where it was
messages := []*message{
{
},
}
Breakdown of PR #552
This PR includes the encryption/decryption changes at consumer side.