Skip to content

Commit ceaf437

Browse files
garyrussellartembilan
garyrussell
authored and
artembilan
committed
GH-2164: Remove More Dead Code
1 parent 03a2124 commit ceaf437

11 files changed

+73
-177
lines changed

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

-2
Original file line numberDiff line numberDiff line change
@@ -437,7 +437,6 @@ public void setMaxAge(Duration maxAge) {
437437
* properties applied
438438
*/
439439
@Override
440-
@SuppressWarnings("deprecation")
441440
public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> overrideProperties) {
442441
Map<String, Object> producerProperties = new HashMap<>(getConfigurationProperties());
443442
producerProperties.putAll(overrideProperties);
@@ -447,7 +446,6 @@ public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> o
447446
getKeySerializerSupplier(),
448447
getValueSerializerSupplier());
449448
newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds());
450-
newFactory.setProducerPerConsumerPartition(isProducerPerConsumerPartition());
451449
newFactory.setProducerPerThread(isProducerPerThread());
452450
for (ProducerPostProcessor<K, V> templatePostProcessor : getPostProcessors()) {
453451
newFactory.addPostProcessor(templatePostProcessor);

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

-12
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,6 @@
5959
import org.springframework.kafka.support.ProducerListener;
6060
import org.springframework.kafka.support.SendResult;
6161
import org.springframework.kafka.support.TopicPartitionOffset;
62-
import org.springframework.kafka.support.TransactionSupport;
6362
import org.springframework.kafka.support.converter.MessagingMessageConverter;
6463
import org.springframework.kafka.support.converter.RecordMessageConverter;
6564
import org.springframework.kafka.support.micrometer.MicrometerHolder;
@@ -496,14 +495,6 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
496495
Assert.state(this.transactional, "Producer factory does not support transactions");
497496
Producer<K, V> producer = this.producers.get();
498497
Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");
499-
String transactionIdSuffix;
500-
if (this.producerFactory.isProducerPerConsumerPartition()) {
501-
transactionIdSuffix = TransactionSupport.getTransactionIdSuffix();
502-
TransactionSupport.clearTransactionIdSuffix();
503-
}
504-
else {
505-
transactionIdSuffix = null;
506-
}
507498

508499
producer = this.producerFactory.createProducer(this.transactionIdPrefix);
509500

@@ -534,9 +525,6 @@ public <T> T executeInTransaction(OperationsCallback<K, V, T> callback) {
534525
throw e;
535526
}
536527
finally {
537-
if (transactionIdSuffix != null) {
538-
TransactionSupport.setTransactionIdSuffix(transactionIdSuffix);
539-
}
540528
this.producers.remove();
541529
closeProducer(producer, false);
542530
}

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/core/ProducerFactory.java

+4
Original file line numberDiff line numberDiff line change
@@ -92,7 +92,11 @@ default void closeProducerFor(String transactionIdSuffix) {
9292
* Return the producerPerConsumerPartition.
9393
* @return the producerPerConsumerPartition.
9494
* @since 1.3.8
95+
* @deprecated no longer necessary because
96+
* {@code org.springframework.kafka.listener.ContainerProperties.EOSMode#V1} is no
97+
* longer supported.
9598
*/
99+
@Deprecated
96100
default boolean isProducerPerConsumerPartition() {
97101
return false;
98102
}

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

+49-113
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,6 @@
7676
import org.springframework.kafka.KafkaException;
7777
import org.springframework.kafka.core.ConsumerFactory;
7878
import org.springframework.kafka.core.KafkaResourceHolder;
79-
import org.springframework.kafka.core.ProducerFactory;
8079
import org.springframework.kafka.event.ConsumerFailedToStartEvent;
8180
import org.springframework.kafka.event.ConsumerPartitionPausedEvent;
8281
import org.springframework.kafka.event.ConsumerPartitionResumedEvent;
@@ -102,7 +101,6 @@
102101
import org.springframework.kafka.support.LogIfLevelEnabled;
103102
import org.springframework.kafka.support.TopicPartitionOffset;
104103
import org.springframework.kafka.support.TopicPartitionOffset.SeekPosition;
105-
import org.springframework.kafka.support.TransactionSupport;
106104
import org.springframework.kafka.support.micrometer.MicrometerHolder;
107105
import org.springframework.kafka.support.serializer.DeserializationException;
108106
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
@@ -738,8 +736,6 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
738736

739737
private Producer<?, ?> producer;
740738

741-
private boolean producerPerConsumerPartition;
742-
743739
private boolean commitRecovered;
744740

745741
private boolean wasIdle;
@@ -1022,10 +1018,6 @@ boolean isPartitionPaused(TopicPartition topicPartition) {
10221018

10231019
@Nullable
10241020
private TransactionTemplate determineTransactionTemplate() {
1025-
if (this.kafkaTxManager != null) {
1026-
this.producerPerConsumerPartition =
1027-
this.kafkaTxManager.getProducerFactory().isProducerPerConsumerPartition();
1028-
}
10291021
if (this.transactionManager != null) {
10301022
TransactionTemplate template = new TransactionTemplate(this.transactionManager);
10311023
TransactionDefinition definition = this.containerProperties.getTransactionDefinition();
@@ -1740,9 +1732,6 @@ private void wrapUp(@Nullable Throwable throwable) {
17401732
// No-op. Continue process
17411733
}
17421734
}
1743-
else {
1744-
closeProducers(partitions);
1745-
}
17461735
}
17471736
else {
17481737
this.logger.error("Fatal consumer exception; stopping container");
@@ -1995,11 +1984,6 @@ private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
19951984
@Nullable final List<ConsumerRecord<K, V>> recordList) {
19961985

19971986
try {
1998-
if (this.subBatchPerPartition && this.producerPerConsumerPartition) {
1999-
ConsumerRecord<K, V> record = recordList == null ? records.iterator().next() : recordList.get(0);
2000-
TransactionSupport
2001-
.setTransactionIdSuffix(zombieFenceTxIdSuffix(record.topic(), record.partition())); // NOSONAR
2002-
}
20031987
this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
20041988

20051989
@Override
@@ -2028,11 +2012,6 @@ public void doInTransactionWithoutResult(TransactionStatus s) {
20282012
this.logger.error(e, "Transaction rolled back");
20292013
batchRollback(records, recordList, e);
20302014
}
2031-
finally {
2032-
if (this.subBatchPerPartition && this.producerPerConsumerPartition) {
2033-
TransactionSupport.clearTransactionIdSuffix();
2034-
}
2035-
}
20362015
}
20372016

20382017
private void batchRollback(final ConsumerRecords<K, V> records,
@@ -2327,11 +2306,6 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
23272306
this.logger.error(ex, "Transaction rolled back");
23282307
recordAfterRollback(iterator, record, ex);
23292308
}
2330-
finally {
2331-
if (this.producerPerConsumerPartition) {
2332-
TransactionSupport.clearTransactionIdSuffix();
2333-
}
2334-
}
23352309
if (this.commonRecordInterceptor != null) {
23362310
this.commonRecordInterceptor.afterRecord(record, this.consumer);
23372311
}
@@ -2344,10 +2318,6 @@ private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
23442318
}
23452319

23462320
private void invokeInTransaction(Iterator<ConsumerRecord<K, V>> iterator, final ConsumerRecord<K, V> record) {
2347-
if (this.producerPerConsumerPartition) {
2348-
TransactionSupport
2349-
.setTransactionIdSuffix(zombieFenceTxIdSuffix(record.topic(), record.partition()));
2350-
}
23512321
this.transactionTemplate.execute(new TransactionCallbackWithoutResult() {
23522322

23532323
@Override
@@ -3109,25 +3079,6 @@ public String toString() {
31093079
+ "\n]";
31103080
}
31113081

3112-
private void closeProducers(@Nullable Collection<TopicPartition> partitions) {
3113-
if (partitions != null) {
3114-
ProducerFactory<?, ?> producerFactory = this.kafkaTxManager.getProducerFactory();
3115-
partitions.forEach(tp -> {
3116-
try {
3117-
producerFactory.closeProducerFor(zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
3118-
}
3119-
catch (Exception e) {
3120-
this.logger.error(e, () -> "Failed to close producer with transaction id suffix: "
3121-
+ zombieFenceTxIdSuffix(tp.topic(), tp.partition()));
3122-
}
3123-
});
3124-
}
3125-
}
3126-
3127-
private String zombieFenceTxIdSuffix(String topic, int partition) {
3128-
return this.consumerGroupId + "." + topic + "." + partition;
3129-
}
3130-
31313082
private final class ConsumerAcknowledgment implements Acknowledgment {
31323083

31333084
private final ConsumerRecord<K, V> record;
@@ -3246,47 +3197,40 @@ private class ListenerConsumerRebalanceListener implements ConsumerRebalanceList
32463197

32473198
@Override
32483199
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
3200+
if (this.consumerAwareListener != null) {
3201+
this.consumerAwareListener.onPartitionsRevokedBeforeCommit(ListenerConsumer.this.consumer,
3202+
partitions);
3203+
}
3204+
else {
3205+
this.userListener.onPartitionsRevoked(partitions);
3206+
}
32493207
try {
3250-
if (this.consumerAwareListener != null) {
3251-
this.consumerAwareListener.onPartitionsRevokedBeforeCommit(ListenerConsumer.this.consumer,
3252-
partitions);
3253-
}
3254-
else {
3255-
this.userListener.onPartitionsRevoked(partitions);
3256-
}
3257-
try {
3258-
// Wait until now to commit, in case the user listener added acks
3259-
commitPendingAcks();
3260-
fixTxOffsetsIfNeeded();
3261-
}
3262-
catch (Exception e) {
3263-
ListenerConsumer.this.logger.error(e, () -> "Fatal commit error after revocation "
3264-
+ partitions);
3265-
}
3266-
if (this.consumerAwareListener != null) {
3267-
this.consumerAwareListener.onPartitionsRevokedAfterCommit(ListenerConsumer.this.consumer,
3268-
partitions);
3269-
}
3270-
if (ListenerConsumer.this.consumerSeekAwareListener != null) {
3271-
ListenerConsumer.this.consumerSeekAwareListener.onPartitionsRevoked(partitions);
3272-
}
3273-
if (ListenerConsumer.this.assignedPartitions != null) {
3274-
ListenerConsumer.this.assignedPartitions.removeAll(partitions);
3275-
}
3276-
ListenerConsumer.this.pausedForNack.removeAll(partitions);
3277-
partitions.forEach(tp -> ListenerConsumer.this.lastCommits.remove(tp));
3278-
synchronized (ListenerConsumer.this) {
3279-
if (ListenerConsumer.this.offsetsInThisBatch != null) {
3280-
partitions.forEach(tp -> {
3281-
ListenerConsumer.this.offsetsInThisBatch.remove(tp);
3282-
ListenerConsumer.this.deferredOffsets.remove(tp);
3283-
});
3284-
}
3285-
}
3208+
// Wait until now to commit, in case the user listener added acks
3209+
commitPendingAcks();
3210+
fixTxOffsetsIfNeeded();
32863211
}
3287-
finally {
3288-
if (ListenerConsumer.this.kafkaTxManager != null) {
3289-
closeProducers(partitions);
3212+
catch (Exception e) {
3213+
ListenerConsumer.this.logger.error(e, () -> "Fatal commit error after revocation "
3214+
+ partitions);
3215+
}
3216+
if (this.consumerAwareListener != null) {
3217+
this.consumerAwareListener.onPartitionsRevokedAfterCommit(ListenerConsumer.this.consumer,
3218+
partitions);
3219+
}
3220+
if (ListenerConsumer.this.consumerSeekAwareListener != null) {
3221+
ListenerConsumer.this.consumerSeekAwareListener.onPartitionsRevoked(partitions);
3222+
}
3223+
if (ListenerConsumer.this.assignedPartitions != null) {
3224+
ListenerConsumer.this.assignedPartitions.removeAll(partitions);
3225+
}
3226+
ListenerConsumer.this.pausedForNack.removeAll(partitions);
3227+
partitions.forEach(tp -> ListenerConsumer.this.lastCommits.remove(tp));
3228+
synchronized (ListenerConsumer.this) {
3229+
if (ListenerConsumer.this.offsetsInThisBatch != null) {
3230+
partitions.forEach(tp -> {
3231+
ListenerConsumer.this.offsetsInThisBatch.remove(tp);
3232+
ListenerConsumer.this.deferredOffsets.remove(tp);
3233+
});
32903234
}
32913235
}
32923236
}
@@ -3349,33 +3293,25 @@ private void commitCurrentOffsets(Map<TopicPartition, OffsetAndMetadata> offsets
33493293
if (ListenerConsumer.this.transactionTemplate != null
33503294
&& ListenerConsumer.this.kafkaTxManager != null
33513295
&& !AssignmentCommitOption.LATEST_ONLY_NO_TX.equals(ListenerConsumer.this.autoCommitOption)) {
3352-
try {
3353-
offsetsToCommit.forEach((partition, offsetAndMetadata) -> {
3354-
if (ListenerConsumer.this.producerPerConsumerPartition) {
3355-
TransactionSupport.setTransactionIdSuffix(
3356-
zombieFenceTxIdSuffix(partition.topic(), partition.partition()));
3357-
}
3358-
ListenerConsumer.this.transactionTemplate
3359-
.execute(new TransactionCallbackWithoutResult() {
3360-
3361-
@Override
3362-
protected void doInTransactionWithoutResult(TransactionStatus status) {
3363-
KafkaResourceHolder<?, ?> holder =
3364-
(KafkaResourceHolder<?, ?>) TransactionSynchronizationManager
3365-
.getResource(ListenerConsumer.this.kafkaTxManager
3366-
.getProducerFactory());
3367-
if (holder != null) {
3368-
doSendOffsets(holder.getProducer(),
3369-
Collections.singletonMap(partition, offsetAndMetadata));
3370-
}
3296+
3297+
offsetsToCommit.forEach((partition, offsetAndMetadata) -> {
3298+
ListenerConsumer.this.transactionTemplate
3299+
.execute(new TransactionCallbackWithoutResult() {
3300+
3301+
@Override
3302+
protected void doInTransactionWithoutResult(TransactionStatus status) {
3303+
KafkaResourceHolder<?, ?> holder =
3304+
(KafkaResourceHolder<?, ?>) TransactionSynchronizationManager
3305+
.getResource(ListenerConsumer.this.kafkaTxManager
3306+
.getProducerFactory());
3307+
if (holder != null) {
3308+
doSendOffsets(holder.getProducer(),
3309+
Collections.singletonMap(partition, offsetAndMetadata));
33713310
}
3311+
}
33723312

3373-
});
3374-
});
3375-
}
3376-
finally {
3377-
TransactionSupport.clearTransactionIdSuffix();
3378-
}
3313+
});
3314+
});
33793315
}
33803316
else {
33813317
ContainerProperties containerProps = KafkaMessageListenerContainer.this.getContainerProperties();

Diff for: ‎spring-kafka/src/main/java/org/springframework/kafka/support/TransactionSupport.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2019 the original author or authors.
2+
* Copyright 2018-2022 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -21,8 +21,10 @@
2121
*
2222
* @author Gary Russell
2323
* @since 1.3.7
24+
* @deprecated No longer used.
2425
*
2526
*/
27+
@Deprecated
2628
public final class TransactionSupport {
2729

2830
private static final ThreadLocal<String> transactionIdSuffix = new ThreadLocal<>(); // NOSONAR

Diff for: ‎spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaProducerFactoryTests.java

-3
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,6 @@
5454
import org.springframework.context.ApplicationContext;
5555
import org.springframework.context.event.ContextStoppedEvent;
5656
import org.springframework.kafka.core.ProducerFactory.Listener;
57-
import org.springframework.kafka.support.TransactionSupport;
5857
import org.springframework.kafka.test.utils.KafkaTestUtils;
5958
import org.springframework.kafka.transaction.KafkaTransactionManager;
6059
import org.springframework.transaction.CannotCreateTransactionException;
@@ -367,7 +366,6 @@ public void producerRemoved(String id, Producer producer) {
367366
assertThat(adds).hasSize(2);
368367
assertThat(removals).hasSize(2);
369368

370-
TransactionSupport.setTransactionIdSuffix("xx");
371369
pf.createProducer("tx").close();
372370
assertThat(adds).hasSize(3);
373371
assertThat(removals).hasSize(2);
@@ -387,7 +385,6 @@ public void producerRemoved(String id, Producer producer) {
387385
pf.reset();
388386
assertThat(adds).hasSize(4);
389387
assertThat(removals).hasSize(4);
390-
TransactionSupport.clearTransactionIdSuffix();
391388

392389
pf.setProducerPerThread(true);
393390
pf.createProducer().close();

Diff for: ‎spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

-1
Original file line numberDiff line numberDiff line change
@@ -516,7 +516,6 @@ void testConfigOverridesWithDefaultKafkaProducerFactory() {
516516
assertThat(template.getProducerFactory().getConfigurationProperties()
517517
.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)).isEqualTo(StringSerializer.class);
518518
assertThat(template.getProducerFactory().getPhysicalCloseTimeout()).isEqualTo(Duration.ofSeconds(6));
519-
assertThat(template.getProducerFactory().isProducerPerConsumerPartition()).isFalse();
520519
assertThat(template.getProducerFactory().isProducerPerThread()).isTrue();
521520
assertThat(template.isTransactional()).isTrue();
522521
assertThat(template.getProducerFactory().getListeners()).isEqualTo(pf.getListeners());

0 commit comments

Comments
 (0)