[jira] [Created] (KAFKA-10428) Mirror Maker connect applies base64 encoding to string headers
Jennifer Thompson created KAFKA-10428: - Summary: Mirror Maker connect applies base64 encoding to string headers Key: KAFKA-10428 URL: https://issues.apache.org/jira/browse/KAFKA-10428 Project: Kafka Issue Type: Bug Components: mirrormaker Affects Versions: 2.6.0, 2.5.0, 2.4.0 Reporter: Jennifer Thompson MirrorSourceTask takes the header value as bytes from the ConsumerRecord, which does not have a header schema, and adds it to the SourceRecord headers using "addBytes". This uses Schema.BYTES as the schema for the header, and somehow, base64 encoding gets applied when the record gets committed. This means that my original header value "with_headers" (created with a python producer, and stored as a 12 character byte array) becomes the string value "d2l0aF9oZWFkZXJz", a 16 character byte array, which is the base64 encoded version of the original. If I try to preempt this using "d2l0aF9oZWFkZXJz" to start with, and base64 encoding the headers everywhere, it just gets double encoded to "ZDJsMGFGOW9aV0ZrWlhKeg==" after passing through the MirrorSourceTask. I think the base64 encoding may be coming from Values#append (https://github.com/apache/kafka/blob/trunk/connect/api/src/main/java/org/apache/kafka/connect/data/Values.java#L674), but I'm not sure how. That is invoked by SimpleConnectorHeader#fromConnectHeader via Values#convertToString. SimpleHeaderConverter#toConnectHeader produces the correct schema in this case, and solves the problem for me, but it seems to guess at the schema, so I'm not sure if it is the right solution. Since schemas seem to be required for SourceRecord headers, but not available from ConsumerRecord headers, I'm not sure what other option we have. I will open a PR with this solution -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Resolved] (KAFKA-7845) Kafka clients do not re-resolve ips when a broker is replaced.
[
https://issues.apache.org/jira/browse/KAFKA-7845?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jennifer Thompson resolved KAFKA-7845.
--
Resolution: Fixed
Fix Version/s: 2.1.1
> Kafka clients do not re-resolve ips when a broker is replaced.
> --
>
> Key: KAFKA-7845
> URL: https://issues.apache.org/jira/browse/KAFKA-7845
> Project: Kafka
> Issue Type: Bug
> Components: clients
>Affects Versions: 2.1.0
>Reporter: Jennifer Thompson
>Priority: Major
> Fix For: 2.1.1
>
>
> When one of our Kafka brokers dies and a new one replaces it (via an aws
> ASG), the clients that publish to Kafka still try to publish to the old
> brokers.
> We see errors like
> {code:java}
> 2019-01-18 20:16:16 WARN NetworkClient:721 - [Producer clientId=producer-1]
> Connection to node 2 (/10.130.98.111:9092) could not be established. Broker
> may not be available.
> 2019-01-18 20:19:09 WARN Sender:596 - [Producer clientId=producer-1] Got
> error produce response with correlation id 3414 on topic-partition aa.pga-2,
> retrying (4 attempts left). Error: NOT_LEADER_FOR_PARTITION
> 2019-01-18 20:19:09 WARN Sender:641 - [Producer clientId=producer-1] Received
> invalid metadata error in produce request on partition aa.pga-2 due to
> org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is
> not the leader for that topic-partition.. Going to request metadata update now
> 2019-01-18 20:21:19 WARN NetworkClient:721 - [Producer clientId=producer-1]
> Connection to node 2 (/10.130.98.111:9092) could not be established. Broker
> may not be available.
> 2019-01-18 20:21:50 ERROR ProducerBatch:233 - Error executing user-provided
> callback on message for topic-partition 'aa.test-liz-0'{code}
> and
> {code:java}
> [2019-01-18 20:28:47,732] ERROR WorkerSourceTask{id=rabbit-vpc-2-kafka-1}
> Failed to flush, timed out while waiting for producer to flush outstanding 27
> messages (org.apache.kafka.connect.runtime.WorkerSourceTask)
> [2019-01-18 20:28:47,732] ERROR WorkerSourceTask{id=rabbit-vpc-2-kafka-1}
> Failed to commit offsets
> (org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter)
> {code}
> The ip address referenced is for the broker that died. We have Kafka Manager
> running as well, and that picks up the new broker.
> We already set
> {code:java}
> networkaddress.cache.ttl=60{code}
> in
> {code:java}
> jre/lib/security/java.security{code}
> Our java version is "Java(TM) SE Runtime Environment (build 1.8.0_192-b12)"
> This started happening after we upgraded to 2.1. When had Kafka 1.1, brokers
> could failover without a problem.
> One thing that might be considered unusual about our deployment is that we
> reuse the same broker id and EBS volume for the new broker, so that
> partitions do not have to be reassigned.
> In kafka-connect, the logs look like
> {code}
> [2019-01-28 22:11:02,364] WARN [Consumer clientId=consumer-1,
> groupId=connect-cluster] Connection to node 3 (/10.130.153.120:9092) could
> not be established. Broker may not be available.
> (org.apache.kafka.clients.NetworkClient)
> [2019-01-28 22:11:02,365] INFO [Consumer clientId=consumer-1,
> groupId=connect-cluster] Error sending fetch request (sessionId=201133590,
> epoch=INITIAL) to node 3: org.apache.kafka.common.errors.DisconnectException.
> (org.apache.kafka.clients.FetchSessionHandler)
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-7845) NotLeaderForPartitionException error when publishing after a broker has died
Jennifer Thompson created KAFKA-7845:
Summary: NotLeaderForPartitionException error when publishing
after a broker has died
Key: KAFKA-7845
URL: https://issues.apache.org/jira/browse/KAFKA-7845
Project: Kafka
Issue Type: Bug
Components: clients
Affects Versions: 2.1.0
Reporter: Jennifer Thompson
When one of our Kafka brokers dies and a new one replaces it (via an aws ASG),
the clients that publish to Kafka still try to publish to the old brokers.
We see errors like
{code:java}
2019-01-18 20:16:16 WARN NetworkClient:721 - [Producer clientId=producer-1]
Connection to node 2 (/10.130.98.111:9092) could not be established. Broker may
not be available.
2019-01-18 20:19:09 WARN Sender:596 - [Producer clientId=producer-1] Got error
produce response with correlation id 3414 on topic-partition aa.pga-2, retrying
(4 attempts left). Error: NOT_LEADER_FOR_PARTITION
2019-01-18 20:19:09 WARN Sender:641 - [Producer clientId=producer-1] Received
invalid metadata error in produce request on partition aa.pga-2 due to
org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is
not the leader for that topic-partition.. Going to request metadata update now
2019-01-18 20:21:19 WARN NetworkClient:721 - [Producer clientId=producer-1]
Connection to node 2 (/10.130.98.111:9092) could not be established. Broker may
not be available.
2019-01-18 20:21:50 ERROR ProducerBatch:233 - Error executing user-provided
callback on message for topic-partition 'aa.test-liz-0'{code}
and
{code:java}
[2019-01-18 20:28:47,732] ERROR WorkerSourceTask{id=rabbit-vpc-2-kafka-1}
Failed to flush, timed out while waiting for producer to flush outstanding 27
messages (org.apache.kafka.connect.runtime.WorkerSourceTask)
[2019-01-18 20:28:47,732] ERROR WorkerSourceTask{id=rabbit-vpc-2-kafka-1}
Failed to commit offsets
(org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter)
{code}
The ip address referenced is for the broker that died. We have Kafka Manager
running as well, and that picks up the new broker.
This started happening after we upgraded to 2.1. When had Kafka 1.1, brokers
could failover without a problem.
One thing that might be considered unusual about our deployment is that we
reuse the same broker id and EBS volume for the new broker, so that partitions
do not have to be reassigned.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
