Filter operation is not working.

[himanshuankercloud]

We are currently setting up the MSK Connect pipeline to transfer data from a Kafka Topic to OpenSearch using the Aiven connector as a plugin. We have successfully created a topic and produced dummy data on that topic. Additionally, we've configured an MSK connector to read and write the data, successfully writing it into OpenSearch.

However, when attempting to filter the data based on a specific field, data is not being written to OpenSearch(It is also not creating the Index in OpenSearch). Below is the configuration for the MSK Connect:

connector.class=io.aiven.kafka.connect.opensearch.OpensearchSinkConnector
type.name=kafka-connect
connection.password=***********
transforms.Filter.type=io.aiven.kafka.connect.transforms.FilterByFieldValue$Value
transforms.Filter.field.name=EventId
tasks.max=1
topics=json-topic
connection.username=sample
transforms=Filter
bootstrap.servers="**************************"
key.ignore=true
schema.ignore=true
key.converter.schemas.enable=false
value.converter.schemas.enable=false
transforms.Filter.field.value=4690
transforms.Filter.field.value.matches=true
value.converter=org.apache.kafka.connect.json.JsonConverter
connection.url=""
key.converter=org.apache.kafka.connect.storage.StringConverter

It is not working. I am not getting any errors but the filter data is not written to OpenSearch. I checked the data in topic also it contains the event id 4690.

Can you please help us.

[jeqo]

Hi @himanshuankercloud.

It's hard to help troubleshooting this without an idea of how input messages look like. Could you share some sample input messages to see how this transformation may be applied?
You could also disable the SMT, and share how input and output messages look like. We can get some idea of what may be missing from these.

You could also look into the FilterByFieldValueTest tests to potentially reproduce the issue.

PS. I suggest to not copying real endpoints on public issues as it could lead to security issues.

[himanshuankercloud]

Hi Jorge, Please find the sample input for the topic and output as a message:- *You could also look into the FilterByFieldValueTest tests to potentially reproduce the issue*: can you explain how we can use it? I have attached the python script using which we are uploading the data. *Input:- * {"Id":"DkiLMaqgagcCVZ4","EventId":4663,"TimeStampEvent":"2021-07-19T15:28:01.1055284+00:00","TimeStamp":"2021-07-19T15:28:01.3095677+00:00","Machine":"win7x64.mydomain.local","Properties":{"subjectUserSid":"S-1-5-21-2482720086-1306900138-933716382-500","subjectDomainName":"MYDOMAIN","objectServer":"Security","objectName":"C:\\Windows\\System32\\rundll32.exe","accessList":"%%4421\n\t\t\t\t","processId":"0x16a0"},"EventRecordId":520668533,"ProcId":0,"ThreadId":0} {"Id":"pz1zkEXx4JV04","EventId":4688,"TimeStampEvent":"2021-07-19T15:28:01.1055284+00:00","TimeStamp":"2021-07-19T15:28:01.3256558+00:00","Machine":"win7x64.mydomain.local","Properties":{"subjectUserSid":"S-1-5-21-2482720086-1306900138-933716382-500","subjectDomainName":"MYDOMAIN","newProcessId":"0x17b0","tokenElevationType":"%%1936","commandLine":"rundll32 637"},"EventRecordId":520668534,"ProcId":0,"ThreadId":0} *Output:- * [image: image.png]

…

On Wed, Jan 24, 2024 at 12:28 AM Jorge Esteban Quilcate Otoya < ***@***.***> wrote: Hi @himanshuankercloud <https://github.com/himanshuankercloud>. It's hard to help troubleshooting this without an idea of how input messages look like. Could you share some sample input messages to see how this transformation may be applied? You could also disable the SMT, and share how input and output messages look like. We can get some idea of what may be missing from these. You could also look into the FilterByFieldValueTest tests to potentially reproduce the issue. PS. I suggest to not copying real endpoints on public issues as it could lead to security issues. — Reply to this email directly, view it on GitHub <#126 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AWTROQ2IF6DDOCRUMYNY4ZDYQAB5BAVCNFSM6AAAAABB4N337GVHI2DSMVQWIX3LMV43OSLTON2WKQ3PNVWWK3TUHMYTSMBWG4ZTANRUGI> . You are receiving this because you were mentioned.Message ID: <Aiven-Open/transforms-for-apache-kafka-connect/issues/126/1906730642@ github.com>

-- Thanks & Best Regards, *Himanshu Shukla * *Senior Data Engineer* *Ankercloud Technologies Pvt. Ltd.* 91Springboard, Creaticity Mall, Off, Airport Rd, opposite Golf Course, Shastrinagar, Yerawada, Pune, Maharashtra, India, 411006. *M *+91-(8920773869) *www.ankercloud.com <http://www.ankercloud.com/>* ***@***.*** ***@***.***>* Founder & CEO : Santhosh Jayaprakash

[himanshuankercloud]

@jeqo
Can you please share your feedback.
Its urgent. your help will be very useful for us.
Thanks in Advance.

[jeqo]

@himanshuankercloud thanks for sharing details on how to reproduce.

Can confirm that this is an issue. Seems to be a bug on how value schema is inferred on the Connect schema library and how validation happens on the transform:

• Inferred schema from input message is INT16 (integer)
• But inferred schema parsed from expected value is INT8 (short)

This cause the filter test to always fail. See test https://github.com/Aiven-Open/transforms-for-apache-kafka-connect/pull/127/files

I see a couple of alternatives on how to fix:

• Add a new config to set the type for expected value. In this case INT16.
• If value is numeric, then validate against every type possible. In this case, instead of only testing input value as INT16, also test against INT8 and INT32.

Given this knowledge of mismatching numeric types, I wonder if you could apply some workaround with other SMTs to cast the value properly and apply the filter condition correctly, e.g. set the EventId as Key, then cast to INT16, and apply filter on the Key instead of Value. Maybe this would be enough to fix your use-case while this bug is getting fixed.