I’m stuck on Stage #describetopicpartition stage
Here are my logs:
remote: [tester::#YK1] ✓ Correlation ID: 717410973
remote: [tester::#YK1] ✓ Error code: 0 (NO_ERROR)
remote: [tester::#YK1] Expected API keys array to include atleast 2 keys, got 1
remote: [tester::#YK1] Test failed
its not identifying 2nd key
And here’s a snippet of my code:
def handle_client(addr,port):
try:
while True:
# a = 4 + 12 + 3
a= 4 + 12 +3
data = addr.recv(1024) ## each time loop iterates server waits for incoming data then only proceed
##we used while True to allow continues processing of multiple requests without restarting the server
if not data: # Client disconnected
print(f"Client {port} disconnected")
break
corrId = getCorrId(data)
errorCode = checkApiVersion(data)
apiKey = data[4:6]
api_keys_array = (
int(18).to_bytes(2, byteorder="big") # APIVersions key
+ int(0).to_bytes(2, byteorder="big") # Min version
+ int(4).to_bytes(2, byteorder="big") # Max version
+ int(75).to_bytes(2, byteorder="big") # DescribeTopicPartitions key
+ int(0).to_bytes(2, byteorder="big") # Min version
+ int(0).to_bytes(2, byteorder="big"), # Max version
)
# Combine the array of bytes into a single byte sequence
api_keys_bytes = b"".join(api_keys_array)
addr.sendall(
a.to_bytes(4, byteorder="big")
+ corrId
+ errorCode
+ int(1 + 1).to_bytes(1, byteorder="big") #number of api keys
+ api_keys_bytes
+ int(0).to_bytes(4, byteorder="big")#tag
)
except Exception as e:
print(f"Error handling client: {e}")
finally:
addr.close() ```
@faith-in-het, could you upload your code to GitHub and share the link? It will be much easier to debug if I can run it directly.
Hey @faith-in-het, there might be a few issues at play. Here are the first ones I noticed:
- Incorrect number of API keys:
- Missing TAG_BUFFERs (one byte of 0) for each api key:
- Missing .throttle_time_ms
- Incorrect message size:
Feel free to use BinspecVisualizer to view the structure of the APIVersions Response.
import socket # noqa: F401
import threading
def getCorrId(data):
corrId = data[8:12]
print(corrId, int.from_bytes(corrId, byteorder="big"))
return corrId
def checkApiVersion(data):
errCode = 35
apiVersion = int.from_bytes(data[6:8], byteorder="big")
if 0 <= apiVersion <= 4:
errCode = 0
return errCode.to_bytes(2, byteorder="big")
def handle_client(addr,port):
try:
while True:
a= 30
data = addr.recv(1024) ## each time loop iterates server waits for incoming data then only proceed
##we used while True to allow continues processing of multiple requests without restarting the server
if not data: # Client disconnected
print(f"Client {port} disconnected")
break
corrId = getCorrId(data)
errorCode = checkApiVersion(data)
apiKey = data[4:6]
api_keys_bytes = b"".join([
int(18).to_bytes(2, byteorder="big"), # APIVersions key
int(0).to_bytes(2, byteorder="big"), # Min version
int(4).to_bytes(2, byteorder="big"), # Max version
int(0).to_bytes(1, byteorder="big"), # Extra byte
int(75).to_bytes(2, byteorder="big"), # DescribeTopicPartitions key
int(0).to_bytes(2, byteorder="big"), # Min version
int(0).to_bytes(2, byteorder="big"), # Max version
int(0).to_bytes(1, byteorder="big") # Extra byte
])
#Combine the array of bytes into a single byte sequence
#api_keys_bytes = b"".join(api_keys_array)
a=19
addr.sendall(
a.to_bytes(4, byteorder="big")
+ corrId
+ errorCode
+ int(2).to_bytes(1, byteorder="big") #number of api keys
+ api_keys_bytes
+ int(0).to_bytes(4, byteorder="big")#throttle time
)
# response_data = (
# corrId
# + errorCode
# + int(1).to_bytes(1, byteorder="big") # number of API keys
# + api_keys_bytes
# + int(0).to_bytes(4, byteorder="big") # throttle time
# )
# a = len(response_data)
# addr.sendall(
# a.to_bytes(4, byteorder="big") # Response size prefix
# + response_data
# )
except Exception as e:
print(f"Error handling client: {e}")
finally:
print(a)
addr.close()
def main():
# You can use print statements as follows for debugging,
# they'll be visible when running tests.
print("Logs from your program will appear here!")
# Uncomment this to pass the first stage
#
server = socket.create_server(("localhost", 9092), reuse_port=True)
while True:
addr, port = server.accept() # wait for client
client_thread=threading.Thread(target=handle_client,args=(addr,port))
client_thread.start()
if __name__ == "__main__":
main()
when everything got its only accepting the first api key not accepting the 2nd one
@faith-in-het, the ApiVersions array is a COMPACT_ARRAY, which means its length should be encoded as the actual number of entries plus one.
In this BinSpec example, there are 3 ApiVersion entries, so the length is encoded as 4:
Let me know if you need further clarification!
So The issue was I have to write length in unsigned_varient that i got by this code
def encode_unsigned_varint(value):
"""Encode an integer as UNSIGNED_VARINT."""
encoded = []
while value > 0x7F:
encoded.append((value & 0x7F) | 0x80)
value >>= 7
encoded.append(value)
return bytes(encoded)
Thank you soo much for your help bro bless you 
