ADL Gen2 sink: x-ms-range-get-content-md5 header error

[christineyost]

Stream Reactor version: 9.0.2
Kafka Connect version: confluentinc/cp-kafka-connect 7.9.2

I’m trying to use the Azure Data Lake sink connector. It seems to authenticate and connect to Azure Data Lake just fine, but basically as soon as it starts running I get an error about an invalid header value:

[2025-08-01 18:36:27,803] ERROR Status code 400, “InvalidHeaderValueThe value for one of the HTTP headers is not in the correct format.RequestId:…Time:2025-08-01T18:36:27.7435817Zx-ms-range-get-content-md5true” (com.azure.storage.file.datalake.DataLakeFileClient)

This header doesn’t seem to be controlled by any of the connector settings. Is there something I can do to get past this error?

Here is the entire stack trace:

org.apache.kafka.connect.errors.ConnectException: error loading file (.indexes/.locks/Topic(__event_bus_avro_canary)/0.lock) Status code 400, "<?xml version="1.0" encoding="utf-8"?><Error><Code>InvalidHeaderValue</Code><Message>The value for one of the HTTP headers is not in the correct format.RequestId:c82f0b65-501e-00d7-5f0f-030d54000000
Time:2025-08-01T18:10:26.5830568Z</Message><HeaderName>x-ms-range-get-content-md5</HeaderName><HeaderValue>true</HeaderValue></Error>"
    at io.lenses.streamreactor.connect.cloud.common.sink.CloudSinkTask.handleErrors(CloudSinkTask.scala:114)
    at io.lenses.streamreactor.connect.cloud.common.sink.CloudSinkTask.open(CloudSinkTask.scala:261)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.openPartitions(WorkerSinkTask.java:661)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.access$1300(WorkerSinkTask.java:78)
    at org.apache.kafka.connect.runtime.WorkerSinkTask$HandleRebalance.onPartitionsAssigned(WorkerSinkTask.java:752)
    at org.apache.kafka.clients.consumer.internals.ConsumerRebalanceListenerInvoker.invokePartitionsAssigned(ConsumerRebalanceListenerInvoker.java:65)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:425)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:504)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:415)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:511)
    at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.updateAssignmentMetadataIfNeeded(ClassicKafkaConsumer.java:657)
    at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.poll(ClassicKafkaConsumer.java:616)
    at org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer.poll(ClassicKafkaConsumer.java:596)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:874)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.pollConsumer(WorkerSinkTask.java:498)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:340)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:247)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:216)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:226)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:281)
    at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:238)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
    at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.base/java.lang.Thread.run(Unknown Source)
Caused by: com.azure.storage.file.datalake.models.DataLakeStorageException: Status code 400, "<?xml version="1.0" encoding="utf-8"?><Error><Code>InvalidHeaderValue</Code><Message>The value for one of the HTTP headers is not in the correct format.
RequestId:c82f0b65-501e-00d7-5f0f-030d54000000
Time:2025-08-01T18:10:26.5830568Z</Message><HeaderName>x-ms-range-get-content-md5</HeaderName><HeaderValue>true</HeaderValue></Error>"
    at com.azure.storage.file.datalake.implementation.util.DataLakeImplUtils.transformSingleBlobStorageException(DataLakeImplUtils.java:46)
    at com.azure.storage.file.datalake.implementation.util.DataLakeImplUtils.transformBlobStorageException(DataLakeImplUtils.java:29)
    at com.azure.storage.file.datalake.implementation.util.DataLakeImplUtils.returnOrConvertException(DataLakeImplUtils.java:53)
    at com.azure.storage.file.datalake.DataLakeFileClient.readWithResponse(DataLakeFileClient.java:1093)
    at io.lenses.streamreactor.connect.datalake.storage.DatalakeStorageInterface.$anonfun$getBlobAsStringAndEtag$2(DatalakeStorageInterface.scala:291)
    at scala.util.Using$.resource(Using.scala:296)
    at io.lenses.streamreactor.connect.datalake.storage.DatalakeStorageInterface.$anonfun$getBlobAsStringAndEtag$1(DatalakeStorageInterface.scala:282)
    at scala.util.Try$.apply(Try.scala:217)
    at io.lenses.streamreactor.connect.datalake.storage.DatalakeStorageInterface.getBlobAsStringAndEtag(DatalakeStorageInterface.scala:282)
    at io.lenses.streamreactor.connect.cloud.common.storage.StorageInterface.getBlobAsObject(StorageInterface.scala:80)
    at io.lenses.streamreactor.connect.cloud.common.storage.StorageInterface.getBlobAsObject$(StorageInterface.scala:72)
    at io.lenses.streamreactor.connect.datalake.storage.DatalakeStorageInterface.getBlobAsObject(DatalakeStorageInterface.scala:51)
    at io.lenses.streamreactor.connect.cloud.common.sink.seek.IndexManagerV2.tryOpen(IndexManagerV2.scala:206)
    at io.lenses.streamreactor.connect.cloud.common.sink.seek.IndexManagerV2.$anonfun$open$5(IndexManagerV2.scala:116)
    at scala.util.Either.flatMap(Either.scala:360)
    at io.lenses.streamreactor.connect.cloud.common.sink.seek.IndexManagerV2.open(IndexManagerV2.scala:115)
    at io.lenses.streamreactor.connect.cloud.common.sink.seek.IndexManagerV2.$anonfun$open$2(IndexManagerV2.scala:92)
    at cats.effect.IOFiber.runLoop(IOFiber.scala:357)
    at cats.effect.IOFiber.execR(IOFiber.scala:1397)
    at cats.effect.IOFiber.run(IOFiber.scala:122)
    at cats.effect.unsafe.WorkerThread.lookForWork$1(WorkerThread.scala:545)
    at cats.effect.unsafe.WorkerThread.run(WorkerThread.scala:858)

[christineyost]

Assuming the scala DataLakeFileClient.readWithResponse(...) method matches the Java one, I think the issue is that the connector is asking Azure to return the contentMD5 for the specified FileRange, but it's not specifying a FileRange. I think the true in this call should be false, or else the second parameter should be non-null.

I tried this out with a small Java script locally. When I change the true to false I get file not found instead of this header error.

FileReadResponse response =
        client
            .getFileSystemClient("fileSystem")
            .getFileClient(
                "path/to/file")
            .readWithResponse(bao, null, null, null, false, null, Context.NONE);

[stheppi]

Hi @christineyost , would you be willing to open a PR?

[christineyost]

Created a PR for this.

[ddzz]

@stheppi How long would it take to get this fix merged and released? And would it be included in v10? Thanks.