kafka将某些消息复制了数百次

我在使用kafka2.3.0的集群中使用3个kafka代理。然后我有一个流应用程序，它使用来自另一个kafka集群的数据，该集群转换数据并将这些数据推送到前面提到的3 kafka代理集群。
流应用程序有一个使用spring cloud stream greenwhich.sr1用java编写的生产者。此生产者使用以下代码推送消息：

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Slf4j
@Component
@EnableBinding(SensorDataBinding.class)
public class SensorDataProducer {

    private final SensorDataBinding sensorDataOut;
    private final long sendTimeoutInMilliseconds;

    public SensorDataProducer(SensorDataBinding binding,
                              @Value("${sendTimeoutInMilliseconds}") long sendTimeoutInMilliseconds) {
        this.sensorDataOut = binding;
        this.sendTimeoutInMilliseconds = sendTimeoutInMilliseconds;
    }

    public void produce(SensorData sensorMeasurement) {
        send(sensorMeasurement);
    }

    private void send(SensorData sensorMeasurement) {
        log.trace("sending message with contents: {}", sensorMeasurement.toString());
        Message<SensorData> message = MessageBuilder
                .withPayload(sensorMeasurement)
                .setHeader(KafkaHeaders.MESSAGE_KEY, getMessageKey(sensorMeasurement))
                .build();

        failSafeMessageSend(message);
    }

    private void failSafeMessageSend(Message<SensorData> message) {
        boolean sendSucceeded = false;
        do {
            try {
                this.sensorDataOut.sensorDataOut().send(message, this.sendTimeoutInMilliseconds);
                sendSucceeded = true;
            }
            catch (Exception ex) {
                log.error("Exception when sending message: {}", ex.getMessage());
            }
        }
        while (!sendSucceeded);
    }

    private byte[] getMessageKey(SensorData measurement) {
        return (measurement.getMessageKey()).getBytes();
    }
}

生产者配置：

ProducerConfig values: 
    acks = 1
    batch.size = 16384
    bootstrap.servers = [localhost:9095]
    buffer.memory = 33554432
    client.id = 
    compression.type = none
    connections.max.idle.ms = 540000
    enable.idempotence = false
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
    linger.ms = 0
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 0
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

ProducerConfig values: 
    acks = 1
    batch.size = 16384
    bootstrap.servers = [https://....:...]
    buffer.memory = 33554432
    client.id = client-1ae836b8-9a13-4903-aad6-09ce11a4be08-StreamThread-1-producer
    compression.type = none
    connections.max.idle.ms = 540000
    enable.idempotence = false
    interceptor.classes = []
    key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
    linger.ms = 100
    max.block.ms = 60000
    max.in.flight.requests.per.connection = 5
    max.request.size = 1048576
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
    receive.buffer.bytes = 32768
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retries = 10
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = SSL
    send.buffer.bytes = 131072
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
    ssl.endpoint.identification.algorithm = https
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.location = keystore/prod/client.keystore.p12
    ssl.keystore.password = [hidden]
    ssl.keystore.type = PKCS12
    ssl.protocol = TLS
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.location = keystore/prod/client.truststore.jks
    ssl.truststore.password = [hidden]
    ssl.truststore.type = JKS
    transaction.timeout.ms = 60000
    transactional.id = null
    value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

在我们的环境中，我们有这个应用程序的8个示例，它们是一个使用者组的一部分，消费来自外部集群上60个分区的主题。如前所述，这些数据在我们自己的3代理kafka集群设置中进行转换和推送。数据被推送到一个传感器数据主题，该主题有30个分区，保留时间为7天，并进行删除压缩。
我完全了解使用至少一次语义的重复消息，但我看到有些消息有时被重复300多次，导致所需的磁盘大小急剧增长，而其他消息最多被重复3或4次。
以下是一些度量的示例，这些度量显示消息被复制233次，其中时间戳数组显示每个消息的时间戳，偏移量数组显示被视为重复的每个消息的偏移量：

Key: b'100083952:300793850'|-|1591011300000. Value: {'count': 233, 'partition': 3, 'offset': 26637463, 'timestamps': [1594133472060, 1594133472062, 1594133472064, 1594133472066, 1594133472068, 1594133472071, 1594133472072, 1594133472074, 1594133472076, 1594133472081, 1594133472084, 1594133472085, 1594133472087, 1594133472090, 1594133472092, 1594133472095, 1594133472097, 1594133472099, 1594133472102, 1594133472103, 1594133472105, 1594133472107, 1594133472111, 1594133472113, 1594133472115, 1594133472117, 1594133472119, 1594133472121, 1594133472139, 1594133472141, 1594133472155, 1594133472157, 1594133472160, 1594133472163, 1594133472166, 1594133472169, 1594133472171, 1594133472174, 1594133472179, 1594133472181, 1594133472188, 1594133472190, 1594133472193, 1594133472195, 1594133472197, 1594133472199, 1594133472201, 1594133472204, 1594133472207, 1594133472209, 1594133472212, 1594133472215, 1594133472217, 1594133472219, 1594133472221, 1594133472222, 1594133472224, 1594133472227, 1594133472229, 1594133472231, 1594133472234, 1594133472235, 1594133472237, 1594133472240, 1594133472242, 1594133472243, 1594133472247, 1594133472249, 1594133472252, 1594133472255, 1594133472257, 1594133472262, 1594133472267, 1594133472270, 1594133472272, 1594133472275, 1594133472277, 1594133472279, 1594133472282, 1594133472289, 1594133472293, 1594133472295, 1594133472297, 1594133472299, 1594133472300, 1594133472303, 1594133472305, 1594133472307, 1594133472309, 1594133472311, 1594133472312, 1594133472316, 1594133472319, 1594133472321, 1594133472323, 1594133472325, 1594133472327, 1594133472333, 1594133472335, 1594133472337, 1594133472338, 1594133472341, 1594133472343, 1594133472346, 1594133472351, 1594133472356, 1594133472359, 1594133472361, 1594133472363, 1594133472364, 1594133472366, 1594133472371, 1594133472374, 1594133472377, 1594133472379, 1594133472381, 1594133472385, 1594133472393, 1594133472399, 1594133472401, 1594133472402, 1594133472404, 1594133472406, 1594133472409, 1594133472412, 1594133472413, 1594133472415, 1594133472417, 1594133472419, 1594133472421, 1594133472423, 1594133472425, 1594133472427, 1594133472429, 1594133472432, 1594133472434, 1594133472436, 1594133472439, 1594133472442, 1594133472469, 1594133472479, 1594133472483, 1594133472485, 1594133472488, 1594133472491, 1594133472494, 1594133472496, 1594133472498, 1594133472500, 1594133472503, 1594133472506, 1594133472508, 1594133472510, 1594133472512, 1594133472515, 1594133472520, 1594133472522, 1594133472524, 1594133472526, 1594133472528, 1594133472530, 1594133472532, 1594133472534, 1594133472535, 1594133472537, 1594133472539, 1594133472541, 1594133472543, 1594133472545, 1594133472547, 1594133472549, 1594133472551, 1594133472552, 1594133472554, 1594133472556, 1594133472558, 1594133472560, 1594133472562, 1594133472564, 1594133472566, 1594133472568, 1594133472570, 1594133472572, 1594133472573, 1594133472575, 1594133472577, 1594133472579, 1594133472581, 1594133472583, 1594133472587, 1594133472589, 1594133472593, 1594133472595, 1594133472596, 1594133472598, 1594133472599, 1594133472601, 1594133472603, 1594133472605, 1594133472606, 1594133472609, 1594133472611, 1594133472613, 1594133472615, 1594133472619, 1594133472622, 1594133472624, 1594133472626, 1594133472631, 1594133472633, 1594133472635, 1594133472637, 1594133472639, 1594133472641, 1594133472643, 1594133472644, 1594133472646, 1594133472649, 1594133472651, 1594133472653, 1594133472654, 1594133472657, 1594133472659, 1594133472660, 1594133472662, 1594133472664, 1594133472666, 1594133472667, 1594133472669, 1594133472671, 1594133472673, 1594133472675, 1594133472676], 
'offsets': 
[26637463, 26637464, 26637465, 26637466, 26637467, 26637468, 26637469, 26637470, 26637471, 26637472, 26637473, 26637474, 26637475, 26637476, 26637477, 26637478, 26637479, 26637480, 26637481, 26637482, 26637483, 26637484, 26637485, 26637486, 26637487, 26637488, 26637489, 26637490, 26637491, 26637492, 26637493, 26637494, 26637495, 26637496, 26637497, 26637498, 26637499, 26637500, 26637501, 26637502, 26637503, 26637504, 26637505, 26637506, 26637507, 26637508, 26637509, 26637510, 26637511, 26637512, 26637513, 26637514, 26637515, 26637516, 26637517, 26637518, 26637519, 26637520, 26637521, 26637522, 26637523, 26637524, 26637525, 26637526, 26637527, 26637528, 26637529, 26637530, 26637531, 26637532, 26637533, 26637534, 26637535, 26637536, 26637537, 26637538, 26637539, 26637540, 26637541, 26637542, 26637543, 26637544, 26637545, 26637546, 26637547, 26637548, 26637549, 26637550, 26637551, 26637552, 26637553, 26637554, 26637555, 26637556, 26637557, 26637558, 26637559, 26637560, 26637561, 26637562, 26637563, 26637564, 26637565, 26637566, 26637567, 26637568, 26637569, 26637570, 26637571, 26637572, 26637573, 26637574, 26637575, 26637576, 26637577, 26637578, 26637579, 26637580, 26637581, 26637582, 26637583, 26637584, 26637585, 26637586, 26637587, 26637588, 26637589, 26637590, 26637591, 26637592, 26637593, 26637594, 26637595, 26637596, 26637597, 26637598, 26637599, 26637600, 26637601, 26637602, 26637603, 26637604, 26637605, 26637606, 26637607, 26637608, 26637609, 26637610, 26637611, 26637612, 26637613, 26637614, 26637615, 26637616, 26637617, 26637618, 26637619, 26637620, 26637621, 26637622, 26637623, 26637624, 26637625, 26637626, 26637627, 26637628, 26637629, 26637630, 26637631, 26637632, 26637633, 26637634, 26637635, 26637636, 26637637, 26637638, 26637639, 26637640, 26637641, 26637642, 26637643, 26637644, 26637645, 26637646, 26637647, 26637648, 26637649, 26637650, 26637651, 26637652, 26637653, 26637654, 26637655, 26637656, 26637657, 26637658, 26637659, 26637660, 26637661, 26637662, 26637663, 26637664, 26637665, 26637666, 26637667, 26637668, 26637669, 26637670, 26637671, 26637672, 26637673, 26637674, 26637675, 26637676, 26637677, 26637678, 26637679, 26637680, 26637681, 26637682, 26637683, 26637684, 26637685, 26637686, 26637687, 26637688, 26637689, 26637690, 26637691, 26637692, 26637693, 26637694, 26637695], 
Date: 2020-06-01 13:35:00

可以看到偏移量单调增加。
我想知道是什么原因导致了只有几条重复消息和几百条重复消息之间的差异。我希望retries属性能够限制重复的数量，尽管它并没有真正显示出来。

kafka将某些消息复制了数百次

暂无答案！

相关问题

热门标签

最新问答