Go to MongoDB Using Kafka Connectors - Ultimate Agent Guide

1
go mod init example/main

1
go get -u gopkg.in/confluentinc/confluent-kafka-go.v1/kafka

1
package main
2
3
import (
4
   "fmt"
5
   "encoding/json"
6
   "time"
7
   "os"
8
   "path/filepath"
9
)
10
11
type Message struct {
12
  Name string
13
  Size float64
14
  Time int64
15
}
16
17
func samplePath (startPath string)  error {
18
   err := filepath.Walk(startPath,
19
    func(path string, info os.FileInfo, err error) error {
20
 
21
       var bytes int64
22
       bytes = info.Size()
23
24
25
       var kilobytes int64
26
       kilobytes = (bytes / 1024)
27
28
       var megabytes float64
29
       megabytes = (float64)(kilobytes / 1024)   // cast to type float64
30
31
       var gigabytes float64
32
       gigabytes = (megabytes / 1024)
33
34
      now := time.Now().Unix()*1000
35
36
37
38
39
      m := Message{info.Name(), gigabytes, now}
40
      value, err := json.Marshal(m)
41
42
      
43
      if err != nil {
44
         panic(fmt.Sprintf("Failed to parse JSON: %s", err))
45
      }
46
47
48
      fmt.Printf("value: %v\n", string(value))
49
      return nil;
50
  })
51
  if err != nil {
52
     return err
53
  }
54
  return nil;
55
}
56
57
func main() {
58
   for {
59
       err := samplePath("./files");
60
       if err != nil {
61
            panic(fmt.Sprintf("Failed to run sample : %s", err))
62
       }
63
       time.Sleep(time.Minute)
64
   }
65
66
}

1
go run main.go

1
value: {"Name":"files","Size":0,"Time":1643881924000}
2
value: {"Name":"test.txt","Size":0.185546875,"Time":1643881924000}

1
import (
2
   "fmt"
3
   "encoding/json"
4
   "time"
5
   "os"
6
   "path/filepath"
7
   "github.com/confluentinc/confluent-kafka-go/kafka"
8
)

1
const (
2
   bootstrapServers = “<CONFLUENT-SERVER>:<CONFLUENT-PORT>"
3
   ccloudAPIKey     = "<ACCESS-KEY>"
4
   ccloudAPISecret  = "<SECRET-KEY>"
5
)

1
               topic := "files"
2
              // Produce a new record to the topic...
3
               producer, err := kafka.NewProducer(&kafka.ConfigMap{
4
                   "bootstrap.servers":       bootstrapServers,
5
                   "sasl.mechanisms":         "PLAIN",
6
                   "security.protocol":       "SASL_SSL",
7
                   "sasl.username":           ccloudAPIKey,
8
                   "sasl.password":           ccloudAPISecret})
9
10
               if err != nil {
11
                   panic(fmt.Sprintf("Failed to create producer: %s", err))
12
               }
13
14
               producer.Produce(&kafka.Message{
15
                   TopicPartition: kafka.TopicPartition{Topic: &topic,
16
                       Partition: kafka.PartitionAny},
17
                   Value: []byte(value)}, nil)
18
19
               // Wait for delivery report
20
               e := <-producer.Events()
21
22
               message := e.(*kafka.Message)
23
               if message.TopicPartition.Error != nil {
24
                   fmt.Printf("failed to deliver message: %v\n",
25
                       message.TopicPartition)
26
               } else {
27
                   fmt.Printf("delivered to topic %s [%d] at offset %v\n",
28
                       *message.TopicPartition.Topic,
29
                       message.TopicPartition.Partition,
30
                       message.TopicPartition.Offset)
31
               }
32
33
               producer.Close()

1
package main
2
3
4
import (
5
   "fmt"
6
   "encoding/json"
7
   "time"
8
   "os"
9
   "path/filepath"
10
   "github.com/confluentinc/confluent-kafka-go/kafka")
11
12
type Message struct {
13
  Name string
14
  Size float64
15
  Time int64
16
}
17
18
19
const (
20
   bootstrapServers = "<SERVER-DNS>:<SERVER-PORT>"
21
   ccloudAPIKey     = "<ACCESS-KEY>"
22
   ccloudAPISecret  = "<SECRET-KEY>"
23
)
24
25
func samplePath (startPath string)  error {
26
  
27
   err := filepath.Walk(startPath,
28
    func(path string, info os.FileInfo, err error) error {
29
    if err != nil {
30
            return err
31
    }
32
    fmt.Println(path, info.Size())
33
34
       var bytes int64
35
       bytes = info.Size()
36
37
38
       var kilobytes int64
39
       kilobytes = (bytes / 1024)
40
41
       var megabytes float64
42
       megabytes = (float64)(kilobytes / 1024)   // cast to type float64
43
44
       var gigabytes float64
45
       gigabytes = (megabytes / 1024)
46
47
      now := time.Now().Unix()*1000
48
49
50
      
51
52
       m := Message{info.Name(), gigabytes, now}
53
       value, err := json.Marshal(m)
54
55
      
56
       if err != nil {
57
         panic(fmt.Sprintf("Failed to parse JSON: %s", err))
58
       }
59
60
61
      fmt.Printf("value: %v\n", string(value))
62
63
               topic := "files"
64
               // Produce a new record to the topic...
65
                   producer, err := kafka.NewProducer(&kafka.ConfigMap{
66
                       "bootstrap.servers":       bootstrapServers,
67
                       "sasl.mechanisms":         "PLAIN",
68
                       "security.protocol":       "SASL_SSL",
69
                       "sasl.username":           ccloudAPIKey,
70
                       "sasl.password":           ccloudAPISecret})
71
  
72
                   if err != nil {
73
                       panic(fmt.Sprintf("Failed to create producer: %s", err))
74
                   }
75
  
76
                   producer.Produce(&kafka.Message{
77
                       TopicPartition: kafka.TopicPartition{Topic: &topic,
78
                           Partition: kafka.PartitionAny},
79
                       Value: []byte(value)}, nil)
80
  
81
                   // Wait for delivery report
82
                   e := <-producer.Events()
83
  
84
                   message := e.(*kafka.Message)
85
                   if message.TopicPartition.Error != nil {
86
                       fmt.Printf("failed to deliver message: %v\n",
87
                           message.TopicPartition)
88
                   } else {
89
                       fmt.Printf("delivered to topic %s [%d] at offset %v\n",
90
                           *message.TopicPartition.Topic,
91
                           message.TopicPartition.Partition,
92
                           message.TopicPartition.Offset)
93
                   }
94
  
95
                   producer.Close()
96
97
               return nil;
98
})
99
if err != nil {
100
       return err
101
}
102
   return nil;
103
}
104
105
func main() {
106
   for {
107
       err := samplePath("./files");
108
       if err != nil {
109
            panic(fmt.Sprintf("Failed to run sample : %s", err))
110
       }
111
     time.Sleep(time.Minute)
112
  
113
   }
114
115
}