This browser is no longer supported.
Upgrade to Microsoft Edge to take advantage of the latest features, security updates, and technical support.
![](data:image/svg+xml, %3Csvg xmlns='http://www.w3.org/2000/svg' height='64' class='font-weight-bold' style='font: 600 30.11764705882353px "SegoeUI", Arial' width='64'%3E%3Ccircle fill='hsl(128, 28.000000000000004%, 22%)' cx='32' cy='32' r='32' /%3E%3Ctext x='50%25' y='55%25' dominant-baseline='middle' text-anchor='middle' fill='%23FFF' %3EKS%3C/text%3E%3C/svg%3E)
I have a Kafka enabled azure eventhub where am publishing the data from my client application. To publish i use the confluent kafka dot net package.
Each message has a User header (message headers). The Eventhub has 10 partitions. Multiple tenants (5) data are published to this eventhub across the partition.
In user header, have the key called TenantId which holds the tenant id value.
Through Stream Analytics Job i want to filter the specific tenant from the User Header and push to different Output. You can consider pushing it to another eventhub as well.
I see we can us "GetMetadataPropertyValue" function to pass it. But always it returns null only. But am sure that I have header value which i verified from Service Bus explorer.
Even ASA extension in VS Code also I tried it tells that "Cannot edit in read-only editor".
Could you guide me the right path how to achieve this.
@K S Devadoss, Vishnu Kumar - Thanks for the question and using MS Q&A platform.
To filter the specific tenant from the User Header and push it to a different output using Azure Stream Analytics, you can use the GetMetadataPropertyValue function. However, it seems that you are facing issues with this function returning null values.
One possible reason for this issue could be that the User header is not being propagated to the metadata of the event in the Event Hub. To ensure that the User header is propagated to the metadata, you can use the Message.SetUserProperty method in the Confluent Kafka .NET package to set the User header as a user-defined property.
Once you have confirmed that the User header is being propagated to the metadata, you can use the GetMetadataPropertyValue function in your Stream Analytics query to extract the TenantId value from the User header. Here's an example query:
SELECT *
INTO output_eventhub
FROM input_eventhub
WHERE GetMetadataPropertyValue(User, 'TenantId') = 'your_tenant_id'
In this query, input_eventhub is the input event hub, output_eventhub is the output event hub, and your_tenant_id is the ID of the tenant that you want to filter for.
If you are still facing issues with the GetMetadataPropertyValue function, you can try using the GetArrayElements function to extract the User header as an array and then use the ArrayElement function to extract the TenantId value. Here's an example query:
SELECT *
INTO output_eventhub
FROM input_eventhub
WHERE ArrayElement(GetArrayElements(User), 0) = 'TenantId' AND ArrayElement(GetArrayElements(User), 1) = 'your_tenant_id'
In this query, input_eventhub is the input event hub, output_eventhub is the output event hub, and your_tenant_id is the ID of the tenant that you want to filter for.
I hope this helps you achieve your goal. Let me know if you have any further questions or concerns.
Thanks @PRADEEPCHEEKATLA-MSFT for the swift response.
I'm very sure that I have a user property in the message. Below is for reference.
I tried the below query but I see the error Expression is not supported : 'User'
SELECT * INTO[dera-telematics-ingress-1p] FROM [dera-telematics-ingress]
WHERE GetMetadataPropertyValue(User, 'TenantId') = '034e7f0d-3474-4415-90b9-92b2167733c7';
Using the ArrayElement approach I see the below error
Function 'arrayelement' is either not supported or not usable in this context. User defined function calls must start with "udf." prefix. Aggregate functions must be used together with a GROUP BY clause. Analytic functions must be used together with an OVER clause.
SELECT * INTO [dera-telematics-ingress-1p] FROM [dera-telematics-ingress]
WHERE ArrayElement(GetArrayElements(User), 0) = 'TenantId' AND ArrayElement(GetArrayElements(User), 1) = '034e7f0d-3474-4415-90b9-92b2167733c7'
Is it something am I missing?
Hello @K S Devadoss, Vishnu Kumar, using that GetMetatDataPropertyValue is likely the best way.
But always it returns null only
Notice that the Azure portal query editor cannot visualize user properties. There you will see null in the result table while there are values. Please use a test output like another eventhub or an Azure function to check the format outputted my your ASA query.
Thanks for the input @Sander van de Velde I will try that.
Hi @Sander van de Velde , I tried to do the same outputting the data from Eventhub to another by configuring Input and Output with the below query.
SELECT * INTO [OutputEgress] FROM [InputEH]
Source Eventhub has the Headers attached in my first comment. Here is for quick reference
But destination eventhub does not have the headers Job did not propagate the header. But I see the new headers which is created by the job.
Not sure how to get the headers and put it back in the headers of the target eventhub.
To be sure, can you try:
SELECT *
, GetMetadataPropertyValue(ehInput, '[User]') AS userprops
, GetMetadataPropertyValue(ehInput, 'EventHub') AS ehprops
INTO [OutputEgress] FROM [InputEH]
This outputs both the body, user-properties and system-properties of event hub messages.
This is documented here. Do you see any header values then?
Thanks @Sander van de Velde for the input. I tried this and I see the null values only in the portal.
But through Azure Stream Analytics extension in VS Code I could see the EventHub properties and User Properties.
But for user properties its has byte array value. Is there a way to covert the value from byte to string and filter in the Analytics query itself.
Hello @K S Devadoss, Vishnu Kumar, do you have an example of the byte array? I expect this is base64 encoded (with it with eg this tool). If so, try to decode it with a custom function.
The value is in UTF8 byte format.
I created a UDF in C# and ran the below query in VS Code with ASA extension
public static string GetStringFromUtf8(byte[] input)
{
try
{
return Encoding.UTF8.GetString(input);
}
catch (Exception e)
{
return e.Message;
}
}
SELECT * , udf.ByteToString(GetMetadataPropertyValue(InputEH, '[User].[CorrelationId]')) as converted, GetMetadataPropertyValue(InputEH, '[User]') AS userprops , GetMetadataPropertyValue(InputEH, 'EventHub') AS ehprops INTO [OutputEgress] FROM [InputEH]
But I got the below error caz function is expecting the varbinary(max)
So I tried to use the casting like below. But varabinary(max) it returns null and varchar(max) return System.Byte[].
SELECT *
, TRY_CAST(GetMetadataPropertyValue(InputEH, '[User].[CorrelationId]') as nvarchar(MAX)) as CorrelationIdVarChar
, TRY_CAST(GetMetadataPropertyValue(InputEH, '[User].[CorrelationId]') as varbinary(MAX)) as CorrelationIdVarBinary
, GetMetadataPropertyValue(InputEH, '[User]') AS userprops
, GetMetadataPropertyValue(InputEH, 'EventHub') AS ehprops
INTO [OutputEgress] FROM [InputEH]
Not sure what else I need to try to get the string out of the byte[]
Can you share an example of the returned values? Can you show the raw JSON being outputted?
Sure @Sander van de Velde . This is the example Input.txt file (Converted the json to txt file since json is not supported while uploading). And I could not able to export the data on the User properties and Eventhub properties. So pasted the content below from one message
| Eventhub Properties |
|---|
{"EventEnqueuedUtcTime":"2024-02-12T10:33:02.1200000Z","PartitionId":0,"Publisher":"","PartitionKey":"","SequenceNumber":24731,"Offset":"55836596448","IoTConnectionDeviceId":"","IoTAuthMethod":"","IoTAuthGenerationId":"","IoTEnqueueTime":"0001-01-01T00:00:00.0000000","IoTConnectionModuleId":"","IoTInterfaceName":"","IoTMessageSource":"","IoTDataSchema":"","IoTSubject":""} |
| User properties (Kafka Custom Headers) | |
|---|---|
This is was I could see while using the GetMetadataPropertyValue(InputEH, '[User]') AS userprops {"CorrelationId":"System.Byte[]","TenantId":"System.Byte[]","IsCompressed":"System.Byte[]","SourceKey":"System.Byte[]","InternalKey":"System.Byte[]"} |
The actual value below I extracted from different tool
CorrelationId: 096ee7ce-1d63-4803-9fee-7592060467be, InternalKey: 7d50cba2-f455-4ce5-95a8-906f7a28d723, IsCompressed: False, SourceKey: 4b7f78d7-fb29-489c-a253-68203ce6950e, TenantId: 034e7f0d-3474-4415-90b9-92b2167733c7
Hi @PRADEEPCHEEKATLA-MSFT and @Sander van de Velde ,
Just for the follow up on the current status. Do we have any way to filter it from the header?
Could you validate and guide on the right path in case if anything am i missing from my end.Thanks!!!