Allow Publishers to produce messages by thread ID

zendesk · Jun 29, 2020 · 4d9034a · 4d9034a
1 parent 1063dd0
commit 4d9034a
Show file tree

Hide file tree

Showing 8 changed files with 17 additions and 6 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,11 @@
 # Maxwell changelog
 
+### [v1.26.5](https://github.com/zendesk/maxwell/releases/tag/v1.26.5): "Producer partitioning by thread id"
+
+
+ - Allows Kinesis Producer to partition Kinesis Records by thread id
+
+
 ### [v1.26.4](https://github.com/zendesk/maxwell/releases/tag/v1.26.4): "Now function precision support"
 
 

diff --git a/Dockerfile b/Dockerfile
@@ -1,5 +1,5 @@
 FROM maven:3.6-jdk-11
-ENV MAXWELL_VERSION=1.26.4 KAFKA_VERSION=1.0.0
+ENV MAXWELL_VERSION=1.26.5 KAFKA_VERSION=1.0.0
 
 RUN apt-get update \
     && apt-get -y upgrade \

diff --git a/README.md b/README.md
@@ -9,7 +9,7 @@ search indexing and inter-service communication.  Maxwell gives you some of the
 benefits of event sourcing without having to re-architect your entire platform.
 
 <b>Download:</b><br>
-[https://github.com/zendesk/maxwell/releases/download/v1.26.4/maxwell-1.26.4.tar.gz](https://github.com/zendesk/maxwell/releases/download/v1.26.4/maxwell-1.26.4.tar.gz)
+[https://github.com/zendesk/maxwell/releases/download/v1.26.5/maxwell-1.26.5.tar.gz](https://github.com/zendesk/maxwell/releases/download/v1.26.5/maxwell-1.26.5.tar.gz)
 <br/>
 <b>Source:</b><br>
 [https://github.com/zendesk/maxwell](https://github.com/zendesk/maxwell)

diff --git a/config.properties.example b/config.properties.example
@@ -174,7 +174,7 @@ kafka.acks=1
 #           *** partitioning ***
 
 # What part of the data do we partition by?
-#producer_partition_by=database # [database, table, primary_key, transaction_id, column]
+#producer_partition_by=database # [database, table, primary_key, transaction_id, thread_id, column]
 
 # specify what fields to partition by when using producer_partition_by=column
 # column separated list.

diff --git a/docs/docs/quickstart.md b/docs/docs/quickstart.md
@@ -1,13 +1,13 @@
 ### Download
 ***
-- Download binary distro: [https://github.com/zendesk/maxwell/releases/download/v1.26.4/maxwell-1.26.4.tar.gz](https://github.com/zendesk/maxwell/releases/download/v1.26.4/maxwell-1.26.4.tar.gz)
+- Download binary distro: [https://github.com/zendesk/maxwell/releases/download/v1.26.5/maxwell-1.26.5.tar.gz](https://github.com/zendesk/maxwell/releases/download/v1.26.5/maxwell-1.26.5.tar.gz)
 - Sources and bug tracking is available on github: [https://github.com/zendesk/maxwell](https://github.com/zendesk/maxwell)
 - Obligatory copy/paste to terminal:
 
 ```
 curl -sLo - https://github.com/zendesk/maxwell/releases/download/v1.26.4/maxwell-1.26.4.tar.gz \
        | tar zxvf -
-cd maxwell-1.26.4
+cd maxwell-1.26.5
 ```
 
 or get the docker image:

diff --git a/pom.xml b/pom.xml
@@ -4,7 +4,7 @@
 
   <groupId>com.zendesk</groupId>
   <artifactId>maxwell</artifactId>
-  <version>1.26.4</version>
+  <version>1.26.5</version>
   <packaging>jar</packaging>
 
   <name>maxwell</name>

diff --git a/src/main/java/com/zendesk/maxwell/producer/partitioners/AbstractMaxwellPartitioner.java b/src/main/java/com/zendesk/maxwell/producer/partitioners/AbstractMaxwellPartitioner.java
@@ -26,6 +26,8 @@ private PartitionBy partitionByForString(String key) {
 				return PartitionBy.PRIMARY_KEY;
 			case "transaction_id":
 				return PartitionBy.TRANSACTION_ID;
+			case "thread_id":
+				return PartitionBy.THREAD_ID;
 			case "column":
 				return PartitionBy.COLUMN;
 			case "random":
@@ -65,6 +67,8 @@ public String getHashString(RowMap r, PartitionBy by) {
 				return r.getRowIdentity().toConcatString();
 			case TRANSACTION_ID:
 				return String.valueOf(r.getXid());
+			case THREAD_ID:
+				return String.valueOf(r.getThreadId());
 			case COLUMN:
 				String s = r.buildPartitionKey(partitionColumns);
 				if ( s.length() > 0 )

diff --git a/src/main/java/com/zendesk/maxwell/producer/partitioners/PartitionBy.java b/src/main/java/com/zendesk/maxwell/producer/partitioners/PartitionBy.java
@@ -5,6 +5,7 @@ public enum PartitionBy {
 	TABLE,
 	PRIMARY_KEY,
 	TRANSACTION_ID,
+	THREAD_ID,
 	COLUMN,
 	RANDOM
 }