You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Sai Pradeep Amaravadhi (Jira)" <ji...@apache.org> on 2020/11/27 01:57:00 UTC
[jira] [Updated] (KAFKA-10771) Unable to consume data from kafka
[ https://issues.apache.org/jira/browse/KAFKA-10771?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Sai Pradeep Amaravadhi updated KAFKA-10771:
-------------------------------------------
Description:
Below is the code and i am unable to consume data because of "1966 [main] WARN org.apache.kafka.clients.NetworkClient - [Consumer clientId=di-claims-sub-cats-test, groupId=di-claims-sub-cats-test] Connection to node -7 (sp-dev-broker-6.streawsnprd.massmutual.com/10.97.217.35:9092) terminated during authentication. This may happen due to any of the following reasons: (1) Authentication failed due to invalid credentials with brokers older than 1.0.0, (2) Firewall blocking Kafka TLS traffic (eg it may only allow HTTPS traffic), (3) Transient network issue.",
Code:
package kafka;package kafka;
import java.util.Arrays;import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.config.SaslConfigs;import org.apache.kafka.common.config.SslConfigs;import org.apache.kafka.common.serialization.StringDeserializer;
import com.fabricapi.GlobalVariables;
public class kafkaLastRecord<K, V> extends java.lang.Object { private final static String BOOTSTRAP_SERVERS = "sp-dev-broker-0.streawsnprd.massmutual.com:9092,sp-dev-broker-1.streawsnprd.massmutual.com:9092,sp-dev-broker-2.streawsnprd.massmutual.com:9092,sp-dev-broker-3.streawsnprd.massmutual.com:9092, sp-dev-broker-4.streawsnprd.massmutual.com:9092,sp-dev-broker-5.streawsnprd.massmutual.com:9092,sp-dev-broker-6.streawsnprd.massmutual.com:9092"; private final static String TOPIC = "claims.event.diagreement.initial.request";
public static void main(String[] args) { Properties props = new Properties(); props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT"); props.put(SaslConfigs.SASL_MECHANISM, "PLAIN"); props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=" + '"' + "******" + '"' + " password=" + '"****' + "" + '"' + ";"); props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "jks file location"); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "*****"); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.GROUP_ID_CONFIG, "di-claims-sub-cats-test"); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1"); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
KafkaConsumer<String, String> consumer = null; // List of topics to subscribe to while (true) { try { consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(TOPIC)); ConsumerRecords<String, String> records = consumer.poll(100); int length = 0; for (ConsumerRecord<String, String> record : records)
{ length = length + 1; System.out.printf("Offset = %d\n", record.offset()); System.out.printf("Key = %s\n", record.key()); System.out.printf("Value = %s\n", record.value()); }
if (GlobalVariables.recordsCountBeforeSendingrequest == length) break; } catch (Exception e) \{ e.printStackTrace(); } finally \{ consumer.close(); } } }}
was:
Below is the code and i am unable to consume data because of "1966 [main] WARN org.apache.kafka.clients.NetworkClient - [Consumer clientId=di-claims-sub-cats-test, groupId=di-claims-sub-cats-test] Connection to node -7 (sp-dev-broker-6.streawsnprd.massmutual.com/10.97.217.35:9092) terminated during authentication. This may happen due to any of the following reasons: (1) Authentication failed due to invalid credentials with brokers older than 1.0.0, (2) Firewall blocking Kafka TLS traffic (eg it may only allow HTTPS traffic), (3) Transient network issue.",
Code:
package kafka;package kafka;
import java.util.Arrays;import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.config.SaslConfigs;import org.apache.kafka.common.config.SslConfigs;import org.apache.kafka.common.serialization.StringDeserializer;
import com.fabricapi.GlobalVariables;
public class kafkaLastRecord<K, V> extends java.lang.Object { private final static String BOOTSTRAP_SERVERS = "sp-dev-broker-0.streawsnprd.massmutual.com:9092,sp-dev-broker-1.streawsnprd.massmutual.com:9092,sp-dev-broker-2.streawsnprd.massmutual.com:9092,sp-dev-broker-3.streawsnprd.massmutual.com:9092, sp-dev-broker-4.streawsnprd.massmutual.com:9092,sp-dev-broker-5.streawsnprd.massmutual.com:9092,sp-dev-broker-6.streawsnprd.massmutual.com:9092"; private final static String TOPIC = "claims.event.diagreement.initial.request";
public static void main(String[] args) { Properties props = new Properties(); props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT"); props.put(SaslConfigs.SASL_MECHANISM, "PLAIN"); props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=" + '"' + "di-claims-app" + '"' + " password=" + '"' + "di-claims-secret" + '"' + ";"); props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "/src/test/resources/data/mm-cert-bundle.jks"); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "truststorepass"); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.GROUP_ID_CONFIG, "di-claims-sub-cats-test"); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1"); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
KafkaConsumer<String, String> consumer = null; // List of topics to subscribe to while (true) \{ try { consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(TOPIC)); ConsumerRecords<String, String> records = consumer.poll(100); int length = 0; for (ConsumerRecord<String, String> record : records) { length = length + 1; System.out.printf("Offset = %d\n", record.offset()); System.out.printf("Key = %s\n", record.key()); System.out.printf("Value = %s\n", record.value()); } if (GlobalVariables.recordsCountBeforeSendingrequest == length) break; } catch (Exception e) \{ e.printStackTrace(); } finally \{ consumer.close(); } } }}
> Unable to consume data from kafka
> ---------------------------------
>
> Key: KAFKA-10771
> URL: https://issues.apache.org/jira/browse/KAFKA-10771
> Project: Kafka
> Issue Type: Bug
> Components: consumer
> Reporter: Sai Pradeep Amaravadhi
> Priority: Major
>
> Below is the code and i am unable to consume data because of "1966 [main] WARN org.apache.kafka.clients.NetworkClient - [Consumer clientId=di-claims-sub-cats-test, groupId=di-claims-sub-cats-test] Connection to node -7 (sp-dev-broker-6.streawsnprd.massmutual.com/10.97.217.35:9092) terminated during authentication. This may happen due to any of the following reasons: (1) Authentication failed due to invalid credentials with brokers older than 1.0.0, (2) Firewall blocking Kafka TLS traffic (eg it may only allow HTTPS traffic), (3) Transient network issue.",
>
> Code:
> package kafka;package kafka;
> import java.util.Arrays;import java.util.Properties;
> import org.apache.kafka.clients.CommonClientConfigs;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.config.SaslConfigs;import org.apache.kafka.common.config.SslConfigs;import org.apache.kafka.common.serialization.StringDeserializer;
> import com.fabricapi.GlobalVariables;
> public class kafkaLastRecord<K, V> extends java.lang.Object { private final static String BOOTSTRAP_SERVERS = "sp-dev-broker-0.streawsnprd.massmutual.com:9092,sp-dev-broker-1.streawsnprd.massmutual.com:9092,sp-dev-broker-2.streawsnprd.massmutual.com:9092,sp-dev-broker-3.streawsnprd.massmutual.com:9092, sp-dev-broker-4.streawsnprd.massmutual.com:9092,sp-dev-broker-5.streawsnprd.massmutual.com:9092,sp-dev-broker-6.streawsnprd.massmutual.com:9092"; private final static String TOPIC = "claims.event.diagreement.initial.request";
> public static void main(String[] args) { Properties props = new Properties(); props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT"); props.put(SaslConfigs.SASL_MECHANISM, "PLAIN"); props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=" + '"' + "******" + '"' + " password=" + '"****' + "" + '"' + ";"); props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "jks file location"); props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "*****"); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); props.put(ConsumerConfig.GROUP_ID_CONFIG, "di-claims-sub-cats-test"); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1"); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
> KafkaConsumer<String, String> consumer = null; // List of topics to subscribe to while (true) { try { consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList(TOPIC)); ConsumerRecords<String, String> records = consumer.poll(100); int length = 0; for (ConsumerRecord<String, String> record : records)
> { length = length + 1; System.out.printf("Offset = %d\n", record.offset()); System.out.printf("Key = %s\n", record.key()); System.out.printf("Value = %s\n", record.value()); }
> if (GlobalVariables.recordsCountBeforeSendingrequest == length) break; } catch (Exception e) \{ e.printStackTrace(); } finally \{ consumer.close(); } } }}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)