Skip to content
This repository was archived by the owner on Feb 23, 2023. It is now read-only.

Commit 72566cb

Browse files
garyrussellgaryrussell
committed
GH-687: Apache Kafka with Avro Serialization
Resolves #687 Add hints for the Confluent Apache Avro (De)Serializers, as well as the Confluent schema registry client.
1 parent 7f1ae17 commit 72566cb

File tree

12 files changed

+618
-0
lines changed

12 files changed

+618
-0
lines changed

‎samples/kafka-avro/.gitignore

+33
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
HELP.md
2+
target/
3+
!.mvn/wrapper/maven-wrapper.jar
4+
!**/src/main/**/target/
5+
!**/src/test/**/target/
6+
7+
### STS ###
8+
.apt_generated
9+
.classpath
10+
.factorypath
11+
.project
12+
.settings
13+
.springBeans
14+
.sts4-cache
15+
16+
### IntelliJ IDEA ###
17+
.idea
18+
*.iws
19+
*.iml
20+
*.ipr
21+
22+
### NetBeans ###
23+
/nbproject/private/
24+
/nbbuild/
25+
/dist/
26+
/nbdist/
27+
/.nb-gradle/
28+
build/
29+
!**/src/main/**/build/
30+
!**/src/test/**/build/
31+
32+
### VS Code ###
33+
.vscode/

‎samples/kafka-avro/README.md

+3
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
## Apache Kafka with Confluent Avro Serialization "Hello, world!" App
2+
3+
Spring Boot project with Apache Kafka and Confluent Avro Serialization, and Avro schema-registry client.

‎samples/kafka-avro/build.sh

+9
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
#!/usr/bin/env bash
2+
3+
RC=0
4+
5+
docker-compose up -d schema-registry
6+
${PWD%/*samples/*}/scripts/compileWithMaven.sh && ${PWD%/*samples/*}/scripts/test.sh || RC=$?
7+
docker-compose down
8+
9+
exit $RC

‎samples/kafka-avro/docker-compose.yml

+35
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
version: '3'
2+
services:
3+
kafka:
4+
image: wurstmeister/kafka
5+
container_name: kafka
6+
hostname: kafka
7+
ports:
8+
- "9092:9092"
9+
environment:
10+
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
11+
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
12+
KAFKA_LISTENERS: PLAINTEXT://:29092,PLAINTEXT_HOST://:9092
13+
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
14+
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
15+
depends_on:
16+
- zookeeper
17+
18+
zookeeper:
19+
image: wurstmeister/zookeeper
20+
ports:
21+
- "2181:2181"
22+
environment:
23+
- KAFKA_ADVERTISED_HOST_NAME=zookeeper
24+
25+
schema-registry:
26+
image: confluentinc/cp-schema-registry
27+
hostname: schema-registry
28+
container_name: schema-registry
29+
depends_on:
30+
- kafka
31+
ports:
32+
- "8081:8081"
33+
environment:
34+
SCHEMA_REGISTRY_HOST_NAME: schema-registry
35+
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'kafka:29092'

‎samples/kafka-avro/generated-sources/com/example/kafka/Thing.java

+307
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,307 @@
1+
/**
2+
* Autogenerated by Avro
3+
*
4+
* DO NOT EDIT DIRECTLY
5+
*/
6+
package com.example.kafka;
7+
8+
import org.apache.avro.specific.SpecificData;
9+
import org.apache.avro.message.BinaryMessageEncoder;
10+
import org.apache.avro.message.BinaryMessageDecoder;
11+
import org.apache.avro.message.SchemaStore;
12+
13+
@SuppressWarnings("all")
14+
@org.apache.avro.specific.AvroGenerated
15+
public class Thing extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
16+
private static final long serialVersionUID = -719126663737953940L;
17+
public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Thing\",\"namespace\":\"com.example.kafka\",\"fields\":[{\"name\":\"stringField\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}},{\"name\":\"intField\",\"type\":\"int\"}]}");
18+
public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
19+
20+
private static SpecificData MODEL$ = new SpecificData();
21+
22+
private static final BinaryMessageEncoder<Thing> ENCODER =
23+
new BinaryMessageEncoder<Thing>(MODEL$, SCHEMA$);
24+
25+
private static final BinaryMessageDecoder<Thing> DECODER =
26+
new BinaryMessageDecoder<Thing>(MODEL$, SCHEMA$);
27+
28+
/**
29+
* Return the BinaryMessageDecoder instance used by this class.
30+
*/
31+
public static BinaryMessageDecoder<Thing> getDecoder() {
32+
return DECODER;
33+
}
34+
35+
/**
36+
* Create a new BinaryMessageDecoder instance for this class that uses the specified {@link SchemaStore}.
37+
* @param resolver a {@link SchemaStore} used to find schemas by fingerprint
38+
*/
39+
public static BinaryMessageDecoder<Thing> createDecoder(SchemaStore resolver) {
40+
return new BinaryMessageDecoder<Thing>(MODEL$, SCHEMA$, resolver);
41+
}
42+
43+
/** Serializes this Thing to a ByteBuffer. */
44+
public java.nio.ByteBuffer toByteBuffer() throws java.io.IOException {
45+
return ENCODER.encode(this);
46+
}
47+
48+
/** Deserializes a Thing from a ByteBuffer. */
49+
public static Thing fromByteBuffer(
50+
java.nio.ByteBuffer b) throws java.io.IOException {
51+
return DECODER.decode(b);
52+
}
53+
54+
@Deprecated public java.lang.String stringField;
55+
@Deprecated public int intField;
56+
57+
/**
58+
* Default constructor. Note that this does not initialize fields
59+
* to their default values from the schema. If that is desired then
60+
* one should use <code>newBuilder()</code>.
61+
*/
62+
public Thing() {}
63+
64+
/**
65+
* All-args constructor.
66+
* @param stringField The new value for stringField
67+
* @param intField The new value for intField
68+
*/
69+
public Thing(java.lang.String stringField, java.lang.Integer intField) {
70+
this.stringField = stringField;
71+
this.intField = intField;
72+
}
73+
74+
public org.apache.avro.Schema getSchema() { return SCHEMA$; }
75+
// Used by DatumWriter. Applications should not call.
76+
public java.lang.Object get(int field$) {
77+
switch (field$) {
78+
case 0: return stringField;
79+
case 1: return intField;
80+
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
81+
}
82+
}
83+
84+
// Used by DatumReader. Applications should not call.
85+
@SuppressWarnings(value="unchecked")
86+
public void put(int field$, java.lang.Object value$) {
87+
switch (field$) {
88+
case 0: stringField = (java.lang.String)value$; break;
89+
case 1: intField = (java.lang.Integer)value$; break;
90+
default: throw new org.apache.avro.AvroRuntimeException("Bad index");
91+
}
92+
}
93+
94+
/**
95+
* Gets the value of the 'stringField' field.
96+
* @return The value of the 'stringField' field.
97+
*/
98+
public java.lang.String getStringField() {
99+
return stringField;
100+
}
101+
102+
/**
103+
* Sets the value of the 'stringField' field.
104+
* @param value the value to set.
105+
*/
106+
public void setStringField(java.lang.String value) {
107+
this.stringField = value;
108+
}
109+
110+
/**
111+
* Gets the value of the 'intField' field.
112+
* @return The value of the 'intField' field.
113+
*/
114+
public java.lang.Integer getIntField() {
115+
return intField;
116+
}
117+
118+
/**
119+
* Sets the value of the 'intField' field.
120+
* @param value the value to set.
121+
*/
122+
public void setIntField(java.lang.Integer value) {
123+
this.intField = value;
124+
}
125+
126+
/**
127+
* Creates a new Thing RecordBuilder.
128+
* @return A new Thing RecordBuilder
129+
*/
130+
public static com.example.kafka.Thing.Builder newBuilder() {
131+
return new com.example.kafka.Thing.Builder();
132+
}
133+
134+
/**
135+
* Creates a new Thing RecordBuilder by copying an existing Builder.
136+
* @param other The existing builder to copy.
137+
* @return A new Thing RecordBuilder
138+
*/
139+
public static com.example.kafka.Thing.Builder newBuilder(com.example.kafka.Thing.Builder other) {
140+
return new com.example.kafka.Thing.Builder(other);
141+
}
142+
143+
/**
144+
* Creates a new Thing RecordBuilder by copying an existing Thing instance.
145+
* @param other The existing instance to copy.
146+
* @return A new Thing RecordBuilder
147+
*/
148+
public static com.example.kafka.Thing.Builder newBuilder(com.example.kafka.Thing other) {
149+
return new com.example.kafka.Thing.Builder(other);
150+
}
151+
152+
/**
153+
* RecordBuilder for Thing instances.
154+
*/
155+
public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<Thing>
156+
implements org.apache.avro.data.RecordBuilder<Thing> {
157+
158+
private java.lang.String stringField;
159+
private int intField;
160+
161+
/** Creates a new Builder */
162+
private Builder() {
163+
super(SCHEMA$);
164+
}
165+
166+
/**
167+
* Creates a Builder by copying an existing Builder.
168+
* @param other The existing Builder to copy.
169+
*/
170+
private Builder(com.example.kafka.Thing.Builder other) {
171+
super(other);
172+
if (isValidValue(fields()[0], other.stringField)) {
173+
this.stringField = data().deepCopy(fields()[0].schema(), other.stringField);
174+
fieldSetFlags()[0] = true;
175+
}
176+
if (isValidValue(fields()[1], other.intField)) {
177+
this.intField = data().deepCopy(fields()[1].schema(), other.intField);
178+
fieldSetFlags()[1] = true;
179+
}
180+
}
181+
182+
/**
183+
* Creates a Builder by copying an existing Thing instance
184+
* @param other The existing instance to copy.
185+
*/
186+
private Builder(com.example.kafka.Thing other) {
187+
super(SCHEMA$);
188+
if (isValidValue(fields()[0], other.stringField)) {
189+
this.stringField = data().deepCopy(fields()[0].schema(), other.stringField);
190+
fieldSetFlags()[0] = true;
191+
}
192+
if (isValidValue(fields()[1], other.intField)) {
193+
this.intField = data().deepCopy(fields()[1].schema(), other.intField);
194+
fieldSetFlags()[1] = true;
195+
}
196+
}
197+
198+
/**
199+
* Gets the value of the 'stringField' field.
200+
* @return The value.
201+
*/
202+
public java.lang.String getStringField() {
203+
return stringField;
204+
}
205+
206+
/**
207+
* Sets the value of the 'stringField' field.
208+
* @param value The value of 'stringField'.
209+
* @return This builder.
210+
*/
211+
public com.example.kafka.Thing.Builder setStringField(java.lang.String value) {
212+
validate(fields()[0], value);
213+
this.stringField = value;
214+
fieldSetFlags()[0] = true;
215+
return this;
216+
}
217+
218+
/**
219+
* Checks whether the 'stringField' field has been set.
220+
* @return True if the 'stringField' field has been set, false otherwise.
221+
*/
222+
public boolean hasStringField() {
223+
return fieldSetFlags()[0];
224+
}
225+
226+
227+
/**
228+
* Clears the value of the 'stringField' field.
229+
* @return This builder.
230+
*/
231+
public com.example.kafka.Thing.Builder clearStringField() {
232+
stringField = null;
233+
fieldSetFlags()[0] = false;
234+
return this;
235+
}
236+
237+
/**
238+
* Gets the value of the 'intField' field.
239+
* @return The value.
240+
*/
241+
public java.lang.Integer getIntField() {
242+
return intField;
243+
}
244+
245+
/**
246+
* Sets the value of the 'intField' field.
247+
* @param value The value of 'intField'.
248+
* @return This builder.
249+
*/
250+
public com.example.kafka.Thing.Builder setIntField(int value) {
251+
validate(fields()[1], value);
252+
this.intField = value;
253+
fieldSetFlags()[1] = true;
254+
return this;
255+
}
256+
257+
/**
258+
* Checks whether the 'intField' field has been set.
259+
* @return True if the 'intField' field has been set, false otherwise.
260+
*/
261+
public boolean hasIntField() {
262+
return fieldSetFlags()[1];
263+
}
264+
265+
266+
/**
267+
* Clears the value of the 'intField' field.
268+
* @return This builder.
269+
*/
270+
public com.example.kafka.Thing.Builder clearIntField() {
271+
fieldSetFlags()[1] = false;
272+
return this;
273+
}
274+
275+
@Override
276+
@SuppressWarnings("unchecked")
277+
public Thing build() {
278+
try {
279+
Thing record = new Thing();
280+
record.stringField = fieldSetFlags()[0] ? this.stringField : (java.lang.String) defaultValue(fields()[0]);
281+
record.intField = fieldSetFlags()[1] ? this.intField : (java.lang.Integer) defaultValue(fields()[1]);
282+
return record;
283+
} catch (java.lang.Exception e) {
284+
throw new org.apache.avro.AvroRuntimeException(e);
285+
}
286+
}
287+
}
288+
289+
@SuppressWarnings("unchecked")
290+
private static final org.apache.avro.io.DatumWriter<Thing>
291+
WRITER$ = (org.apache.avro.io.DatumWriter<Thing>)MODEL$.createDatumWriter(SCHEMA$);
292+
293+
@Override public void writeExternal(java.io.ObjectOutput out)
294+
throws java.io.IOException {
295+
WRITER$.write(this, SpecificData.getEncoder(out));
296+
}
297+
298+
@SuppressWarnings("unchecked")
299+
private static final org.apache.avro.io.DatumReader<Thing>
300+
READER$ = (org.apache.avro.io.DatumReader<Thing>)MODEL$.createDatumReader(SCHEMA$);
301+
302+
@Override public void readExternal(java.io.ObjectInput in)
303+
throws java.io.IOException {
304+
READER$.read(this, SpecificData.getDecoder(in));
305+
}
306+
307+
}

0 commit comments

Comments
 (0)