kafka consumer cannot connect

See also questions close to this topic

• Is there any way of changing a base image or a service and do not have to build the whole application again?

  I am new to docker and I have faced some complications understanding how big multi-level applications work with docker. I want to use visual studio Asp .Net core and firstly, I need to know If I make a trivial change to one service do I need to build the whole application and make a new image of it? How I am suppose to inform user of the change? Do I have to deploy the whole app again in case of a little change? Then I need to know that if production images do not have access to source code how the developer is supposed to make change to production image? Finally, I need to know that if my client is not willing to make a container of their database can I use a docker network to make a connection with that external SQL server?
• AWS codebuild error while building => YAML_FILE_ERROR Message: Runtime version selection is not supported by this build image

  I trying to build Docker image using AWS CodeBuild

  I follow AWS codebuild sample below https://docs.aws.amazon.com/codebuild/latest/userguide/sample-docker.html

  below is what i did.

  I write buildspec.yml like this.(exactly same as what I saw as previous link)

  and I start build project using aws CodeBuild, but I got this error message.

  Codebuild Log

  YAML_FILE_ERROR Message: Runtime version selection is not supported by this build image. <== this message means what? Is any other docker runtime version except 18? what should I do?

  please help...
• build riot helloworld with docker

  setup:

  1. docker pull riot/riotbuild
  2. wget https://github.com/RIOT-OS/RIOT/archive/2019.04.zip
  3. unzip 2019.04.zip
  4. cd RIOT-2019.04/examples/hello-world/

  5. make BUILD_IN_DOCKER=1 BOARD=stm32f4discovery all

     Is OK.

  6. make BUILD_IN_DOCKER=1 BOARD=stm32f4discovery flash

     This step Failed. Error is :

  >   Building application "hello-world" for "stm32f4discovery" with MCU
  > "stm32f4". "make" -C /data/riotbuild/riotbase/boards/stm32f4discovery
  > "make" -C /data/riotbuild/riotbase/core "make" -C
  > /data/riotbuild/riotbase/cpu/stm32f4 "make" -C
  > /data/riotbuild/riotbase/cpu/cortexm_common "make" -C
  > /data/riotbuild/riotbase/cpu/cortexm_common/periph "make" -C
  > /data/riotbuild/riotbase/cpu/stm32_common "make" -C
  > /data/riotbuild/riotbase/cpu/stm32_common/periph "make" -C
  > /data/riotbuild/riotbase/cpu/stm32f4/periph "make" -C
  > /data/riotbuild/riotbase/drivers "make" -C
  > /data/riotbuild/riotbase/drivers/periph_common "make" -C
  > /data/riotbuild/riotbase/sys "make" -C
  > /data/riotbuild/riotbase/sys/auto_init "make" -C
  > /data/riotbuild/riotbase/sys/isrpipe "make" -C
  > /data/riotbuild/riotbase/sys/newlib_syscalls_default "make" -C
  > /data/riotbuild/riotbase/sys/pm_layered "make" -C
  > /data/riotbuild/riotbase/sys/stdio_uart "make" -C
  > /data/riotbuild/riotbase/sys/tsrb    text    data     bss     dec    
  > hex filename    8756     140    2620   11516    2cfc
  > /data/riotbuild/riotbase/examples/hello-world/bin/stm32f4discovery/hello-world.elf
  > /home/huaxing/oproj/riot/RIOT-2019.04/dist/tools/openocd/openocd.sh
  > flash
  > /home/huaxing/oproj/riot/RIOT-2019.04/examples/hello-world/bin/stm32f4discovery/hello-world.elf
  > Flashing Target  sh: 1: openocd: not found
  > /home/huaxing/oproj/riot/RIOT-2019.04/examples/hello-world/../../Makefile.include:538:
  > recipe for target 'flash' failed make: *** [flash] Error 127 
  
• The latest (version: Fishtown.SR2) spring-cloud-stream doesn't support kafka client version 2.1.0 or higher?

  I am using latest Greenwich.SR1 which include spring-cloud-stream (version: Fishtown.SR2) and locally starting kafka client version:2.2.0 using kafka_2.12-2.2.0.jar

  I want to use the latest kafka client 2.1 or higher with spring-cloud-stream because it contains some important bug fixes. But when I run my spring app, its logs says

  INFO 37812 --- [main] o.a.kafka.common.utils.AppInfoParser : Kafka version : 2.0.1
  

  How can I use spring-cloud-stream with the latest kafka client? I want to use max.task.idle.ms from the latest StreamsConfig of kafka client but it seems the latest spring-cloud-stream kafka stream binder doesn't support kafka client 2.1.0 or higher version?
• Kafka Streams : ktable as lookup and destination to merge streams

  Hi I have 3 streams of events I want to merge using Kafka Streams.

  I can't find a simple solution to tackle the concurrency problem :

      // merged values Ktable
      KTable<String, ProdForecastPowerPlantAgg> mergedTable = builder.table(
              getMergedValuesTopic(),
              [...]);
  
  
      // records A
  
      // stream
      KStream<String, RecordA> recordsAStream = builder.stream(
              getRecordATopic(),
              [...]);
  
      // rekeyed stream
      KStream<String, ProductionRecordValue> recordsABySomeId = recordsAStream
              .selectKey((k, v) -> getKey(v);
  
  
      // records B
  
      // stream
      KStream<String, RecordB> recordsBStream = builder.stream(
              getRecordBTopic(),
              [...]);
  
      // rekeyed stream
      KStream<String, RecordB> recordsBBySomeId = recordsBStream
              .selectKey((k, v) -> getKey(v);
  
  
      // records C
  
      // stream
      KStream<String, RecordA> recordsCStream = builder.stream(
              getRecordCTopic(),
              [...]);
  
      // rekeyed stream
      KStream<String, ProductionRecordValue> recordsCBySomeId = recordsCStream
              .selectKey((k, v) -> getKey(v);
  
  
      // when a recordA arrives
      KStream<String, RecordA> aggFromA = recordsABySomeId
              .filter((k, v) -> v != null)
              // join recordA and current join result together
              .leftJoin(mergedTable, (recA, oldMerge) -> {
                          if (oldMerge != null) {
                              return new Merge(recA, oldMerge.B, oldMerge.C);
                          }
                          return new Merge(recA, null, null)
                      },
                      [...]
              );
  
      // when a recordB arrives
      KStream<String, RecordB> aggFromB = recordsBBySomeId
              .filter((k, v) -> v != null)
              // join recordB and current join result together
              .leftJoin(mergedTable, (recB, oldMerge) -> {
                          if (oldMerge != null) {
                              return new Merge(oldMerge.A, recB, oldMerge.C);
                          }
                          return new Merge(null, recB, null)
                      },
                      [...]
              );
  
  
      // when a recordC arrives
      KStream<String, RecordB> aggFromC = recordsCBySomeId
              .filter((k, v) -> v != null)
              // join recordB and current join result together
              .leftJoin(mergedTable, (recC, oldMerge) -> {
                          if (oldMerge != null) {
                              return new Merge(oldMerge.A, oldMerge.B, recC);
                          }
                          return new Merge(null, null, recC)
                      },
                      [...]
              );
  
  
      // save aggreagtion
  aggFromA.merge(aggFromB).merge(aggFromC)
              .to(getMergedValuesTopic(), Produced.with(Serdes.String(), aggSerdes));
  
  
  
      return builder.build();
  

  Indeed this snippet is invalid : the getMergedValuesTopic based KTable does not reflect the latest state of the merge when the lookup is done : when two different records arrive at the same time, one update can cancel the other (as the lookup is outdated).

  Does anyone have a simple solution to this problem, using Kafka streams ?
• find partition(s) assigned to Kafka stream instance

  I have a Kafka streaming application that subscribes to many topics and each topic has many partitions. When I create the application topology and start it, do I know what partitions of what topics are assigned to current instance of my application? I want to know this independent of any records are processed or not by this instance yet.

  I know when I get a record, that time I can do processorContext.partition() and processorContext.topic() to get partition/ topic information of current record being processed. But I am not looking for that.

  I am looking for an equivalent of KafkaConsumer.assigment on kafka streams side.

  I also tried the following code but I get size of s as 0.

  <Prepare builder and sconfig>
  kafkaStream = new KafkaStreams (builder, sconfig);
  kafkaStream.start ();
  Collection<StreamsMetadata> s = kafkaStream.allMetadata();
  System.out.println("StreamsMetadata: size is " + s.size());
  for (StreamsMetadata m : s) {
      Set<TopicPartition> tp = m.topicPartitions();
      System.out.println  ("TopicPartition: " + tp.toString());
  }