kafka consumer cannot connect

See also questions close to this topic

• Is there any way to enable Kubernetes on Docker for Mac via terminal?

  I'm developing an Electron app and want to distribute the back-end portion of the web application (PHP) via Docker and Kubernetes (using Helm charts). I plan to package the expanded dmg of Docker, but haven't found a way to configure Docker from terminal. Is this possible - enable Kubernetes and increase CPU size and RAM via terminal?
• Tkinter display error running in Ubuntu container

  I am new to docker and ubuntu. I am currently trying to display a simple Tkinter GUI in an ubuntu container from a docker toolbox.

  Already installed python 2.7 in my ubuntu container and when i run my .py file i will receive the following error: _tkinter.TclError: no display name and no $DISPLAY environment variable.

  import Tkinter window = Tkinter.Tk()
• nginx port redirection sends empty json object

  I have a docker-compose app with an nginx 'gateway' that redirects request from port 3000 to port 5000 but sends an empty object.

  When the container on port 3000 makes a request to '/api/example', the nginx container will redirect the request to the container on port 5000 that will save the object in the database.

  this is the nginx configuration file:

  upstream client {
    server client:3000;
  }
  
  upstream api {
    server api:5000;
  }
  
  server {
    listen 80;
  
    location / {
      proxy_pass http://client;
    }
  
    location /api {
      rewrite /api/(.*) /$1 break;
      proxy_pass http://api;
    }
  }
  

  and the Dockerfile:

  FROM nginx
  COPY ./default.conf /etc/nginx/conf.d/default.conf
  

  docker-compose:

  version: '3'
  services: 
    mongo:
      container_name: mongo
      restart: always
      image: mongo
      env_file: .env
      environment:
      environment:
        MONGO_INITDB_ROOT_PASSWORD: ${MONGO_USER}
        MONGO_INITDB_ROOT_USERNAME: ${MONGO_PASSWORD}
        MONGO_INITDB_DATABASE: ${MONGO_DATABASE}
      volumes:
        - ./mongoUser.js:/docker-entrypoint-initdb.d/001_users.js:ro
        - ./mongodb:/data/db
      ports:
        - '27017:27017'
    api:
      env_file: .env
      build:
        dockerfile: Dockerfile
        context: ./api
      links:
        - mongo
      depends_on:
        - mongo
      volumes:
        - /node_modules
        - ./api:/app
      environment:
        - MONGO_USER=${MONGO_USER}
        - MONGO_PASSWORD=${MONGO_PASSWORD}
        - MONGO_DATABASE=${MONGO_DATABASE}
    client:
      build:
        dockerfile: Dockerfile
        context: ./client
      volumes:
      - /app/node_modules
      - ./client:/app
    nginx:
      restart: always
      depends_on:
        - mongo
        - api
        - client
      build:
        dockerfile: Dockerfile
        context: ./nginx
      ports:
        - '80:80'
  

  I've tried to insert data in the api thought postman and it succeeds. However, when making a simple axios post from the client the server receives an empty object.

  front-end:

  axios.post('/api/example/', { item: 'something' });
  

  express - mongoose logs:

  api_1     | posting:  {} // This is the body of the request.
  api_1     | { ValidationError: item validation failed: item: Path `item` is required.
  api_1     |     at ValidationError.inspect (/node_modules/mongoose/lib/error/validation.js:59:24)
  api_1     |     at formatValue (internal/util/inspect.js:521:31)
  api_1     |     at inspect (internal/util/inspect.js:196:10)
  api_1     |     at Object.formatWithOptions (util.js:84:12)
  api_1     |     at Console.(anonymous function) (internal/console/constructor.js:274:15)
  api_1     |     at Console.log (internal/console/constructor.js:283:59)
  api_1     |     at newItem.save.then.catch.err (/controllers/example.js:21:27)
  api_1     |     at processTicksAndRejections (internal/process/next_tick.js:81:5)
  api_1     |   errors:
  api_1     |    { item:
  api_1     |       { ValidatorError: Path `item` is required.
  api_1     |           at new ValidatorError (/node_modules/mongoose/lib/error/validator.js:29:11)
  api_1     |           at validate (/node_modules/mongoose/lib/schematype.js:926:13)
  api_1     |           at /node_modules/mongoose/lib/schematype.js:979:11
  api_1     |           at Array.forEach (<anonymous>)
  api_1     |           at SchemaString.SchemaType.doValidate (/node_modules/mongoose/lib/schematype.js:935:19)
  api_1     |           at /node_modules/mongoose/lib/document.js:1936:9
  api_1     |           at processTicksAndRejections (internal/process/next_tick.js:74:9)
  api_1     |         message: 'Path `item` is required.',
  api_1     |         name: 'ValidatorError',
  api_1     |         properties: [Object],
  api_1     |         kind: 'required',
  api_1     |         path: 'item',
  api_1     |         value: undefined,
  api_1     |         reason: undefined,
  api_1     |         [Symbol(mongoose:validatorError)]: true } },
  api_1     |   _message: 'item validation failed',
  api_1     |   name: 'ValidationError' }
  nginx_1   | 172.18.0.1 - - [18/Feb/2019:01:07:28 +0000] "POST /api/example/ HTTP/1.1" 504 569 "http://localhost/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.109 Safari/537.36" "-"
  nginx_1   | 2019/02/18 01:07:28 [error] 7#7: *4 upstream timed out (110: Connection timed out) while reading response header from upstream, client: 172.18.0.1, server: , request: "POST /api/example/ HTTP/1.1", upstream: "http://172.18.0.4:5000/example/", host: "localhost", referrer: "http://localhost/"
  
• How to seperate a consumer from consumer group without losing offset?

  I have a logstash kafka consumer group subscribing nearly 20topics which isn't performing well for a particular high priority kafka topic to process, so I've decided to remove one topic out of the consumer group and launch a separate consumer group for high priority topics, but unfortunately with that i'm losing the offset that was there in old consumer group.

  Is there anyway I can start the new logstash consumer group with the initial offset from last consumer group?

  Thanks
• Pyspark window function not printing results

  I have a pyspark streaming code which reads heartbeat data from a kafka server. I can confirm that I receive data using the line below.

  heartBeats = mac_id.groupBy('macId').count()
  

  I want to use window functions to keep track of devices which have not sent hearbeats, say in the past few minutes.

  windowedCountsDF = \
    mac_id \
      .withWatermark("time", "10 seconds") \
      .groupBy(
        mac_id.macId,
        window(mac_id.time, "10 seconds", "5 seconds")) \
      .count()
  

  However this does not display any results whatsoever when the streaming execution starts , even though the data arrives.

  query = windowedCountsDF.writeStream.outputMode('complete').format('console').option('truncate',False).start()

  Is there anything going wrong here ? Thanks !
• SchemaException - Error reading field 'leader_id': String length -1 cannot be negative - Azure event hubs - Kafka

  I am getting this exception

  org.apache.kafka.common.protocol.types.SchemaException
  

  while the kafka is re-balancing

  These are the details:

  1. Using Azure event hubs. Accessing it using the kafka API

  2. "Kafka Enabled" = yes , in azure portal

  3. using: compile group: 'org.apache.kafka', name: 'kafka-clients', version: '1.0.2'

  4. Using a consumer group

  5. Properties properties = new Properties();
     properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
     properties.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);
     properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
     properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, String.format("%s.servicebus.windows.net:9093", this.namespace));
     properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
     properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MeasurementDeSerializer.class.getName());
     properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 120000);
     properties.put(ConsumerConfig.GROUP_ID_CONFIG, this.groupName);
     properties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
     

  I have 2 clients on 2 different PC's

  When they both run , each gets 16 partitions out of the available 32 partitions.

  When i shutdown one of them , all the parts are re-balanced to the other one.

  on the instance that is still running i will get:

  1. Partitions Revoked [16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31]

  2. Then , from the pool loop i will get this exception:

     org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'leader_id': String length -1 cannot be negative at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:76) at org.apache.kafka.common.protocol.ApiKeys.parseResponse(ApiKeys.java:279) at org.apache.kafka.clients.NetworkClient.parseStructMaybeUpdateThrottleTimeMetrics(NetworkClient.java:586) at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:686) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:469) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:258) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:230) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:190) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.joinGroupIfNeeded(AbstractCoordinator.java:364) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:316) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:295) at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1146) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1111)

  On the other hand, when going the other way, there is no problem

  1. Start the first instance

  2. Instance 1 get all 32 partitions

  3. Start instance 2

  4. Re-balancing starts

  5. Instance 1 is loosing 16 parts

  6. instance 2 gets 16 parts

  Any idea what could cause this exception?