GKE Ingestion Jobs Failing

[lgvital]

Expected Behavior

Ingestion jobs successfully run and bigquery table (feast.customer_project_customer_transactions_v1) gets populated with data.

Current Behavior

>>> client.ingest("customer_transactions", customer_features)
Waiting for feature set to be ready for ingestion...
  0%|                                                                                                | 0/155 [04:48<?, ?rows/s]
Ingestion complete!

Ingestion statistics:
Success: 0/155
Removing temporary file(s)...

BQ table is empty:

SELECT COUNT(*) FROM `feast.customer_project_customer_transactions_v1`;

is 0.

Steps to reproduce

Follow latest GKE setup docs with the following differences:

• my-project as project ID
• Use image 0.4.2
• Use NodePort to expose service to local client
• In basic example, use $FEAST_CORE_URL and $FEAST_BATCH_SERVING_URL
• Open up nodeports in GCP firewall 32090, 32091, 32092:

gcloud compute firewall-rules create feast-core-port --allow tcp:32090
gcloud compute firewall-rules create feast-online-port --allow tcp:32091
gcloud compute firewall-rules create feast-batch-port --allow tcp:32092

my-feast-values.yaml ends up looking like:

feast-core:
  enabled: true
  image:
    tag: "0.4.2"
  jvmOptions:
  - -Xms1024m
  - -Xmx1024m
  resources:
    requests:
      cpu: 1000m
      memory: 1024Mi
  service:
    type: NodePort
    grpc:
      nodePort: 32090
  gcpServiceAccount:
    useExistingSecret: true
feast-serving-online:
  enabled: true
  redis:
    enabled: true
  image:
    tag: "0.4.2"
  jvmOptions:
  - -Xms1024m
  - -Xmx1024m
  resources:
    requests:
      cpu: 500m
      memory: 1024Mi
  service:
    type: NodePort
    grpc:
      nodePort: 32091
  store.yaml:
    name: redis
    type: REDIS
    redis_config:
      port: 6379
    subscriptions:
    - name: "*"
      project: "*"
      version: "*"
feast-serving-batch:
  enabled: true
  redis:
    enabled: false
  image:
    tag: "0.4.2"
  jvmOptions:
  - -Xms1024m
  - -Xmx1024m
  resources:
    requests:
      cpu: 500m
      memory: 1024Mi
  service:
    type: NodePort
    grpc:
      nodePort: 32092
  gcpServiceAccount:
    useExistingSecret: true
  application.yaml:
    feast:
      jobs:
        staging-location: gs://my-project_feast_bucket/serving/batch
        store-type: REDIS
        store-options:
          host: localhost
          port: 6379
  store.yaml:
    name: bigquery
    type: BIGQUERY
    bigquery_config:
      project_id: my-project
      dataset_id: feast
    subscriptions:
    - name: "*"
      project: "*"
      version: "*"

Specifications

• Version: 0.4.2, latest master python SDK (2a33f7b)
• Platform: Installing GKE on local Mac OSX
• Subsystem: Python 3.7.6, helm v2.16.1, kubectl v1.17.0

Possible Solution

Default Kafka configs might need adjusting? Maybe related to NodePort config?

kubectl logs for feast-feast-core appear to show kafka jobs were successfully created and sent to DirectRunner:

22:57:52 [pool-5-thread-1] INFO  feast.ingestion.ImportJob - Starting import job with settings:
Current Settings:
  appName: DirectRunnerJobManager
  blockOnRun: false
  enforceEncodability: true
  enforceImmutability: true
  featureSetJson: [{
  "name": "customer_transactions",
  "version": 1,
  "entities": [{
    "name": "customer_id",
    "valueType": "INT64"
  }],
  "features": [{
    "name": "total_transactions",
    "valueType": "INT64"
  }, {
    "name": "daily_transactions",
    "valueType": "DOUBLE"
  }],
  "maxAge": "86400s",
  "source": {
    "type": "KAFKA",
    "kafkaSourceConfig": {
      "bootstrapServers": "feast-kafka:9092",
      "topic": "feast"
    }
  },
  "project": "customer_project"
}]
  gcsPerformanceMetrics: false
  optionsId: 0
  project:
  runner: class org.apache.beam.runners.direct.DirectRunner
  stableUniqueNames: WARNING
  storeJson: [{
  "name": "bigquery",
  "type": "BIGQUERY",
  "subscriptions": [{
    "name": "*",
    "version": "*",
    "project": "*"
  }],
  "bigqueryConfig": {
    "projectId": "my-project",
    "datasetId": "feast"
  }
}]

22:57:54 [pool-5-thread-1] INFO  feast.ingestion.utils.StoreUtil - Writing to existing BigQuery table 'my-project:feast.customer_project_customer_transactions_v1'
22:57:54 [pool-6-thread-1] INFO  org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Partitions assigned to split 0 (total 1): feast-0
2020-01-09 22:57:54.984 AUDIT feast-feast-core-dc485b44d-qg75w --- [pool-6-thread-1] f.c.l.AuditLogger                        : {action=STATUS_CHANGE, detail=Job submitted to runner DirectRunner with ext id kafka-to-redis1578610671583., id=kafka-to-redis1578610671583, resource=JOB, timestamp=Thu Jan 09 22:57:54 UTC 2020}
22:57:55 [pool-5-thread-1] INFO  org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Partitions assigned to split 0 (total 1): feast-0
2020-01-09 22:57:55.373 AUDIT feast-feast-core-dc485b44d-qg75w --- [pool-5-thread-1] f.c.l.AuditLogger                        : {action=STATUS_CHANGE, detail=Job submitted to runner DirectRunner with ext id kafka-to-bigquery1578610671583., id=kafka-to-bigquery1578610671583, resource=JOB, timestamp=Thu Jan 09 22:57:55 UTC 2020}
22:57:55 [pool-2-thread-1] INFO  feast.core.service.JobCoordinatorService - Updating feature set status
22:57:58 [direct-runner-worker] INFO  org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0: reading from feast-0 starting at offset 0
22:57:58 [direct-runner-worker] INFO  org.apache.beam.sdk.io.kafka.KafkaUnboundedSource - Reader-0: reading from feast-0 starting at offset 0

And a feast topic was successfully created according to the kafka-config pod:

Waiting for Zookeeper...
Waiting for Kafka...
Applying runtime configuration using confluentinc/cp-kafka:5.0.1
Created topic "feast".
Configs for topic 'feast' are 

But, I'm getting errors thrown related to topics in feast-kafka-0 logs:

[2020-01-09 22:57:57,799] INFO [Log partition=__consumer_offsets-2, dir=/opt/kafka/data/logs] Truncating to 0 has no effect as the largest offset in the log is -1 (kafka.log.Log)
[2020-01-09 22:57:57,804] ERROR [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Error for partition __consumer_offsets-8 at offset 0 (kafka.server.ReplicaFetcherThread)
org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
[2020-01-09 22:57:57,804] ERROR [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Error for partition __consumer_offsets-35 at offset 0 (kafka.server.ReplicaFetcherThread)
org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.
[2020-01-09 22:57:57,805] ERROR [ReplicaFetcher replicaId=0, leaderId=1, fetcherId=0] Error for partition __consumer_offsets-41 at offset 0 (kafka.server.ReplicaFetcherThread)

Happy to share any other relevant logs. I'm admittedly not familiar with Kafka, so I could be off here. Just trying to get the GKE feast guide + basic example working end to end. Once it works, happy to put up a PR to update the guide for 0.4.X.

[lgvital]

FWIW same issue with 0.4.3.

[woop]

@lgvital Thanks for the super detailed logs. Can you confirm that you can the Kafka brokers from the environment where you are running the SDK/CLI? You should be able to telnet to feast-kafka:9092

It could be that the SDK is not returning a proper exception and this is related to connectivity.

[lgvital]

Hi @woop, I'm currently running the feast SDK/CLI from my local environment--should I be expected to be able to access feast-kafka without a NodePort? I was not able to telnet 9092 with the feast IP (external) or internal ip, or maybe I'm not clear on what exactly you wanted me to try.

Perhaps I need to set up a NodePort for kafka? I tried a few different variations, e.g.

  kafka:
    enabled: true
    external:
      enabled: true
    configurationOverrides:
      advertised.listeners: |-
        EXTERNAL://${FEAST_IP}:$((31090 + ${KAFKA_BROKER_ID}))
      listener.security.protocol.map: |-
        PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT

Unsure what to put under advertised.listeners though. I also tried
EXTERNAL://kafka.cluster.local:$((31090 + ${KAFKA_BROKER_ID})) from the kafka config docs. Opening up the 31090 port in gcloud's firewall (similar to above) did not seem to help.

feast-core pod ends up failing to run with the following error related to kafka:

Failed to instantiate [feast.core.model.Source]: Factory method 'getDefaultSource' threw exception; nested exception is org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient

Some guidance on what we could try next would be appreciated!

[woop]

I was not able to telnet 9092 with the feast IP (external) or internal ip, or maybe I'm not clear on what exactly you wanted me to try.

You have to be able to telnet to that port. It means that there is something wrong with your connectivity.

FEAST_IP should be the external IP of one of your nodes. This can be a public internet IP, but it must be whitelisted in your firewall if it isnt already. Can you try that and let me know if it makes a difference?

[lgvital]

Thanks for the response. I have $FEAST_IP defined as:

export FEAST_IP=$(kubectl describe nodes | grep ExternalIP | awk '{print $2}' | head -n 1)

I have opened up the 9092 port in my GCP firewall using:

gcloud compute firewall-rules create kafka-port --allow tcp:9092

I am unable to telnet into 9092.

telnet $FEAST_IP 9092
Trying <external_ip_redacted>...
telnet: connect to address <external_ip_redacted>: Connection refused
telnet: Unable to connect to remote host

From my understanding though, to have a port be accessible, I need to set up a node port similar to the other feast services:

>> telnet $FEAST_IP 32090
Trying <external_ip_redacted>...
Connected to <ip>.bc.googleusercontent.com.
Escape character is '^]'.

However, from what I tried, I wasn't able to successfully start my services. If I have connectivity issues with 9092, should I be setting up a node port for 9092?

[woop]

Thanks for the response. I have $FEAST_IP defined as:

export FEAST_IP=$(kubectl describe nodes | grep ExternalIP | awk '{print $2}' | head -n 1)

I have opened up the 9092 port in my GCP firewall using:

gcloud compute firewall-rules create kafka-port --allow tcp:9092

I am unable to telnet into 9092.

telnet $FEAST_IP 9092
Trying <external_ip_redacted>...
telnet: connect to address <external_ip_redacted>: Connection refused
telnet: Unable to connect to remote host

From my understanding though, to have a port be accessible, I need to set up a node port similar to the other feast services:

>> telnet $FEAST_IP 32090
Trying <external_ip_redacted>...
Connected to <ip>.bc.googleusercontent.com.
Escape character is '^]'.

However, from what I tried, I wasn't able to successfully start my services. If I have connectivity issues with 9092, should I be setting up a node port for 9092?

Yes, you should be enabling NodePort for Kafka as well. Thank you for being so patient with the charts and docs being out of date.

The values-demo.yaml is enabling NodePort for Kafka. Can you try that and see if it works for you?

[lgvital]

Thank you for being so patient with the charts and docs being out of date.

No worries! Like I mentioned, my goal is just to get this working end to end in GKE. Happy to update the charts/docs afterward.

The values-demo.yaml is enabling NodePort for Kafka. Can you try that and see if it works for you?

Yup! I've tried that, replacing feast.example.com with the output of $FEAST_IP. I end up with the following exception thrown in feast-core:

[feast/core/config/FeatureStreamConfig.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [feast.core.model.Source]: Factory method 'getDefaultSource' threw exception; nested exception is org.apache.kafka.common.KafkaException: Failed to create new KafkaAdminClient

[lgvital]

To clarify, using the NodePort config, I am able to telnet

$ telnet $FEAST_IP 31091
Trying <external_ip_redacted>...
Connected to <ip>.bc.googleusercontent.com.
Escape character is '^]'.
^CConnection closed by foreign host.

but feast-core shuts itself down due to the org.apache.kafka.common.KafkaException.

[lgvital]

Was missing a key piece in my config, the application.yaml:

  application.yaml:
    feast:
      stream:
        options:
          bootstrapServers: <external_ip_redacted>:31090

However, when ingesting, I get similar errors in my kafka logs as the OP:

[2020-01-16 00:25:54,276] ERROR [ReplicaFetcher replicaId=0, leaderId=2, fetcherId=0] Error for partition __consumer_offsets-30 at offset 0 (kafka.server.ReplicaFetcherThread)
[2020-01-16 00:25:54,276] ERROR [ReplicaFetcher replicaId=0, leaderId=2, fetcherId=0] Error for partition __consumer_offsets-21 at offset 0 (kafka.server.ReplicaFetcherThread)
[2020-01-16 00:25:54,276] ERROR [ReplicaFetcher replicaId=0, leaderId=2, fetcherId=0] Error for partition __consumer_offsets-27 at offset 0 (kafka.server.ReplicaFetcherThread)
[2020-01-16 00:25:54,276] ERROR [ReplicaFetcher replicaId=0, leaderId=2, fetcherId=0] Error for partition __consumer_offsets-9 at offset 0 (kafka.server.ReplicaFetcherThread)

I did notice something else odd in the logs that could be relevant:

[2020-01-16 00:25:55,198] TRACE [Controller id=0 epoch=1] Received response {error_code=0} for request UPDATE_METADATA with correlation id 2 sent to broker <ip>:9092 (id: 2 rack: null) (state.change.logger)

where <ip> does not map to my $FEAST_IP.

[lgvital]

Okay, getting close.

It was related to the wrong IP. After updating my bootstrapServers in my helm chart, the existing customer_fs was using my old config, so it was using feast-kafka:9092 as the broker. I'm not sure if that's expected behavior?

I forced it to use new config by creating a new feature set and was able to ingest successfully!

100%|█████████████████████████████████████████████████████████████████████████████████████| 155/155 [00:00<00:00, 227.54rows/s]
Ingestion complete!███████████████████████████████████████████████████████████████████████| 155/155 [00:00<00:00, 227.78rows/s]

Batch serving is throwing me this error though, but hopefully a similar fix:

2020-01-16 23:57:49.861 ERROR feast-feast-serving-batch-757cd5d649-4xj8d --- [ault-executor-1] f.s.s.RedisBackedJobService              : Failed to upsert job: Failed connecting to host localhost:6379
2020-01-16 23:57:49.955  WARN feast-feast-serving-batch-757cd5d649-4xj8d --- [ault-executor-1] f.s.c.ServingServiceGRpcController       : Failed to get Job
redis.clients.jedis.exceptions.JedisConnectionException: Failed connecting to host localhost:6379
	at redis.clients.jedis.Connection.connect(Connection.java:207)
	at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:101)
	at redis.clients.jedis.Connection.sendCommand(Connection.java:126)
	at redis.clients.jedis.Connection.sendCommand(Connection.java:117)
	at redis.clients.jedis.Jedis.get(Jedis.java:155)
	at feast.serving.service.RedisBackedJobService.get(RedisBackedJobService.java:45)
	at feast.serving.service.BigQueryServingService.getJob(BigQueryServingService.java:171)
	at feast.serving.controller.ServingServiceGRpcController.getJob(ServingServiceGRpcController.java:100)
	at feast.serving.ServingServiceGrpc$MethodHandlers.invoke(ServingServiceGrpc.java:503)
	at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:171)
	at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:283)
	at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:710)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.net.ConnectException: Connection refused (Connection refused)
	at java.net.PlainSocketImpl.socketConnect(Native Method)
	at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
	at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
	at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
	at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
	at java.net.Socket.connect(Socket.java:589)
	at redis.clients.jedis.Connection.connect(Connection.java:184)
	... 16 more
2020-01-16 23:58:00.336 ERROR feast-feast-serving-batch-757cd5d649-4xj8d --- [       Thread-6] f.s.s.RedisBackedJobService              : Failed to upsert job: Failed connecting to host localhost:6379
2020-01-16 23:58:49.026  INFO feast-feast-serving-batch-757cd5d649-4xj8d --- [pool-3-thread-1] f.s.s.CachedSpecService                  : loaded store config at /etc/feast/feast-serving/store.yaml: 

[woop]

Okay, getting close.

It was related to the wrong IP. After updating my bootstrapServers in my helm chart, the existing customer_fs was using my old config, so it was using feast-kafka:9092 as the broker. I'm not sure if that's expected behavior?

I forced it to use new config by creating a new feature set and was able to ingest successfully!

100%|█████████████████████████████████████████████████████████████████████████████████████| 155/155 [00:00<00:00, 227.54rows/s]
Ingestion complete!███████████████████████████████████████████████████████████████████████| 155/155 [00:00<00:00, 227.78rows/s]

Batch serving is throwing me this error though, but hopefully a similar fix:

2020-01-16 23:57:49.861 ERROR feast-feast-serving-batch-757cd5d649-4xj8d --- [ault-executor-1] f.s.s.RedisBackedJobService              : Failed to upsert job: Failed connecting to host localhost:6379
2020-01-16 23:57:49.955  WARN feast-feast-serving-batch-757cd5d649-4xj8d --- [ault-executor-1] f.s.c.ServingServiceGRpcController       : Failed to get Job
redis.clients.jedis.exceptions.JedisConnectionException: Failed connecting to host localhost:6379
	at redis.clients.jedis.Connection.connect(Connection.java:207)
	at redis.clients.jedis.BinaryClient.connect(BinaryClient.java:101)
	at redis.clients.jedis.Connection.sendCommand(Connection.java:126)
	at redis.clients.jedis.Connection.sendCommand(Connection.java:117)
	at redis.clients.jedis.Jedis.get(Jedis.java:155)
	at feast.serving.service.RedisBackedJobService.get(RedisBackedJobService.java:45)
	at feast.serving.service.BigQueryServingService.getJob(BigQueryServingService.java:171)
	at feast.serving.controller.ServingServiceGRpcController.getJob(ServingServiceGRpcController.java:100)
	at feast.serving.ServingServiceGrpc$MethodHandlers.invoke(ServingServiceGrpc.java:503)
	at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:171)
	at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:283)
	at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:710)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.net.ConnectException: Connection refused (Connection refused)
	at java.net.PlainSocketImpl.socketConnect(Native Method)
	at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
	at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
	at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
	at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
	at java.net.Socket.connect(Socket.java:589)
	at redis.clients.jedis.Connection.connect(Connection.java:184)
	... 16 more
2020-01-16 23:58:00.336 ERROR feast-feast-serving-batch-757cd5d649-4xj8d --- [       Thread-6] f.s.s.RedisBackedJobService              : Failed to upsert job: Failed connecting to host localhost:6379
2020-01-16 23:58:49.026  INFO feast-feast-serving-batch-757cd5d649-4xj8d --- [pool-3-thread-1] f.s.s.CachedSpecService                  : loaded store config at /etc/feast/feast-serving/store.yaml: 

I have also been going through the helm installation and resolved most issues, but I have discovered new holes that I am trying to plug. Here is my current list of changes that I want to make to the docs/charts

1. Open up firewall to allow NodePorts

gcloud compute firewall-rules create feast-core-port --allow tcp:32090
gcloud compute firewall-rules create feast-online-port --allow tcp:32091
gcloud compute firewall-rules create feast-batch-port --allow tcp:32092
gcloud compute firewall-rules create feast-redis-port --allow tcp:32101
gcloud compute firewall-rules create feast-kafka --allow tcp:31090-31095

2. Create a bucket and update the helm values in batch serving to have the correct path.
3. Update the dataset and project in store.yaml key in helm values
4. Update store-options: to use an external IP and NodePort. Feast batch serving uses Redis as its job store to retain state, so it needs to be able to connect to Redis. Its better to use the Redis that online serving uses so that we don't need to deploy two servers. (FYI: this is tech debt and will be cleared in the future).

application.yaml:
    feast:
      jobs: 
        staging-location: gs://[bucket]/staging
        store-type: REDIS
        store-options:
          host: [EXTERNAL IP]
          port: 32101

update 
    master:
      service:
        nodePort: 36379
        type: NodePort

5. Update image tag to 0.4.3
6. Add NodePort to all services, for example

  service:
    type: NodePort
    grpc:
      nodePort: 32092

7. Set the Feast Online Serving store to use an external IP and port

 store.yaml:
    name: redis
    type: REDIS
    redis_config:
      host: [EXTERNAL IP]
      port: 32101

• Make sure that a Redis instance is deployed with online serving and that it uses NodePort

  redis:
    enabled: true
    master:
      service:
        nodePort: 32101
        type: NodePort

8. Make sure Kafka is also using NodePort

  kafka:
    enabled: true
    persistence:
      enabled: false
    external:
      enabled: true
      type: NodePort
      domain: [EXTERNAL IP]
    configurationOverrides:
      "advertised.listeners": |-
        EXTERNAL://[EXTERNAL IP]:$((31090 + ${KAFKA_BROKER_ID}))
      "listener.security.protocol.map": |-
        PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT

I think the problem you have right now is that Feast Batch Serving is trying to find that Redis instance but it cant find it on localhost. If you are using DirectRunner then the Redis does not need to be exposed outside of the cluster and you can just provide Feast Batch Serving with the internal IP and port of the Redis service. If you are going to use Dataflow then you need to use the externally exposed Redis IP (as I have done above) when configuring Feast Batch Serving.

[lgvital]

Great exhaustive list! Aside from 7, I've come to very similar changes. I implemented 7 and it fixed my batch serving / redis issue, thanks!

Make sure Kafka is also using NodePort

I think you also want to make sure to include the bootstrapServers config similar to values-demo.yaml.

  application.yaml:
    feast:
      stream:
        options:
          bootstrapServers: [EXTERNAL IP]:31090