WriteMessages "Unexpected EOF" , config with sasl-plain #795
Comments
|
I can get all topics from kafka server |
tong3jie
changed the title
tong3jie
changed the title
|
is anybody here? |
|
I found the error is at here Line 174 in e88d48a because read 4 bytes but not enough |
|
not sure if it's the same issue, but #799 (comment) might help you |
|
@mrkagelui same error, used your case |
|
I hit |
|
wait, you said some operation was successful while others aren't? that's odd, not authentication issue then |
|
yes ,I can list all topics and read messages |
|
Hi, thanks for your patience, Are you still experiencing this issue? |
|
Hi, I have the same err with the version v0.4.28 and 0.10.2 for Kafka server version. But if I use Conn, it's work. With version 0.4.0 and the older versions, it works. The error seems to come with this commit: 81bd03c |
|
Unfortunately I'm still not able to reproduce this issue. I haven't been able to successfully run version 0.10.2 with TLS locally so I've only been testing with more recent versions of Kafka. If you're able to, please provide a reproduction that includes a set-up for Kafka. I'm wondering if there's potentially a difference between go versions, what version of go are you running? If you're not able to provide a runnable reproduction could you potentially provide a dump of the communication between kafka-go and your Kafka server? Thanks! |
|
I use the go version go1.17 linux/amd64. |
|
A tool like wireshark can be used to dump the network communication between kafka-go and the server. Setting a logger on the kafka.Writer type could also potentially give more information as to what's going on. |
|
I meet the bug too, and I found the error is at here
|
#490 (comment) |
|
I appreciate the additional reports; but without a reproduction or any information about the Kafka server configuration I'm not sure what else I can do to debug this. I do encourage anyone experiencing the issue to submit a PR with a fix. Thanks! |
OS : win10 |
|
You mentioned your os is windows10 is that for the client? I wonder if that's related? Unfortunately I don't have a way to test from Windows currently. Is anyone else experiencing this error on windows? |
|
Hi there, Thanks! |
|
Hi @rhansen2 , I have the same issue
But 2. |
|
Sadly I don't get the In my setup using the NewWriter approach seams to get no response from the server.
|
|
@rhansen2 ok it seams to end up in en endless loop when decoding the first message from the server. There internally in protocol/decode.go it never leaves the for loop at at line 109 since d.remain never decrements
|
|
My kafka server has version 2.6.0 |
|
HI @ThomasObenaus, Are you able to provide a reproduction of your issue? Thanks! |
|
@rhansen2 sorry I have no full example for reproduction since I don't know how I should provide the Kafka server configuration of my company. But the code snipped I use is: package main
import (
"context"
"crypto"
"crypto/tls"
"crypto/x509"
"fmt"
"io/ioutil"
"log"
"time"
"github.com/pkg/errors"
"github.com/segmentio/kafka-go"
"software.sslmate.com/src/go-pkcs12"
)
func main() {
host := "my-kafka-server:443"
topic := "my-topic"
trustStorePassword := "abcdefg"
keyStorePassword := "xyzuvw"
tlsConfig, err := tlsCfg("ca.p12", trustStorePassword, "user.p12", keyStorePassword)
if err != nil {
log.Fatal(err)
}
dialer := &kafka.Dialer{
Timeout: 10 * time.Second,
DualStack: true,
TLS: tlsConfig,
}
if err := writeUsingConnection(dialer, host, topic); err != nil {
log.Fatal(err)
}
if err := writeUsingWriter(dialer, host, topic); err != nil {
log.Fatal(err)
}
if err := read(dialer, host, topic, time.Second*10); err != nil {
log.Fatal(err)
}
}
func logger(silence bool) kafka.LoggerFunc {
return func(msg string, args ...interface{}) {
if silence {
return
}
fmt.Printf(fmt.Sprintf("[DBG] %s\n", msg), args...)
}
}
func errLogger() kafka.LoggerFunc {
return func(msg string, args ...interface{}) {
fmt.Printf(fmt.Sprintf("[Error] %s\n", msg), args...)
}
}
func tlsCfg(trustStoreFile string, trustStorePassword string, keyStoreFile string, keyStorePassword string) (*tls.Config, error) {
trustStore, err := ioutil.ReadFile(trustStoreFile)
if err != nil {
return nil, errors.Wrap(err, "loading trust store")
}
trustStoreCerts, err := pkcs12.DecodeTrustStore(trustStore, trustStorePassword)
if err != nil {
return nil, errors.Wrap(err, "decoding trust store")
}
certPool, err := x509.SystemCertPool()
if err != nil {
return nil, errors.Wrap(err, "opening system cert pool")
}
for _, cert := range trustStoreCerts {
certPool.AddCert(cert)
}
keyStore, err := ioutil.ReadFile(keyStoreFile)
if err != nil {
return nil, errors.Wrap(err, "loading key store")
}
keyStoreKey, keyStoreCert, err := pkcs12.Decode(keyStore, keyStorePassword)
if err != nil {
return nil, errors.Wrap(err, "decoding key store")
}
clientCert := tls.Certificate{
Certificate: [][]byte{keyStoreCert.Raw},
PrivateKey: keyStoreKey.(crypto.PrivateKey),
Leaf: keyStoreCert,
}
return &tls.Config{
InsecureSkipVerify: false,
MaxVersion: tls.VersionTLS12,
Certificates: []tls.Certificate{clientCert},
RootCAs: certPool,
}, nil
}
func writeUsingConnection(dialer *kafka.Dialer, host string, topic string) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
defer cancel()
conn, err := dialer.DialContext(ctx, "tcp", host)
if err != nil {
return errors.Wrap(err, "opening connection")
}
defer conn.Close()
fmt.Println("[INFO] Connected")
fmt.Println("[INFO] Start write (using connection)")
connLeader, err := dialer.DialLeader(ctx, "tcp", host, topic, 0)
if err != nil {
return errors.Wrap(err, "opening connection to leader")
}
defer connLeader.Close()
connLeader.SetWriteDeadline(time.Now().Add(2 * time.Second))
bWritten, err := connLeader.WriteMessages(
kafka.Message{Key: []byte("Key-Conn"), Value: []byte("Message Written Using Connection")})
if err != nil {
return errors.Wrap(err, "writing message")
}
fmt.Printf("[INFO] Done write (using connection) %d Bytes\n", bWritten)
return nil
}
func writeUsingWriter(dialer *kafka.Dialer, host string, topic string) error {
fmt.Println("[INFO] Start write (using NewWriter)")
w := kafka.NewWriter(kafka.WriterConfig{
Brokers: []string{host},
Topic: topic,
Balancer: &kafka.LeastBytes{},
Dialer: dialer,
ErrorLogger: errLogger(),
Logger: logger(true),
BatchSize: 1,
})
defer func() {
if err := w.Close(); err != nil {
fmt.Printf("[Error] Failed to close writer: %v\n", err)
}
}()
ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
defer cancel()
err := w.WriteMessages(ctx,
kafka.Message{Key: []byte("Key-Writer"), Value: []byte("Message Written Using Writer")},
)
if err != nil {
return errors.Wrap(err, "writing messages")
}
fmt.Println("[INFO] Done write (using NewWriter)")
return nil
}
func read(dialer *kafka.Dialer, host string, topic string, readFor time.Duration) error {
fmt.Println("[INFO] Start reading")
r := kafka.NewReader(kafka.ReaderConfig{
Brokers: []string{host},
GroupID: "consumer-group-1",
Topic: topic,
Dialer: dialer,
ErrorLogger: errLogger(),
Logger: logger(true),
MaxWait: time.Second,
})
defer func() {
if err := r.Close(); err != nil {
fmt.Printf("[Error] Failed to close reader: %v\n", err)
}
}()
ctx, cancel := context.WithTimeout(context.Background(), readFor)
defer cancel()
for i := 0; i < 100; i++ {
subCtx, cancelSubCtx := context.WithTimeout(ctx, time.Second*2)
defer cancelSubCtx()
m, err := r.ReadMessage(subCtx)
if err != nil {
globalDeadline, _ := ctx.Deadline()
if time.Now().After(globalDeadline) {
break
}
subDeadline, _ := subCtx.Deadline()
if time.Now().After(subDeadline) {
fmt.Printf("\r(%0.2fs)...", globalDeadline.Sub(time.Now()).Seconds())
continue
}
fmt.Printf("\n[Error] Reading %v\n", err)
break
}
fmt.Printf("\n[INFO] Message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))
}
fmt.Println("\n[INFO] Done reading")
return nil
} |
|
@rhansen2 The root cause seams to have something to do with the TLS configuration when one (like me) skips the server certificate verification. Does not workwith OutputWorkswith Output |
|
Glad we were able to get that fixed! @tong3jie @Jrmy2402 @lxxwilliam If you can please retry with the latest release and let me know if your issue persists. Thanks! |
|
Hi, I always have the error: unexpected EOF on my side. Line 603 in 2d04c4b Line 351 in 2d04c4b And it's work if I use |
|
@Jrmy2402 To confirm you're still seeing the unexpected EOF with v0.4.30? |
|
Yes i still seeing the unexpected EOF with v0.4.30 |
|
😞 Thanks! Could you possible share the TLS config your kafka brokers are using? Additionally is the code from #795 (comment) still what you're using to test? In that example you don't set |
sad! got unexpected EOF too！ |
|
@lxxwilliam are you able to provide the configuration of your Kafka brokers? |
`kafkaDialer := &kafka.Dialer{ |
|
@lxxwilliam I was hoping for Kafka broker configurations not the Writer config. Sorry for any confusion. |
|
@rhansen2 I'm mostly able to recreate this in my dev env. (not using sasl, but seeing EOF behavior when using Any chance you'd be down for a hangout/screenshare so I can gather whatever data you need? |
|
@curtisr7 Without more information I'm not sure what I'd be looking for beyond the information I've been requesting such as Kafka broker configurations and potentially packet captures between the Writer and the Kafka Brokers. |
|
Gotcha. Here is a snippet from my kafka pod coming up(not sure if this is helpful) |
|
Thanks @curtisr7, unfortunately I don't see anything interesting in the differences between your set-up and the testing set-up I'm using from https://github.com/vdesabou/kafka-docker-playground (environment/sasl-ssl). Are you able to reproduce the issue using that docker environment in the Are you able to determine if the Kafka broker is closing the connections? Additionally, your full code snippet which reproduces the issue may be helpful. Thanks! |
|
It may also be helpful to know if Writes are attempted multiple times via the Writer if they eventually succeed. |
No, I didn't get the time this afternoon to look into that test framework.
In my kafka pod log I'm seeing:
No, that doesn't seem to matter. To test I tried writing using the same writer 10x, with 1sec sleep and same result. Also, as I mentioned before I'm not using sasl.. so my scenario is slightly different. I'm setting |
|
Thanks! The SSL Handshake failure is interesting. One of the differences between the dialer type and the writer is the conn manually performs the handshake but the Writer relies on the tls library performing the handshake on its first write. Looking at a packet capture of two connection methods might help diagnose what's going on. |
Here are two tcpdumps... though not really sure how to interpret those things. Hopefully they make some sense to you. Let me know what else I can collect. Thanks |
|
Thanks @curtisr7. Are you able to provide the code you're using to create your writer vs dialer? |
In responding to this question I think I found the problem. In both cases I was creating a dialer in the same manner. When creating the writer I was using: but I found if I use the follow(note |
|
On closer inspection I need to create the transport something like this |
|
I don't think you should need to set the Context or Dial on the Transport for things to work. |
+1 |
|
@lxxwilliam @tong3jie Thanks! |
|
It's working for me with your version! Thanks a lot :) |
|
Fixed via #869 |
I met another problems when I used v0.4.31, |
|
@lxxwilliam Please open a new issue and provide a runnable reproduction case. Thanks! |
Describe the bug
when writer message to kafka server , it report that "Unexpected EOF"
both of report are about io or net
I can't solve it
Kafka Version
kafka server : 0.10.22
kafka-go : 0.4.23
To Reproduce
Expected behavior
Additional context
because kafka is not priority queue , so I must pause some low priority topics and resume it after little time .
The text was updated successfully, but these errors were encountered: