<commit_before>// Copyright (c) Microsoft Corporation. All rights reserved. See License.txt in the project root for license information.

package com.contoso.app.trident;

import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.microsoft.eventhubs.spout.IStateStore;
import com.microsoft.eventhubs.spout.ZookeeperStateStore;

import backtype.storm.topology.FailedException;
import backtype.storm.tuple.Values;
import storm.trident.operation.BaseAggregator;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.TridentOperationContext;
import storm.trident.topology.TransactionAttempt;
import storm.trident.tuple.TridentTuple;

public class ByteAggregator extends BaseAggregator<BlockState> {
	private static final long serialVersionUID = 1L;
	private static final Logger logger = (Logger) LoggerFactory.getLogger(ByteAggregator.class);
	private static String PARTITION_TXID_KEY_FORMATTER = "partition_%05d_transactionid";
	private static String PARTITION_FIRSTBLOCK_KEY_FORMATTER = "partition_%05d_firstblock";
	private static String PARTITION_LASTBLOCK_KEY_FORMATTER = "partition_%05d_lastblock";

	private long txid;
	private int partitionIndex;
	private long msgCount;
	public static IStateStore stateStore = null;
	public static String partitionTxidKeyStr = null;
	public static String partitionFirstblockKeyStr = null;
	public static String partitionLastblockKeyStr = null;
	//public static String partitionBlocklistKeyStr = null;
	
	static {
		String _PARTITION_TXID_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_TXID_KEY_FORMATTER");
		if (_PARTITION_TXID_KEY_FORMATTER != null) {
			PARTITION_TXID_KEY_FORMATTER = _PARTITION_TXID_KEY_FORMATTER; 
		}
//		String _PARTITION_BLOCKLIST_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_BLOCKLIST_KEY_FORMATTER");
//		if (_PARTITION_BLOCKLIST_KEY_FORMATTER != null) {
//			PARTITION_BLOCKLIST_KEY_FORMATTER = _PARTITION_BLOCKLIST_KEY_FORMATTER; 
//		}
		String _PARTITION_FIRSTBLOCK_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_FIRSTBLOCK_KEY_FORMATTER");
		if (_PARTITION_FIRSTBLOCK_KEY_FORMATTER != null) {
			PARTITION_FIRSTBLOCK_KEY_FORMATTER = _PARTITION_FIRSTBLOCK_KEY_FORMATTER; 
		}
		String _PARTITION_LASTBLOCK_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_LASTBLOCK_KEY_FORMATTER");
		if (_PARTITION_LASTBLOCK_KEY_FORMATTER != null) {
			PARTITION_LASTBLOCK_KEY_FORMATTER = _PARTITION_LASTBLOCK_KEY_FORMATTER; 
		}
	}

	public ByteAggregator() {
		if (LogSetting.LOG_BATCH) {
			logger.info("Constructor");
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public void prepare(@SuppressWarnings("rawtypes") Map conf, TridentOperationContext context) {
		if (LogSetting.LOG_BATCH) {
			logger.info("prepare Begin");
		}
		this.partitionIndex = context.getPartitionIndex();
		ByteAggregator.partitionTxidKeyStr = String.format(PARTITION_TXID_KEY_FORMATTER, partitionIndex);
		ByteAggregator.partitionFirstblockKeyStr = String.format(PARTITION_FIRSTBLOCK_KEY_FORMATTER, partitionIndex);
		ByteAggregator.partitionLastblockKeyStr = String.format(PARTITION_LASTBLOCK_KEY_FORMATTER, partitionIndex);<commit_msg>clean up code<commit_after>// Copyright (c) Microsoft Corporation. All rights reserved. See License.txt in the project root for license information.

package com.contoso.app.trident;

import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.microsoft.eventhubs.spout.IStateStore;
import com.microsoft.eventhubs.spout.ZookeeperStateStore;

import backtype.storm.topology.FailedException;
import backtype.storm.tuple.Values;
import storm.trident.operation.BaseAggregator;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.TridentOperationContext;
import storm.trident.topology.TransactionAttempt;
import storm.trident.tuple.TridentTuple;

public class ByteAggregator extends BaseAggregator<BlockState> {
	private static final long serialVersionUID = 1L;
	private static final Logger logger = (Logger) LoggerFactory.getLogger(ByteAggregator.class);
	private static String PARTITION_TXID_KEY_FORMATTER = "partition_%05d_transactionid";
	private static String PARTITION_FIRSTBLOCK_KEY_FORMATTER = "partition_%05d_firstblock";
	private static String PARTITION_LASTBLOCK_KEY_FORMATTER = "partition_%05d_lastblock";

	private long txid;
	private int partitionIndex;
	private long msgCount;
	public static IStateStore stateStore = null;
	public static String partitionTxidKeyStr = null;
	public static String partitionFirstblockKeyStr = null;
	public static String partitionLastblockKeyStr = null;
	
	static {
		String _PARTITION_TXID_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_TXID_KEY_FORMATTER");
		if (_PARTITION_TXID_KEY_FORMATTER != null) {
			PARTITION_TXID_KEY_FORMATTER = _PARTITION_TXID_KEY_FORMATTER; 
		}
		String _PARTITION_FIRSTBLOCK_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_FIRSTBLOCK_KEY_FORMATTER");
		if (_PARTITION_FIRSTBLOCK_KEY_FORMATTER != null) {
			PARTITION_FIRSTBLOCK_KEY_FORMATTER = _PARTITION_FIRSTBLOCK_KEY_FORMATTER; 
		}
		String _PARTITION_LASTBLOCK_KEY_FORMATTER = ConfigProperties.getProperty("PARTITION_LASTBLOCK_KEY_FORMATTER");
		if (_PARTITION_LASTBLOCK_KEY_FORMATTER != null) {
			PARTITION_LASTBLOCK_KEY_FORMATTER = _PARTITION_LASTBLOCK_KEY_FORMATTER; 
		}
	}

	public ByteAggregator() {
		if (LogSetting.LOG_BATCH) {
			logger.info("Constructor");
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public void prepare(@SuppressWarnings("rawtypes") Map conf, TridentOperationContext context) {
		if (LogSetting.LOG_BATCH) {
			logger.info("prepare Begin");
		}
		this.partitionIndex = context.getPartitionIndex();
		ByteAggregator.partitionTxidKeyStr = String.format(PARTITION_TXID_KEY_FORMATTER, partitionIndex);
		ByteAggregator.partitionFirstblockKeyStr = String.format(PARTITION_FIRSTBLOCK_KEY_FORMATTER, partitionIndex);
		ByteAggregator.partitionLastblockKeyStr = String.format(PARTITION_LASTBLOCK_KEY_FORMATTER, partitionIndex);
{'fraction_non_alphanumeric': 0.07650542941757157, 'fraction_numerical': 0.0029615004935834156, 'mean_word_length': 7.3273972602739725, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6747176', 'n_tokens_mistral': 1910, 'n_tokens_neox': 1841, 'n_words': 336}