Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java (Java)

Code before:
                        .setTopicId(topicId).setName(tp.topic());
                tpd.add(tpData);
            }

            tpData.partitionData().add(new ProduceRequestData.PartitionProduceData()
                    .setIndex(tp.partition())
                    .setRecords(records));
            recordsByPartition.put(tp, batch);
        }

        String transactionalId = null;
        boolean useTransactionV1Version = false;
        if (transactionManager != null && transactionManager.isTransactional()) {
            transactionalId = transactionManager.transactionalId();
            useTransactionV1Version = !transactionManager.isTransactionV2Enabled();
        }

        ProduceRequest.Builder requestBuilder = ProduceRequest.builder(
                new ProduceRequestData()
                        .setAcks(acks)
                        .setTimeoutMs(timeout)
                        .setTransactionalId(transactionalId)
                        .setTopicData(tpd),
                useTransactionV1Version
        );
        RequestCompletionHandler callback = response -> handleProduceResponse(response, recordsByPartition, metadata.topicNames(), time.milliseconds());

        String nodeId = Integer.toString(destination);
        ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0,
                requestTimeoutMs, callback);
        client.send(clientRequest, now);
        log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
    }

    private Map<String, Uuid> topicIdsForBatches(List<ProducerBatch> batches) {
        return batches.stream()
                .collect(Collectors.toMap(
                        b -> b.topicPartition.topic(),
                        b -> metadata.topicIds().getOrDefault(b.topicPartition.topic(), Uuid.ZERO_UUID),
                        (existing, replacement) -> replacement)
                );
    }

    /**
     * Wake up the selector associated with this send thread
     */
    public void wakeup() {
        this.client.wakeup();
    }

    public static Sensor throttleTimeSensor(SenderMetricsRegistry metrics) {

Review comment (suggestion):
@lucasbru : Thanks for reporting the issue. I think this is the problem. This callback is called when the response completes, at which point, the metadata could have changed. Calling `metadata.topicNames()` at that point will potentially return metadata inconsistent with what's being used when generating the produce request. So, we need to call `metadata.topicNames()` outside the callback, save it as a final and pass it to the callback.

Code after:
                tpData = new ProduceRequestData.TopicProduceData()
                        .setTopicId(topicId).setName(tp.topic());
                tpd.add(tpData);
            }

            tpData.partitionData().add(new ProduceRequestData.PartitionProduceData()
                    .setIndex(tp.partition())
                    .setRecords(records));
            recordsByPartition.put(tp, batch);
        }

        String transactionalId = null;
        boolean useTransactionV1Version = false;
        if (transactionManager != null && transactionManager.isTransactional()) {
            transactionalId = transactionManager.transactionalId();
            useTransactionV1Version = !transactionManager.isTransactionV2Enabled();
        }

        ProduceRequest.Builder requestBuilder = ProduceRequest.builder(
                new ProduceRequestData()
                        .setAcks(acks)
                        .setTimeoutMs(timeout)
                        .setTransactionalId(transactionalId)
                        .setTopicData(tpd),
                useTransactionV1Version
        );
        // Fetch topic names from metadata outside callback as topic ids may change during the callback
        // for example if topic was recreated.
        Map<Uuid, String> topicNames = metadata.topicNames();

        RequestCompletionHandler callback = response -> handleProduceResponse(response, recordsByPartition, topicNames, time.milliseconds());

        String nodeId = Integer.toString(destination);
        ClientRequest clientRequest = client.newClientRequest(nodeId, requestBuilder, now, acks != 0,
                requestTimeoutMs, callback);
        client.send(clientRequest, now);
        log.trace("Sent produce request to {}: {}", nodeId, requestBuilder);
    }

    private Map<String, Uuid> topicIdsForBatches(List<ProducerBatch> batches) {
        return batches.stream()
                .collect(Collectors.toMap(
                        b -> b.topicPartition.topic(),
                        b -> metadata.topicIds().getOrDefault(b.topicPartition.topic(), Uuid.ZERO_UUID),
                        (existing, replacement) -> replacement)
                );
    }

    /**
     * Wake up the selector associated with this send thread
     */