httpcore.ConnectError: [Errno 8] nodename nor servname provided, or not known
This error confuses us as something is wrong with kafka but the moment I switch from value_serializer from avro_serializer to 'json' everything works fine
class VmModel(faust.Record, AvroModel, serializer='avro_vm'):
imo: int #= 435677
carrier: str # = "MAERSK"
port: str = "Singapore"
class VmsModel(faust.Record, AvroModel, serializer='avro_vms'):
schedule: typing.List[VmModel] = dataclasses.field(default_factory=lambda: [{"imo": 435677, "carrier": "MAERSK", "port": "Singapore"},{"imo": 432377, "carrier": "ONE", "port": "Shanghai"},{"imo": 493377, "carrier": "YANGMING", "port": "Sydney"}])
# Initialize Schema Registry Client
client = SchemaRegistryClient(url="http://schema-registry:8081")
# VESSEL_MILESTONE
# example of how to use it with dataclasses-avroschema
avro_vm_serializer = FaustSerializer(
client,
"vessel_milestone",
VmModel.avro_schema())
def avro_vm_codec():
return avro_vm_serializer
# VESSEL_MILESTONE_SCHEDULE
# example of how to use it with dataclasses-avroschema
avro_vms_serializer = FaustSerializer(
client,
"vessel_milestone_schedule",
VmsModel.avro_schema())
def avro_vms_codec():
return avro_vms_serializer
codecs.register("avro_vm", avro_vm_codec)
codecs.register("avro_vms", avro_vms_codec)
[2021-03-19 02:33:39,537] [78421] [ERROR] [^-App]: Crashed reason=ConnectError('[Errno 8] nodename nor servname provided, or not known')
Traceback (most recent call last):
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_exceptions.py", line 342, in map_exceptions
yield
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_client.py", line 803, in _send_single_request
timeout=timeout.as_dict(),
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpcore/_sync/connection_pool.py", line 189, in request
method, url, headers=headers, stream=stream, timeout=timeout
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpcore/_sync/connection.py", line 83, in request
self.socket = self._open_socket(timeout)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpcore/_sync/connection.py", line 109, in _open_socket
local_address=self.local_address,
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpcore/_backends/sync.py", line 144, in open_tcp_stream
return SyncSocketStream(sock=sock)
File "/Users/sagungargs/.pyenv/versions/3.7.7/lib/python3.7/contextlib.py", line 130, in __exit__
self.gen.throw(type, value, traceback)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpcore/_exceptions.py", line 12, in map_exceptions
raise to_exc(exc) from None
httpcore.ConnectError: [Errno 8] nodename nor servname provided, or not known
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/mode/services.py", line 802, in _execute_task
await task
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/app/base.py", line 969, in _wrapped
return await task()
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/app/base.py", line 1022, in around_timer
await fun(*args)
File "/Users/sagungargs/Work/portcast2/faustapp/hello_world.py", line 43, in publish_milestones
await milestones.send(value=milestone, value_serializer=avro_vm_serializer)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/agents/agent.py", line 897, in send
force=force,
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/topics.py", line 198, in send
callback=callback,
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/channels.py", line 322, in _send_now
callback,
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/channels.py", line 269, in as_future_message
value, value_serializer, schema, open_headers
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/channels.py", line 703, in prepare_value
self.app, value, serializer=value_serializer, headers=headers
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/serializers/schemas.py", line 132, in dumps_value
serializer=serializer or self.value_serializer,
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/serializers/registry.py", line 180, in dumps_value
return dumps(serializer, value)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/serializers/codecs.py", line 359, in dumps
return get_codec(codec).dumps(obj) if codec else obj
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/faust/serializers/codecs.py", line 224, in dumps
obj = cast(Codec, node)._dumps(obj)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/schema_registry/serializers/faust_serializer.py", line 50, in _dumps
return self.encode_record_with_schema(self.schema_subject, self.schema, payload) # type: ignore
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/schema_registry/serializers/message_serializer.py", line 73, in encode_record_with_schema
schema_id = self.schemaregistry_client.register(subject, avro_schema)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/schema_registry/client/client.py", line 264, in register
response = self.check_version(subject, avro_schema, headers=headers, timeout=timeout)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/schema_registry/client/client.py", line 526, in check_version
result, code = self.request(url, method=method, body=body, headers=headers, timeout=timeout)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/schema_registry/client/client.py", line 219, in request
response = session.request(method, url, headers=_headers, json=body, timeout=timeout)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_client.py", line 674, in request
request, auth=auth, allow_redirects=allow_redirects, timeout=timeout
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_client.py", line 704, in send
request, auth=auth, timeout=timeout, allow_redirects=allow_redirects
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_client.py", line 733, in _send_handling_redirects
request, auth=auth, timeout=timeout, history=history
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_client.py", line 769, in _send_handling_auth
response = self._send_single_request(request, timeout)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_client.py", line 803, in _send_single_request
timeout=timeout.as_dict(),
File "/Users/sagungargs/.pyenv/versions/3.7.7/lib/python3.7/contextlib.py", line 130, in __exit__
self.gen.throw(type, value, traceback)
File "/Users/sagungargs/Work/portcast2/faustapp/.venv/lib/python3.7/site-packages/httpx/_exceptions.py", line 359, in map_exceptions
raise mapped_exc(message, **kwargs) from exc # type: ignore
httpx.ConnectError: [Errno 8] nodename nor servname provided, or not known
The Avro schemas should work for nested objects, but if I switch to serializer type json in "send" then everything works fine but it then defeats Avro requirement
@app.timer(interval=1.0,on_leader=True)
async def publish_milestones():
logger.info('PUBLISHING ON LEADER FOR SINGLE VESSEL MILESTONE RECORD!')
milestone = {"imo": 99999, "carrier": "CMDU", "port": "JNPT"}
logger.info(milestone)
logger.info(VmModel(**milestone))
logger.info(VmModel.fake())
logger.info(VmModel(imo=9999,carrier="CMDU",port="JNPT"))
await milestones.send(value=milestone, value_serializer=avro_vm_serializer)
@app.timer(interval=1.0,on_leader=True)
async def publish_milestones():
logger.info('PUBLISHING ON LEADER FOR SINGLE VESSEL MILESTONE RECORD!')
milestone = {"imo": 99999, "carrier": "CMDU", "port": "JNPT"}
logger.info(milestone)
logger.info(VmModel(**milestone))
logger.info(VmModel.fake())
logger.info(VmModel(imo=9999,carrier="CMDU",port="JNPT"))
await milestones.send(value=milestone, value_serializer="json")