對于我的學士論文，我嘗試使用 http 連接向 kafka 發送機器數據（在這種情況下，使用 python 腳本發送的歷史數據）。我正在使用在 Windows 系統上的 docker 中運行的融合平臺。使用 python 腳本，我嘗試將數據發送到 REST 代理。起初，我收到了關于我能夠解決的數據類型的錯誤響應。import pandas as pdimport csv, os, json, requests, time, datetime, copy, sysif len(sys.argv) > 1:    bgrfc_value = str(sys.argv[1])else:    print("No arguments for bgrfc given, defaulting to 'false'")    bgrfc_value = 'false'if len(sys.argv) > 2:    filePath = str(sys.argv[2])else:    filePath = "path"if len(sys.argv) > 3:    batchSize = int(float(str(sys.argv[3])))else:    batchSize = 10# Build skeleton JSONbasejson = {"message": {"meta" : "", "data": ""}}#metajson = [{'meta_key' : 'sender', 'meta_value': 'OPCR'},#           {'meta_key' : 'receiver', 'meta_value': 'CAT'},#            {'meta_key' : 'message_type', 'meta_value': 'MA1SEK'},#            {'meta_key' : 'bgrfc', 'meta_value': bgrfc_value}]#basejson['message']['meta'] = metajsonurl = "http://127.0.0.1:8082/"headers = {'Content-Type':'application/json','Accept':'application/json'}def assign_timestamps(batch):    newtimestamps = []    oldtimestamps = []    # Batch timestamps to list, add 10 newly generated timestamps to a list    for item in batch['tag_tsp'].values.tolist():        newtimestamps.append(datetime.datetime.now())        oldtimestamps.append(datetime.datetime.strptime(str(item), "%Y%m%d%H%M%S.%f"))    # Sort old timestamps without sorting the original array to preserve variance    temp = copy.deepcopy(oldtimestamps)    temp.sort()    mrtimestamp = temp[0]    # Replicate variance of old timestamps into the new timestamps    for x in range(batchSize):        diff = mrtimestamp - oldtimestamps[x]        newtimestamps[x] = newtimestamps[x] - diff        newtimestamps[x] = newtimestamps[x].strftime("%Y%m%d%H%M%S.%f")[:-3]    # Switch old timestamps with new timestamps    batch['tag_tsp'] = newtimestamps    return batch該腳本發送數據，但作為響應，我得到狀態代碼 500。