How I can change the groupId anonymous to yc-account?
1 reply
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
Map<String, Object> config = new HashMap<>();
setDefaults(config);
return new KafkaStreamsConfiguration(config);
}
@Bean(name = B_STREAMS_BUILDER)
public FactoryBean<StreamsBuilder> myKStreamBuilder(
@Qualifier(KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME) KafkaStreamsConfiguration streamsConfig
) {
return new StreamsBuilderFactoryBean(streamsConfig);
}@Bean("streamA")
public KStream<?, ?> kStream(@Qualifier(DEFAULT_STREAMS_BUILDER_BEAN_NAME) StreamsBuilder kStreamBuilder) {
...
}@Bean("streamB")
public KStream<?, ?> kStream(@Qualifier(B_STREAMS_BUILDER) StreamsBuilder kStreamBuilder) {
...
}
Hi, I'm trying to connect to an AWS MKS kafka cluster using IAM authentication, but my client isn't on EC2. I think I have to integrate this library https://github.com/aws/aws-msk-iam-auth, but the instructions the provide are for the Kafka client not Spring boot. It requires installing the following properties:
# Sets up TLS for encryption and SASL for authN.
security.protocol = SASL_SSL
# Identifies the SASL mechanism to use.
sasl.mechanism = AWS_MSK_IAM
# Binds SASL client implementation.
sasl.jaas.config = software.amazon.msk.auth.iam.IAMLoginModule required;
# Encapsulates constructing a SigV4 signature based on extracted credentials.
# The SASL client bound by "sasl.jaas.config" invokes this class.
sasl.client.callback.handler.class = software.amazon.msk.auth.iam.IAMClientCallbackHandlerSo I'm trying to adapt this to Spring Boot integration now, and I'm not sure how I can load the 'software.amazon.msk.auth.iam.IAMLoginModule' into the spring integration client. What properties would I use? Is this even possible?
4 replies
3 replies
Hi all. I've got an AVRO schema and generated my POJOs. I can consume the messages fine but when I want to turn the parent-POJO into a String with either ObjectMapper or ObjectWriter from Jackson, I get weird exceptions like Caused by: org.apache.avro.AvroRuntimeException: Not an array:.
Any chance anybody has a fix for this?
Have a great weekend. Cheers
2 replies
Hi again guys, I have a problem with the consumer objects usage, that it is leading to a ClassCastException, maybe you can throw some light into the issue.
First, I have this object generated by a schema (avro in this case), that is generated in the directory build/generated-main-avro-java/com/company/GeneratedObject.java
Then, I have the following KafkaListenerContainerFactory
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, GeneratedObject>>
generatedAvroContainerFactory() {
final ConcurrentKafkaListenerContainerFactory<String, GeneratedObject> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerAvroFactory());
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
return factory;
}With the consumer:
@KafkaListener(
topics = "topic",
containerFactory = "generatedAvroContainerFactory",
groupId = "${spring.kafka.consumer.group-id}"
)
public void consumer(ConsumerRecord<String, GeneratedObject> record, Acknowledgment acknowledgment) {Messages arrive to the consumer, but right after I try to access any field of the object,record.value().getField()
I get
Caused by: java.lang.ClassCastException: class org.apache.avro.generic.GenericData$Record cannot be cast to class com.company.GeneratedObject (org.apache.avro.generic.GenericData$Record and com.company.GeneratedObject are in unnamed module of loader 'app')
Could it be I'm missing any kind of configuration? Or does not have it any relation with spring-kafka?
6 replies
Does Kafka Producer/Consumer Api expose any metrics to monitor .I have spring actuator jars.Or how i can monitor themetrics ?
http://localhost:8080/actuator/metrics
1 reply
Here is there error "o.s.k.support.DefaultKafkaHeaderMapper : Could not decode json type: PublishProfilePreferences for key: DeviceMessageName"
This is not creating any issue but it a noise in the logs. Is there a way to avoid this header mapping? that is use of "spring_json_header_types"
5 replies
hello All, I have an issue with my consumer, I got this error :
o.a.k.c.c.CommitFailedException: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.I read about the fact that the time took by the message processing between the call to the poll() and the commit of the offset may be longer than auto.commit.interval.ms, is it the only cause for this problème ?
How can I know if my consumer is using batch mode or not ?
here is my configuration (I'm using spring cloud stream, but I think this have nothing to do with spring cloud stream but the configuration of the consumer ) :
cloud:
stream:
function:
routing:
enabled: true
source: testProducer
bindings:
functionRouter-in-0:
destination: test-consumer-topic
group: test-consumer
testProducer-out-0:
destination: test-producer-topic
producer:
useNativeEncoding: true
kafka:
binder:
brokers: 127.0.0.1:9092
auto-create-topics: false
headerMapperBeanName: kafkaHeaderMapper
configuration:
auto.offset.reset: earliest
producer-properties:
'[key.serializer]': org.apache.kafka.common.serialization.StringSerializer
'[value.serializer]': org.springframework.kafka.support.serializer.JsonSerializer
bindings:
testProducer-out-0:
producer:
messageKeyExpression: headers['kafka_messageKey']
2 replies
class org.apache.kafka.common.serialization.StringDeserializer is not an instance of org.apache.kafka.common.serialization.Serializer
Hello @garyrussell, please help us with our problem...
The goal we want to achieve is to receive all messages from each topic's partitions without using the consumer group, since the application can scale to several instances and each instance must read all messages from all partitions. At the same time, we have another restriction: while application is running, in the case a new topic's partition has been created, application(without restart) should receive messages from this new partition.
- How to create a consumer via @KafkaListener without a consumer group, so that the consumer will assign to all topic's partitions?
- How to make it so, that a consumer which was created without a consumer group automatically detects a newly created partition in the topic?
7 replies
1) resume the Spring Kafka Container
2)pause the Spring Kafka Container
I can pause the listener thread by calling PAUSE rest end point.Is there way to monitor background thread
10 replies
Hi @garyrussell
I’m trying the reactor-kafka, with sendTransactionally.
public Mono<Void> send(Flux<ProducerRecord<String, String>> records) {
return sender.sendTransactionally(Flux.just(records))
.flatMap(Function.identity())
.doOnNext(r -> log.info("Producer#send", r.recordMetadata().toString())).then();
}this send was called inside a flatMap, but I got the Invalid transition attempted from state IN_TRANSACTION to state IN_TRANSACTION a lot. What could be the problem? My transactionId is unique across nodes.
1 reply
6 replies
DefaultKafkaConsumerFactory or ConcurrentKafkaListenerContainerFactory to connect to 2 or more Kafka clusters, then bind it to a single @KafkaListenerwhich will process all the events?
3 replies
receiver.receive()
.doOnNext(record -> {
processMessage(record.value()).doOnSuccess(__ -> {
record.receiverOffset().acknowledge();
}).subscribe();
}).subscribe();hey team, can you confirm when using KafkaTemplate's method send(topic, key, value) it will calculate the partition based on the message key?
if you go to https://github.com/spring-projects/spring-kafka/blob/main/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java#L406
it basically creates a ProducerRecord, like this: https://github.com/a0x8o/kafka/blob/master/clients/src/main/java/org/apache/kafka/clients/producer/ProducerRecord.java#L133
which seems to send null as partition
I found this question https://stackoverflow.com/questions/45556142/spring-kafka-partitioning seems to confirm it, but where does it happen?
5 replies
8 replies
When i set ReceiverOptions.commintInterval in my java configuration, i don't see changes in consumer configuration. There only one property - auto.commit.interaval.ms.
Does ReceiverOptions.commintInterval(Duration d) in reactor-kafka overrides kafka property auto.commit.interval.ms?
When i use reactor-kafka can i set commit interval via kafka properties or i should set it exactly in ReceiverOptions configuration?
2 replies
If i use concatMap for process records via webflux, then in process only in one reactor-netty thread. When i use flat map - there are records processed concurrently by multiple netty threads, but there is no commit order. Should i use only concatMap for processing record via WebFlux?
1 reply
i have an existing
ConcurrentKafkaListenerContainerFactory called kafkaListenerContainerFactory and i want to add new listener with batch listener turned on, so i created another ConcurrentKafkaListenerContainerFactory called kafkaBatchListenerContainerFactory.after that i updated my existing
@KafkaListener to use the new batch listener containerFactory, the topic still the same.the problem is, the messages are consumed from the beginning. i just want to continue from the recent offset just like before. what did i do wrong here?
in case its related, my properties are
spring.kafka.consumer.group-id: test-group and auto-offset-reset: earliest
7 replies
12 replies
Guys, I need to make connection with ssl to kafka, that is I should prepare the kafka configuration server side as well.
I created the keystore.jks + truststore.jks which I read from zone : https://dzone.com/articles/kafka-ssl-client-authentication-in-multi-tenancy-a
keytool -keystore kafka.server.keystore.jks -alias localhost -validity 365 -genkey -keyalg RSA
openssl req -new -x509 -keyout ca-key -out ca-cert -days 365
keytool -keystore kafka.server.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.client1.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias localhost -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial
keytool -keystore kafka.server.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias localhost -import -file cert-signed
keytool -keystore kafka.client1.keystore.jks -alias localhost -validity 365 -genkey -keyalg RSA
keytool -keystore kafka.client1.keystore.jks -alias localhost -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days 365 -CAcreateserial
keytool -keystore kafka.client1.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.client1.keystore.jks -alias localhost -import -file cert-signed And also I change the config/service.properties like this:
listeners=PLAINTEXT://:9092,SSL://:9093
advertised.listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093
security.inter.broker.protocol=SSL
ssl.client.auth=required
ssl.keystore.location=/home/kafka/kafka/conf/kafka.server.keystore.jks
ssl.keystore.password=eg12345
ssl.truststore.location=/home/kafka/kafka/conf/kafka.server.truststore.jks
ssl.truststore.password=eg12345
ssl.key.password=eg12345
ssl.protocol=TLS
ssl.endpoint.identification.algorithm=
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
#ssl.keymanager.algorithm=SunX509
#ssl.trustmanager.algorithm=PKIX
ssl.keystore.type=JKS
ssl.truststore.type=JKS
#authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#allow.everyone.if.no.acl.found=true
javax.net.debug=sslThen kafka is going up without any issue.
but when ever I tried with client which is spring configured spring starter. I received the handshakes failed error form both side. kafka.log and client .
Also this is my config from spring side:
spring:
kafka:
bootstrap-servers:
- broker:9093
ssl:
key-password: eg12345
key-store-location: classpath:/jks/kafka.client.keystore.jks
key-store-password: eg12345
key-store-type: JKS
trust-store-location: classpath:/jks/kafka.client.truststore.jks
trust-store-password: eg12345
trust-store-type: JKS
protocol: SSL
consumer:
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
group-id: test-consumer-group
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializerError Receive from client side:[Consumer clientId=consumer-test-consumer-group-1, groupId=test-consumer-group] Bootstrap broker broker:9093 (id: -1 rack: null) disconnected
Error Receive from kafka side:[SocketServer brokerId=0] Failed authentication with /47.251.66.91 (SSL handshake failed) (org.apache.kafka.common.network.Selector)
2 replies
9 replies
So I'm trying out the Cloud Stream Kafka Binder:
I have a question regarding the functional programming model. My topology has two input topics and one output topic, so I'm using a BiFunction. In my case I need to do some other processing with the input Kstreams that I receive, like filtering and repartition the streams. So can I create two other Functions that process those input streams and then have the bifunction use them as parameters with Function Composition. or should I do all that processing in the BiFunction?
2 replies
2 replies
8 replies
3 replies