Exception: AttributeError: 'str' object has no attribute 'type' python confluent kafka trigger

[sramayanam]

Trying to process many records from confluent kafka and getting the str object error somewhere in the trigger code. Tried with both 3.8 and 3.9 venv. Following is the stack trace. There is no issue in processing record by record.

[2022-09-02T21:27:58.776Z] Exception: AttributeError: 'str' object has no attribute 'type'
[2022-09-02T21:27:58.776Z] Stack: File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure_functions_worker/dispatcher.py", line 388, in _handle__invocation_request
[2022-09-02T21:27:58.776Z] args[pb.name] = bindings.from_incoming_proto(
[2022-09-02T21:27:58.776Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure_functions_worker/bindings/meta.py", line 88, in from_incoming_proto
[2022-09-02T21:27:58.776Z] return binding.decode(datum, trigger_metadata=metadata)
[2022-09-02T21:27:58.776Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure/functions/kafka.py", line 176, in decode
[2022-09-02T21:27:58.776Z] return cls.decode_multiple_events(data, trigger_metadata)
[2022-09-02T21:27:58.776Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure/functions/kafka.py", line 265, in decode_multiple_events
[2022-09-02T21:27:58.776Z] key=cls._decode_typed_data(
[2022-09-02T21:27:58.776Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure/functions/meta.py", line 126, in _decode_typed_data
[2022-09-02T21:27:58.776Z] data_type = data.type
[2022-09-02T21:27:58.776Z] .
[2022-09-02T21:27:58.777Z] Failed to executed function with 64 items in topic_0 / 3 / 2990
[2022-09-02T21:27:58.777Z] System.Private.CoreLib: Exception while executing function: Functions.processeventsconfkafka. System.Private.CoreLib: Result: Failure
[2022-09-02T21:27:58.777Z] Exception: AttributeError: 'str' object has no attribute 'type'
[2022-09-02T21:27:58.777Z] Stack: File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure_functions_worker/dispatcher.py", line 388, in _handle__invocation_request
[2022-09-02T21:27:58.777Z] args[pb.name] = bindings.from_incoming_proto(
[2022-09-02T21:27:58.777Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure_functions_worker/bindings/meta.py", line 88, in from_incoming_proto
[2022-09-02T21:27:58.777Z] return binding.decode(datum, trigger_metadata=metadata)
[2022-09-02T21:27:58.777Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure/functions/kafka.py", line 176, in decode
[2022-09-02T21:27:58.777Z] return cls.decode_multiple_events(data, trigger_metadata)
[2022-09-02T21:27:58.777Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure/functions/kafka.py", line 265, in decode_multiple_events
[2022-09-02T21:27:58.777Z] key=cls._decode_typed_data(
[2022-09-02T21:27:58.777Z] File "/usr/local/Cellar/azure-functions-core-tools@4/4.0.4736/workers/python/3.8/OSX/X64/azure/functions/meta.py", line 126, in _decode_typed_data
[2022-09-02T21:27:58.777Z] data_type = data.type
[2022-09-02T21:27:58.777Z] .

[krishna-kariya]

Hi @sramayanam, can you please share the function body for further debugging?

We have a samples folder in our repo where you can find examples for trigger and output binding for all languages. Adding reference for python samples - https://github.com/Azure/azure-functions-kafka-extension/tree/dev/samples/python

[isaacnikon93]

Hi,
i am facing the same error, i am attaching the code i am using

import json
import logging
import os
import uuid
import pandas as pd
import traceback
from datetime import date
from azure.functions import KafkaEvent
from azure.storage.blob import BlobServiceClient

blob_service_client = BlobServiceClient.from_connection_string(
    os.environ['AZURE_STORAGE_CONNECTION_STRING'])


def transform(a: KafkaEvent):
    return json.loads(json.loads(a.get_body())['Value'])


def writeToBlob(df: pd.DataFrame):
    uniqueClients = df.client.unique()
    for client in uniqueClients:
        blob_client = blob_service_client.get_blob_client(
            container=os.environ['BLOB_CONTAINER_NAME'],
            blob='aaa/{}/aaa/{}/{}.csv'.format(
                client, date.today().strftime("%d-%m-%Y"), uuid.uuid4())
        )
        tempDF = df[df['client'] == client]
        blob_client.upload_blob(tempDF.astype(str)
                                .drop(['_rid', '_self', '_ts', '_etag'], axis=1)
                                .replace('undefined', '')
                                .replace('nan', '').to_parquet(index=False)
                                )


def main(kevents):
    try:
        logging.info("Batch size {}".format(len(kevents)))
        df = pd.DataFrame(map(transform, kevents))
        writeToBlob(df=df)
    except Exception:
        print("----------------------------------------------------")
        traceback.print_exc()
        print("----------------------------------------------------")
        raise Exception("Some Error")

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "type": "kafkaTrigger",
      "direction": "in",
      "name": "kevents",
      "protocol": "SASLSSL",
      "password": "KAFKA_PASSWORD",
      "topic": "xxx",
      "authenticationMode": "PLAIN",
      "cardinality": "MANY",
      "dataType": "string",
      "consumerGroup": "KAFKA_CONSUMER_GROUP",
      "username": "KAFKA_USERNAME",
      "BrokerList": "KAFKA_BROKER_LIST"
    }
  ]
}

Dockerfile

# To enable ssh & remote debugging on app service change the base image to the one below
# FROM mcr.microsoft.com/azure-functions/python:3.0-python3.9-appservice
FROM mcr.microsoft.com/azure-functions/python:3.0-python3.9

ENV AzureWebJobsScriptRoot=/home/site/wwwroot \
    AzureFunctionsJobHost__Logging__Console__IsEnabled=true

COPY requirements.txt /
RUN pip install -r /requirements.txt
ENV LD_LIBRARY_PATH=/home/site/wwwroot/bin/runtimes/linux-x64/native
COPY . /home/site/wwwroot

host.json

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[2.*, 3.0.0)"
  },
  "extensions": {
    "kafka":{
      "maxBatchSize":1000,
      "subscriberIntervalInSeconds":300
    }
  }
}

i tired following this and removing the types, but it didn't help
Azure/azure-functions-python-worker#670

[mortenbnftw]

Also having the same issue. Any update on this?

I am running the following example from this repo

{
      "scriptFile": "main.py",
      "bindings": [
        {
            "type" : "kafkaTrigger",
            "direction": "in",
            "name" : "kevents",
            "protocol" : "SASLSSL",
            "password" : "xxxx",
            "topic" : "xxxx",
            "authenticationMode" : "PLAIN",
            "cardinality" : "MANY",
            "dataType": "string",
            "consumerGroup" : "$Default",
            "username" : "xxxx",
            "BrokerList" : "xxxx"    
        }
    ]
}

import logging
import typing
from azure.functions import KafkaEvent

def main(kevents : typing.List[KafkaEvent]):
    for event in kevents:
        logging.info(event.get_body())

I get the following error:

[2022-09-26T13:16:36.183Z] System.Private.CoreLib: Exception while executing function: Functions.kafkatrigger. System.Private.CoreLib: Result: Failure
Exception: AttributeError: 'str' object has no attribute 'type'
Stack: File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure_functions_worker\dispatcher.py", line 429, in _handle__invocation_request
args[pb.name] = bindings.from_incoming_proto(
File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure_functions_worker\bindings\meta.py", line 88, in from_incoming_proto
return binding.decode(datum, trigger_metadata=metadata)
File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure\functions\kafka.py", line 176, in decode
return cls.decode_multiple_events(data, trigger_metadata)
File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure\functions\kafka.py", line 265, in decode_multiple_events
key=cls._decode_typed_data(
File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.9/WINDOWS/X64\azure\functions\meta.py", line 126, in _decode_typed_data
data_type = data.type

[krishna-kariya]

@mortenbnftw,
It seems to be working fine for me.
I am using func core tools version - 4.0.4785.
If you are using an older version of func, can you upgrade it and check?

[mortenbnftw]

@mortenbnftw, It seems to be working fine for me. I am using func core tools version - 4.0.4785. If you are using an older version of func, can you upgrade it and check?

@krishna-kariya, Yes - Please see attached attached screenshot

[JyothsnaGoalla]

I have tried using python 3.8 venv, python 3.9 venv and used different version of core tools 4.0.4785, 4.0.4829 and few others .

I have used the code present in the sample python azure function kafka extension in git "https://github.com/Azure/azure-functions-kafka-extension/tree/dev/samples/python/KafkaTriggerMany"

Iam trying for quite a few days.Any help would be much appreciated.

[chandrabisen]

I have been facing the same issue since last three weeks, before to that it was working fine and I was able to consume the kafka messages, Not sure what has changed, but I had got confirmation from Kafka source team that they haven't done any changes at their end.

Link to the issue: https://learn.microsoft.com/en-us/answers/questions/1046604/python-azure-function-issue-for-kafka-trigger.html

I have tried below multiple things to resolve this issue but no luck.

1. Have installed core tools version 4.0.785 and older as well.
2. Have tried with multiple older version of the azure function extensions from visual studio code.
3. I have re-installed, visual studio code and python in my system.

Can anyone suggest the root cause of this issue and were you guys able to resolve it? @mortenbnftw @sramayanam @isaacnikon93 @krishna-kariya

[isaacnikon93]

@chandrabisen i haven't been able to solve it, we are still facing the same issue

[chandrabisen]

@isaacnikon93 : Have you got any official confirmation from Microsoft support, if there is any internal issue going on?

[isaacnikon93]

nope I haven't received any official communication, we had to temporarily start using node js azure function which seems to be working fine

[krishna-kariya]

I am looking into this issue.

[shrohilla]

I have tried using python 3.8 venv, python 3.9 venv and used different version of core tools 4.0.4785, 4.0.4829 and few others .

I have used the code present in the sample python azure function kafka extension in git "https://github.com/Azure/azure-functions-kafka-extension/tree/dev/samples/python/KafkaTriggerMany" Iam trying for quite a few days.Any help would be much appreciated.

Hi @JyothsnaGoalla/ @mortenbnftw request you to please share the sample payload where it crashed, that would be helpful to analyze more.