|
@@ -99,21 +99,24 @@ public class CloudKafkaConsumerFromEdgeTopic implements CommandLineRunner {
|
|
|
props.put("auto.offset.reset", autooffet);
|
|
|
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
|
|
|
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
|
|
|
- props.put("security.protocol", securtyprotocol);
|
|
|
- props.put("ssl.endpoint.identification.algorithm",algorithm);
|
|
|
+
|
|
|
+
|
|
|
// properties.put("ssl.truststore.location", ResourceUtils.getFile(truststorelocaltion).getPath());
|
|
|
// properties.put("ssl.truststore.password", keystorepassword);ResourceUtils.getFile(keystorelocaltion).getPath() ResourceUtils.getFile(truststorelocaltion).getPath()
|
|
|
if ("dev".equals(active)){
|
|
|
|
|
|
}else{
|
|
|
- props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, new File(keystorelocaltion).getAbsolutePath());
|
|
|
- props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, new File(truststorelocaltion).getAbsolutePath());
|
|
|
+// props.put("security.protocol", securtyprotocol);
|
|
|
+// props.put("ssl.endpoint.identification.algorithm",algorithm);
|
|
|
+// props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, new File(keystorelocaltion).getAbsolutePath());
|
|
|
+// props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, new File(truststorelocaltion).getAbsolutePath());
|
|
|
+// props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,keystorepassword);
|
|
|
+// props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG,keypassword);
|
|
|
+//
|
|
|
+// props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,truststorepassword);
|
|
|
}
|
|
|
|
|
|
- props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG,keystorepassword);
|
|
|
- props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG,keypassword);
|
|
|
|
|
|
- props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,truststorepassword);
|
|
|
// props.put("sasl.jaas.config",
|
|
|
// "org.apache.kafka.common.security.scram.ScramLoginModule required username='account_name' password='pwd';");
|
|
|
|
|
@@ -157,7 +160,7 @@ public class CloudKafkaConsumerFromEdgeTopic implements CommandLineRunner {
|
|
|
for (ConsumerRecord<String, String> record : records) {
|
|
|
System.out.println(">>>>>>>"+i);
|
|
|
System.out.println("topic = " + record.topic() + ", partition = " + record.partition()+", messages = "+ JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString());
|
|
|
- messageTcpSendQueue.offer(JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString(), 1000, TimeUnit.MICROSECONDS);
|
|
|
+ // messageTcpSendQueue.offer(JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString(), 1000, TimeUnit.MICROSECONDS);
|
|
|
// messageTcpSendQueue.add(JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString());
|
|
|
|
|
|
}
|