Allow Special Character Handling

[e160842]

Describe the issue
Using this programming to transfer data fails for data that includes Special Chars ( $%^& etc.) , especially when those Special Char's are in the Header (field names) of the data :(

Describe the solution you'd like
Enable this programming to 'handle' Special Chars

Describe alternatives you've considered
we have to create extra step to "cleanse" Special Chars before we use this programming - not efficient.

Additional context
Please let me know if any clarifications needed :)

[fhussonnois]

Hi @e160842, could you please provide the connector's configuration you used to process data ? Thanks

[e160842]

@fhussonnois - thank you for your support - below is sample connector config, with proprietary data obscured - Please let me know if any parts are unclear . .. .

apiVersion: platform.confluent.io/v1beta1
kind: Connector
metadata:
name: XXXXXXXXXXXXXXXXXXXXX
namespace: ${namespace}
spec:
name: XXXXXXXXXXXXXXXXXXXXXXXXXX
class: "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector"
taskMax: 1
restartPolicy:
type: "OnFailure"
maxRetry: 10
connectRest:
endpoint: https://connector.${nspace}.xxx.cluster.local:8XXX
authentication:
type: bearer
bearer:
secretRef: v2-connectors-apikeys
configs:
errors.log.enable: "true"
errors.log.include.messages: "true"
connector.class: "io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector"
tasks.max: 1
fs.listing.class: "io.streamthoughts.kafka.connect.filepulse.fs.AmazonS3FileSystemListing"
aws.credentials.provider.class: "com.amazonaws.auth.InstanceProfileCredentialsProvider"
aws.s3.region: "us-east-1"
aws.s3.bucket.name: "XXXXXXXXXXXXXXXXXXXXXX"
aws.s3.bucket.prefix: "XXXXXXXXXXXXXXXXXX"
aws.s3.default.object.storage.class: "STANDARD"
fs.listing.filters: "io.streamthoughts.kafka.connect.filepulse.fs.filter.RegexFileListFilter"
fs.listing.interval.ms: "300000"
tasks.reader.class: "io.streamthoughts.kafka.connect.filepulse.fs.reader.AmazonS3RowFileInputReader"
topic: "${env}.stage.XXXXXXXXXXXXXXXX"
offset.policy.class: "io.streamthoughts.kafka.connect.filepulse.offset.DefaultSourceOffsetPolicy"
filters: "ParseCSVLine,SetKey,ProvideId,AnnotateWithType"
filters.ParseCSVLine.type: "io.streamthoughts.kafka.connect.filepulse.filter.CSVFilter"
filters.ParseCSVLine.trim.column: "true"
filters.ParseCSVLine.separator: "|"
filters.ParseCSVLine.columns:

"AMS: # Alternative - Hybrid Tru|AMS: % Alternative - Hybrid Tru|AMS: $ Electric/Hybrid Cars|AMS: % Electric/Hybrid Cars|AMS: # Alternative - Natural_Gas|AMS: Alternative Power>Natural Ga|Property/Realty: Home Stories|Property/Realty: Home Bath|Property/Realty: Home Bedrooms|Property/Realty: Home Total Rooms|Property/Realty: Home Exterior Wall T|BehaviorBank: Hi-tech owner|BehaviorBank: Internet/online subscri|SRVY:HH Acty/Int:Socl Caus/Con:Enviro"

filters.SetKey.type: "io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter"
filters.SetKey.field: "$key"
filters.SetKey.value: "$value.Number"
filters.AnnotateWithType.type: "io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter"
filters.AnnotateWithType.field: "$._t"
filters.AnnotateWithType.value: "XXXXXXXXXX"
filters.ProvideId.type: "io.streamthoughts.kafka.connect.filepulse.filter.AppendFilter"
filters.ProvideId.field: "$._id"
filters.ProvideId.value: "$value.Number"
fs.cleanup.policy.class: "io.streamthoughts.kafka.connect.filepulse.fs.clean.AmazonS3MoveCleanupPolicy"
fs.cleanup.policy.triggered.on: "COMMITTED"
fs.cleanup.policy.move.success.aws.bucket.name: "SUCCESS_XXXXXXXX"
fs.cleanup.policy.move.success.aws.prefix.path: "SUCCESS/processed"
fs.cleanup.policy.move.failure.aws.bucket.name: "FAILURE_XXXXXX"
fs.cleanup.policy.move.failure.aws.prefix.path: "FAILURE/error"	
file.filter.regex.pattern: ".*\\.csv|.*\\.txt$"
offset.attributes.string: "name"
skip.headers: "1"
tasks.file.status.storage.class: "io.streamthoughts.kafka.connect.filepulse.state.KafkaFileObjectStateBackingStore"
tasks.file.status.storage.bootstrap.servers: "kafka.${namespace}.svc.cluster.local:9071"
tasks.file.status.storage.topic: "${env}.prep.XXXXXXXXData-Status"
tasks.file.status.storage.topic.partitions: "1"
tasks.file.status.storage.topic.replication.factor: "3"
key.converter: "org.apache.kafka.connect.storage.StringConverter"
key.converter.schemas.enable: "false"
value.converter: "org.apache.kafka.connect.json.JsonConverter"
value.converter.schemas.enable: "false"
tasks.file.status.storage.producer.security.protocol : "SASL_SSL"
tasks.file.status.storage.producer.ssl.endpoint.identification.algorithm : "https"
tasks.file.status.storage.producer.sasl.mechanism : "PLAIN"
tasks.file.status.storage.producer.sasl.jaas.config : "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$${file:/mnt/secrets/v2-connectors-apikeys/plain.txt:username}\" password =\"$${file:/mnt/secrets/apikeys/plain.txt:password}\";"
tasks.file.status.storage.producer.request.timeout.ms : "20000"
tasks.file.status.storage.consumer.security.protocol : "SASL_SSL"
tasks.file.status.storage.consumer.ssl.endpoint.identification.algorithm : "https"
tasks.file.status.storage.consumer.sasl.mechanism : "PLAIN"
tasks.file.status.storage.consumer.sasl.jaas.config : "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"$${file:/mnt/secretkeys/plain.txt:username}\" password=\"$${file:/mnt/secretkeys/plain.txt:password}\";"
tasks.file.status.storage.consumer.request.timeout.ms : "20000"
principal.service.name: "$${file:/mnt/secrets/connector-configs/s3-src:username}"
principal.service.password: "$${file:/mnt/secrets/src:password}"

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.