SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Addressing bugs reported by Sonarqube

Browse Source 
Resolves #747
This commit is contained in:
Soby Chacko
2019-10-25 16:42:02 -04:00
parent ad8e67fdc5
commit 06e5739fbd
4 changed files with 37 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderHealthIndicator.java
View File

@@ -115,7 +115,9 @@ public class KafkaStreamsBinderHealthIndicator extends AbstractHealthIndicator {
}
finally {
// Close admin client immediately.
adminClient.close(Duration.ofSeconds(0));
if (adminClient != null) {
adminClient.close(Duration.ofSeconds(0));
}
}
}

15
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CollectionSerde.java
View File

@@ -107,11 +107,12 @@ public class CollectionSerde<E> implements Serde<Collection<E>> {
*/
public CollectionSerde(Class<?> targetTypeForJsonSerde, Class<?> collectionsClass) {
this.collectionClass = collectionsClass;
JsonSerde<E> jsonSerde = new JsonSerde(targetTypeForJsonSerde);
try (JsonSerde<E> jsonSerde = new JsonSerde(targetTypeForJsonSerde)) {
this.inner = Serdes.serdeFrom(
new CollectionSerializer<>(jsonSerde.serializer()),
new CollectionDeserializer<>(jsonSerde.deserializer(), collectionsClass));
this.inner = Serdes.serdeFrom(
new CollectionSerializer<>(jsonSerde.serializer()),
new CollectionDeserializer<>(jsonSerde.deserializer(), collectionsClass));
}
}
@Override
@@ -204,8 +205,10 @@ public class CollectionSerde<E> implements Serde<Collection<E>> {
final int records = dataInputStream.readInt();
for (int i = 0; i < records; i++) {
final byte[] valueBytes = new byte[dataInputStream.readInt()];
dataInputStream.read(valueBytes);
collection.add(valueDeserializer.deserialize(topic, valueBytes));
final int read = dataInputStream.read(valueBytes);
if (read != -1) {
collection.add(valueDeserializer.deserialize(topic, valueBytes));
}
}
}
catch (IOException e) {

15
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicator.java
View File

@@ -99,15 +99,16 @@ public class KafkaBinderHealthIndicator implements HealthIndicator, DisposableBe
}
}
private synchronized Consumer<?, ?> initMetadataConsumer() {
if (this.metadataConsumer == null) {
this.metadataConsumer = this.consumerFactory.createConsumer();
}
return this.metadataConsumer;
}
private Health buildHealthStatus() {
try {
if (this.metadataConsumer == null) {
synchronized (KafkaBinderHealthIndicator.this) {
if (this.metadataConsumer == null) {
this.metadataConsumer = this.consumerFactory.createConsumer();
}
}
}
initMetadataConsumer();
synchronized (this.metadataConsumer) {
Set<String> downMessages = new HashSet<>();
final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = KafkaBinderHealthIndicator.this.binder

39
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java
View File

@@ -174,31 +174,26 @@ public class KafkaBinderMetrics
}
}
private ConsumerFactory<?, ?> createConsumerFactory() {
private synchronized ConsumerFactory<?, ?> createConsumerFactory() {
if (this.defaultConsumerFactory == null) {
synchronized (this) {
if (this.defaultConsumerFactory == null) {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class);
Map<String, Object> mergedConfig = this.binderConfigurationProperties
.mergedConsumerConfiguration();
if (!ObjectUtils.isEmpty(mergedConfig)) {
props.putAll(mergedConfig);
}
if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
this.binderConfigurationProperties
.getKafkaConnectionString());
}
this.defaultConsumerFactory = new DefaultKafkaConsumerFactory<>(
props);
}
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
ByteArrayDeserializer.class);
Map<String, Object> mergedConfig = this.binderConfigurationProperties
.mergedConsumerConfiguration();
if (!ObjectUtils.isEmpty(mergedConfig)) {
props.putAll(mergedConfig);
}
if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
this.binderConfigurationProperties
.getKafkaConnectionString());
}
this.defaultConsumerFactory = new DefaultKafkaConsumerFactory<>(
props);
}
return this.defaultConsumerFactory;
}