Use thread-local instances of MessageDigest in community_id processor (#68032)

x-pack/plugin/ingest/src/main/java/org/elasticsearch/xpack/ingest/CommunityIdProcessor.java

@@ -40,7 +40,7 @@ public final class CommunityIdProcessor extends AbstractProcessor {
     private final String icmpTypeField;
     private final String icmpCodeField;
     private final String targetField;
-    private final MessageDigest messageDigest;
+    private final ThreadLocal<MessageDigest> messageDigest;
     private final byte[] seed;
     private final boolean ignoreMissing;
 
@@ -56,7 +56,7 @@ public final class CommunityIdProcessor extends AbstractProcessor {
         String icmpTypeField,
         String icmpCodeField,
         String targetField,
-        MessageDigest messageDigest,
+        ThreadLocal<MessageDigest> messageDigest,
         byte[] seed,
         boolean ignoreMissing
     ) {
@@ -112,7 +112,7 @@ public final class CommunityIdProcessor extends AbstractProcessor {
     }
 
     public MessageDigest getMessageDigest() {
-        return messageDigest;
+        return messageDigest.get();
     }
 
     public byte[] getSeed() {
@@ -134,7 +134,9 @@ public final class CommunityIdProcessor extends AbstractProcessor {
             }
         }
 
-        ingestDocument.setFieldValue(targetField, flow.toCommunityId(messageDigest, seed));
+        MessageDigest md = messageDigest.get();
+        md.reset();
+        ingestDocument.setFieldValue(targetField, flow.toCommunityId(md, seed));
         return ingestDocument;
     }
 
@@ -255,12 +257,13 @@ public final class CommunityIdProcessor extends AbstractProcessor {
             if (seedInt < 0 || seedInt > 65535) {
                 throw newConfigurationException(TYPE, processorTag, "seed", "must be a value between 0 and 65535");
             }
-            MessageDigest messageDigest;
-            try {
-                messageDigest = MessageDigest.getInstance("SHA-1");
-            } catch (NoSuchAlgorithmException e) {
-                throw new IllegalStateException("unable to obtain SHA-1 hasher", e);
-            }
+            ThreadLocal<MessageDigest> messageDigest = ThreadLocal.withInitial(() -> {
+                try {
+                    return MessageDigest.getInstance("SHA-1");
+                } catch (NoSuchAlgorithmException e) {
+                    throw new IllegalStateException("unable to obtain SHA-1 hasher", e);
+                }
+            });
 
             boolean ignoreMissing = readBooleanProperty(TYPE, processorTag, config, "ignore_missing", true);
             return new CommunityIdProcessor(

x-pack/plugin/ingest/src/test/java/org/elasticsearch/xpack/ingest/CommunityIdProcessorTests.java

@@ -11,6 +11,7 @@ import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
 import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -35,11 +36,17 @@ public class CommunityIdProcessorTests extends ESTestCase {
     // https://github.com/elastic/beats/blob/master/libbeat/processors/communityid/communityid_test.go
 
     private Map<String, Object> event;
-    private MessageDigest messageDigest;
+    private ThreadLocal<MessageDigest> messageDigest;
 
     @Before
     public void setup() throws Exception {
-        messageDigest = MessageDigest.getInstance("SHA-1");
+        messageDigest = ThreadLocal.withInitial(() -> {
+            try {
+                return MessageDigest.getInstance("SHA-1");
+            } catch (NoSuchAlgorithmException e) {
+                throw new IllegalStateException("unable to obtain SHA-1 hasher", e);
+            }
+        });
         event = buildEvent();
     }